SQLAlchemy 直接SQL、execute, from_statement サンプル

サンプル

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, registry
import sqlalchemy as db

mapper_reg = registry()
Base = mapper_reg.generate_base()

class Sample(Base):
    __tablename__ = 'samples'
    id  = db.Column(db.UUID, nullable=False, primary_key=True)
    name = db.Column(db.String(32), nullable=False)
    color = db.Column(db.String(32), nullable=False)
    point = db.Column(db.INT, nullable=False)
    create_time = db.Column(db.TIMESTAMP, nullable=False)

#  接続情報、username, password, host, port, dbname は設定済
engine = create_engine(f'postgresql://{username}:{password}@{host}:{port}/{dbname}')

execute で直接SQLクエリ

with Session(autocommit=False, autoflush=True, bind=engine) as session:
    sql = "SELECT * FROM samples WHERE color = '%s'" % 'Red'
    res = session.execute(text(sql))
    slist = res.mappings().all()
    for s in slist:
        print("%s %s %s %s %d" % (s.get('create_time'), s.get('id'), s.get('name'), s.get('color'), s.get('point')))

from_statement で SQL

with Session(autocommit=False, autoflush=True, bind=engine) as session:
    sql = "SELECT * FROM samples WHERE color = '%s'" % 'Red'
    res = session.query(Sample).from_statement(text(sql)).all()
    for s in res:
        print("%s %s %s %s %d" % (s.create_time, s.id, s.name, s.color, s.point))

from_statement と bind パラメータ

with Session(autocommit=False, autoflush=True, bind=engine) as session:
    sql = "SELECT * FROM samples WHERE color = :color AND point = :point"
    res = session.query(Sample).from_statement(text(sql).params(color='Green',point=34)).all()
    for s in res:
        print("%s %s %s %s %d" % (s.create_time, s.id, s.name, s.color, s.point))

bind パラメータ INSERT 実行、RETURNING句 SQL

with Session(autocommit=False, autoflush=True, bind=engine) as session:
    try:
        sql = """
        INSERT INTO samples (name, color, point)
        VALUES(:name, :color, :point)
        RETURNING id
        """
        res = session.execute(text(sql).params(name='Tokyo',
                                               color='Green',
                                               point=124)
                              ).scalar()
        print(res)
    except Exception as e:
        print(e)
        session.rollback()
        raise
    else:
        session.commit()