SQLAlchemy

• SQLAlchemy 2.0 Docs

설치

python3 -m pip install sqlalchemy

PostgreSQL

python3 -m pip install psycopg[binary]

MySQL

python3 -m pip install pymysql

Engine 생성

PostgreSQL

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy import Engine
from sqlalchemy import create_engine as _create_engine


class DBConfig(BaseSettings):
    host: str = "localhost"
    port: int = 5432
    user: str = "postgres"
    password: str = Field(exclude=True)
    database: str
    connection_options: str | None = None

    model_config = SettingsConfigDict(env_prefix="db_", env_file=(".env", ".env.local"), extra="ignore")

    def to_url(self) -> str:
        url = f"postgresql+psycopg://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"
        if self.connection_options:
            url += f"?{self.connection_options}"
        return url


def create_engine(config: DBConfig) -> Engine:
    return _create_engine(config.to_url())

MySQL

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy import Engine
from sqlalchemy import create_engine as _create_engine


class DBConfig(BaseSettings):
    host: str = "localhost"
    port: int = 3306
    user: str = "root"
    password: str = Field(exclude=True)
    database: str
    connection_options: str | None = None

    model_config = SettingsConfigDict(env_prefix="db_", env_file=(".env", ".env.local"), extra="ignore")

    def to_url(self) -> str:
        url = f"mysql+pymysql://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"
        if self.connection_options:
            url += f"?{self.connection_options}"
        return url


def create_engine(config: DBConfig) -> Engine:
    return _create_engine(config.to_url())

Session 생성

• SQLAlchemy 2.0: Session Basics

# from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker


engine = create_engine(DBConfig())
# Session == sessionmaker(engine)
SessionLocal = sessionmaker(engine, autocommit=False, autoflush=False)

# session을 열고, 닫음
with SessionLocal() as session:
    ...

with SessionLocal() as session:
    # 에러 발생 시 rollback, 성공 시 commit
    with session.begin()
        ...