SQLAlchemy query() の filter メモ

代表的な filter の書き方のサンプル・メモ
（try～catch は省略）

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}')

シンプルなfilter()

def query(p:int) -> list[Sample]:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        res = session.query(Sample).filter(Sample.point < p).all()
        return res

LIKE の実行

def query(k:str) -> list[Sample]:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        res = session.query(Sample).filter(Sample.name.like('%{}%'.format(k))).all()
        return res

OR条件

from sqlalchemy import or_

def query(p:int, d:datetime.datetime) -> list[Sample]:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        res = session.query(Sample).filter(or_(Sample.point > p, Sample.create_time < d)).all()
        return res

AND条件

from sqlalchemy import and_

def query(p:int, d:datetime.datetime) -> list[Sample]:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        res = session.query(Sample).filter(and_(Sample.point > p, Sample.create_time < d)).all()
        return res

ORDER BY ASC

from sqlalchemy import asc

def query() -> list[Sample]:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        res = session.query(Sample).order_by(asc(Sample.create_time)).all()
        return res

ORDER BY DESC

from sqlalchemy import desc

def query() -> list[Sample]:
    with Session(autocommit=False, autoflush=True, bind=engine) as session:
        res = session.query(Sample).order_by(desc(Sample.create_time)).all()
        return res