FastAPI & SQLAlchemy: SQLAlchemy from 1.4 to 2.0

DB 연결

from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from typing_extensions import Annotated

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, expire_on_commit=True)

async def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
        
db = Annotated[Session, Depends(get_db)]

 

 

query count - ORM

 

legacy

query = db.query(MODEL)
result = query.all()
result_count = query.count()

 

2.0 이전의 orm에서는 query() 메소드 사용

 

 

2.0 

from sqlalchemy import select, func


query = select(Model)
result = db.execute(query).all()
result_count = db.execute(select(func.count()).select_from(query)).scalar()

count_by_one_column = db.execute(select(func.count(MODEL.id)).scalar()

 

 

2.0부터는 count() 메소드가 없다.

위와 같이 func.count()를 활용해야 한다.

 

count_by_one_coulumn에서 처럼 특정 테이블을 바로 활용할 수 있는데,

이때 주의할 점은 하나의 컬럼을 지정해줘야 한다.

 

그러니 query가 join등으로 여러 테이블을 합쳤거나 해서 query로부터 하나의 column을 지정해주기가 어려운 경우는

result_count에서 활용한 것처럼, select(func.count()).select_from(query)를 활용

이때, count의 컬럼명, select_from에 지정해주는 query의 별칭 alias를 지정해줄 수 있다.

# select(func.count().label('result_count')).select_from(query.alias('subquery'))

result_count = db.execute(
	select(func.count().label('result_count')).select_from(query.alias('subquery'))
).scalar()

 

 

update, delete등이 아닌 select sql을 쓰기 위해

execute(...).scalar()을 사용하는 경우, 이 orm 대신에 scalar()를 사용함으로써 코드를 더 간결하게 할 수 있다.

# From
result_count = db.execute(select(func.count()).select_from(query)).scalar()

# To
result_count = db.scalar(select(func.count()).select_from(query))

 

 

Exists - ORM (2.0)

from sqlalchemy select, exists

query = select(Sc).where(Sc.id == id, Sc.table == table)

exists_1 = db.scalar(exists(Sc).where(Sc.id == id, Sc.table == table).select())

exists_2 = db.scalar(exists(query).select())

 

exists_1과 같은 exists(Model)을 일반적으로 사용

exists_2와 같이 select()로 작성한 query가 있을 경우 select(Model)을 이용한 query 자체를 이용 가능

  -> 코드 재사용성 올라감

  !! 단 아직 공식 문서를 제대로 확인 못했고, 구체적인 쿼리를 확인해보지 않았음 !!