Snippet/Python/SQLAlchemy

Python et SQL Alchemy

Lecture ligne par ligne (yield méthode) avec l'utilisation d'un SCHEMA dans le nom de la table

Note: Code pas optimisé. Refactoring nécessaire.

Quand on souhaite faire une sélection dans une table, il existe plusieurs méthodes avec SQL Alchemy. La plus simple étant de faire un bête execute. Cependant, si on a beaucoup de données dans la table, vous aurez un joli timeout. Pour cela, on va utiliser le combo "Session" et "Query" avec yield_per.

Pour ajouter de la complexité, notre table se trouve dans un SCHEMA appelé "my_schema". Quand vous faites une requête dedans, vous devez faire "SELECT * FROM my_schema.user": Avec SQLAlchemy, c'est un peu plus complexe, on doit modifier __table_args__ pour ajouter le nom du schema. Sans cela, SQLAlchemy va construire une requête complètement farfelue.

import sqlalchemy from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, String, Integer, ForeignKey

config = {

   'endpoint': '127.0.0.1:5432',
   'user': 'mon_user',
   'pwd' : 'mon_password',
   'name': 'ma_database',

}

class User(declarative_base()):

   __tablename__ = 'user'
   __table_args__ = {'schema' : 'my_schema'}
   uuid = Column(String, primary_key=True, key='uuid')
   name = Column(String, key='name')

url = 'postgresql://{user}:{pwd}@{endpoint}/{name}'.format(**config) database_engine = sqlalchemy.create_engine(url, client_encoding='utf8', echo=False) database_con = database_engine.connect() database_meta = sqlalchemy.MetaData(bind=database_con)

session = sessionmaker() session.configure(bind=database_engine) for user in session().query(User).yield_per(100):

   print(user)