[FastAPI] SQLModel

1. SQLModel이란?

기존 Python SQL ORM 구현체인 SQLAlchemy와 검증 패키지 Pydantic을 결합한 Python 라이브러리이다.

데이터베이스 작업을 보다 간결하고 직관적으로 수행할 수 있다.

SQLAlchemy의 ORM 기능을 유지하면서도 Pydantic의 데이터 검증 기능을 제공하는 것이 특징이다.

 

 

2. SQLAlchemy vs SQLModel

SQLAlchemy는 두 가지 주요 API 스타일을 제공한다.

• SQLAlchemy Core: 저수준 SQL 표현을 사용하여 SQL을 직접 다룰 수 있도록 설계됨
• SQLAlchemy ORM: 객체-관계 매핑(Object-Relational Mapping)을 통해 Python 클래스로 DB 모델을 다룰 수 있음

SQLModel은 SQLAlchemy ORM을 기반으로 동작하며, 다음과 같은 차이점이 있다.

특징	SQLAlchemy	SQLModel
데이터검증	없음	Pydantic 기반 검증
필드 선언 방식	Column() 사용	Python 타입 힌트와 Field() 사용
간결성	상대적으로 복잡	상대적으로 간결

 

 

3. 코드 예시

1. 모델 정의

SQLAlchemy

from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)

 

SQLModel

from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    id: int = Field(primary_key=True)
    name: str
    age: int

 

 

2. 데이터 삽입

SQLAlchemy

SessionLocal = sessionmaker(bind=engine)
session = SessionLocal()

user = User(name="Alice", age=25)
session.add(user)
session.commit()
session.close()

 

SQLModel

from sqlmodel import Session

with Session(engine) as session:
    user = User(name="Alice", age=25)
    session.add(user)
    session.commit()

 

3. 데이터 조회

SQLAlchemy

session = SessionLocal()
users = session.query(User).filter(User.age > 20).all()
for user in users:
    print(user.name, user.age)
session.close()

 

SQLModel

from sqlmodel import select

with Session(engine) as session:
    statement = select(User).where(User.age > 20)
    results = session.exec(statement)
    for user in results:
        print(user.name, user.age)

 

4.  비동기 구현

SQLAlchemy 1.4 부터 비동기(async) 지원이 추가되어, SQLModel도 이를 활용해 비동기 세션을 사용할 수 있다.

1. SQLAlchemy 비동기 엔진 생성 (SQLModel도 동일)

from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine

DATABASE_URL = "sqlite+aiosqlite:///database.db"
async_engine = create_async_engine(DATABASE_URL, echo=True)

 

2-1. SQLAlchemy 비동기 세션 생성 및 활용

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
import asyncio

async def get_users():
    async with AsyncSession(async_engine) as session:
        statement = select(User)
        results = await session.execute(statement)
        users = results.scalars().all()
        for user in users:
            print(user.name, user.age)

asyncio.run(get_users())

 

2-2. SQLModel : SQLAlchemy의 비동기 엔진 활용

아직 별도의 비동기 ORM 기능 내장 x -> SQLModel 래퍼가 동기적으로 작동

from sqlmodel.ext.asyncio.session import AsyncSession

async def get_users_sqlmodel():
    async with AsyncSession(async_engine) as session:
        statement = select(User)
        results = await session.execute(statement)
        users = results.scalars().all()
        for user in users:
            print(user.name, user.age)

asyncio.run(get_users_sqlmodel())

 

 

4. 정리

 

SQLModel은 의 강력한 기능과 Pydantic의 데이터 검증 기능을 결합한 라이브러리로, 간결하고 직관적인 코드가 가능 

• SQLModel = SQLAlchemy ORM + Pydantic 데이터 검증
• FastAPI 와 높은 호환성 (개발자가 FastAPI 개발자 중 한 명)
• 다만 아직 비동기 기능이 자체 탑재 되지 못하고 SQLAlchemy를 직접 활용해야 한다.