Skip to content

SQLAlchemy

Install

pip install sqlalchemy

Quickstart

Declare Models

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

Create an Engine

from sqlalchemy import create_engine

engine = create_engine("sqlite://", echo=True)

INFORMATION

Content

Emit CREATE TABLE DDL

Base.metadata.create_all(engine)

Create Objects and Persist

from sqlalchemy.orm import Session

with Session(engine) as session:
    spongebob = User(
        name="spongebob",
        fullname="Spongebob Squarepants",
        addresses=[Address(email_address="[email protected]")],
    )
    sandy = User(
        name="sandy",
        fullname="Sandy Cheeks",
        addresses=[
            Address(email_address="[email protected]"),
            Address(email_address="[email protected]"),
        ],
    )
    patrick = User(name="patrick", fullname="Patrick Star")
    session.add_all([spongebob, sandy, patrick])
    session.commit()

Simple SELECT

from sqlalchemy import select

session = Session(engine)

stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

for user in session.scalars(stmt):
    print(user)

SELECT with JOIN

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "sandy")
    .where(Address.email_address == "[email protected]")
)
sandy_address = session.scalars(stmt).one()
print(sandy_address)

declarative_base

declarative_baseSQLAlchemy에서 ORM(객체-관계 매핑) 모델을 정의할 때 사용하는 기본 클래스를 제공하는 함수입니다.

이 함수를 사용하면 Python 클래스를 데이터베이스의 테이블과 매핑할 수 있으며, 각 클래스는 데이터베이스의 테이블에 해당하고, 클래스의 속성은 테이블의 컬럼에 해당합니다.

예를 들어, 다음과 같이 사용할 수 있습니다:

from sqlalchemy.ext.declarative import declarative_base

# 기본 클래스를 생성
Base = declarative_base()

# 사용자 테이블을 나타내는 클래스 정의
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

이렇게 declarative_base()를 통해 만든 Base 클래스를 상속받아 데이터베이스 테이블을 표현하는 클래스를 정의할 수 있습니다.

이를 통해 SQLAlchemy가 데이터베이스와 Python 객체 간의 매핑을 자동으로 처리할 수 있게 됩니다.

MsSQL (pymssql) 연결

from sqlalchemy import create_engine
engine = create_engine('mssql+pymssql://username:passwd@host/database', echo=True)

Table 정의 방법

테이블 이름 설정:

from sqlalchemy.ext.declarative import declarative_base

# 기본 클래스를 생성
Base = declarative_base()

# 사용자 테이블을 나타내는 클래스 정의
class TbEpAiscrapchargeIfTable(Base):
    __tablename__ = "TB_EP_AISCRAPCHARGE_IF"
    __table_args__ = {"extend_existing": True}

    idx = Column(Integer, primary_key=True)
    # ...
__tablename__ = "TB_EP_AISCRAPCHARGE_IF"
속성은 데이터베이스에 실제로 생성될 테이블의 이름을 지정합니다. 여기서는 "TB_EP_AISCRAPCHARGE_IF"라는 이름으로 테이블이 생성됩니다.
__table_args__ = {"extend_existing": True}
테이블 생성 시 추가적인 설정을 지정하는 데 사용됩니다.
  • extend_existing=True - 만약 이 이름의 테이블이 이미 존재할 경우, 기존 테이블을 확장하겠다는 의미입니다. 이를 통해 기존 테이블의 스키마를 유지하면서 추가적인 컬럼이나 설정을 덧붙일 수 있습니다.

DateTime 기본 값 설정

from sqlalchemy import Column, DateTime, func

class TbEpAiscrapchargeIfTable(Base, BaseTable):
    __tablename__ = "TB_EP_AISCRAPCHARGE_IF"
    __table_args__ = {"extend_existing": True}

    insert_date = Column("INSERT_DATE", DateTime, default=func.now())

default=func.now() 를 설정하면 된다.

특정 날짜 이후의 rows 를 가져오고, 날짜 기준으로 오름차순 정렬하는 함수

from datetime import datetime
from sqlalchemy.orm import Session
from typing import List

def get_rows_after_date(session: Session, target_date: datetime) -> List[TbEpAiscrapchargeIfTable]:
    """
    특정 날짜 이후 insert_date 컬럼에 저장된 행을 가져오고, 날짜 기준으로 오름차순 정렬하는 함수.

    Parameters:
    - session: SQLAlchemy 세션 객체
    - target_date: datetime 객체로 특정 날짜

    Returns:
    - List[TbEpAiscrapchargeIfTable]: 해당 날짜 이후의 모든 행 목록 (오름차순 정렬)
    """
    return (session.query(TbEpAiscrapchargeIfTable)
            .filter(TbEpAiscrapchargeIfTable.insert_date > target_date)
            .order_by(TbEpAiscrapchargeIfTable.insert_date.asc())
            .all())

context 종료 후 column 접근 안되는 현상

다음과 같은 코드

def all_heat_infos():
    with DBManager.instance().query(TbEpAiscrapchargeIfTable) as q:
        return q.all()

all_heat_infos().idx  # Error!!

쿼리 결과를 접근하면 에러가 발생된다. context 가 종료되는 반환되는 객체는 DB 인스턴스와 연결이 끊어지기 때문이다.

해결하기 위해 expunge_all()를 사용하여 세션에서 조회한 모든 객체를 세션으로부터 분리하는 데 사용됩니다.

expunge_all()을 호출하면 세션이 더 이상 객체를 추적하지 않으므로, 해당 객체들이 독립적으로 존재하게 됩니다.

def all_heat_infos():
    with DBManager.instance().query(TbEpAiscrapchargeIfTable) as q:
        rows = q.all()
        q.session.expunge_all()
        return rows

all_heat_infos().idx  # OK

필터 사용 예제

def get_heat_infos_after_seq(if_seq: int):
    with DBManager.instance().query(TbEpAiscrapchargeIfTable) as q:
        rows = (
            q.filter(TbEpAiscrapchargeIfTable.if_seq >= if_seq)
            .order_by(asc(TbEpAiscrapchargeIfTable.if_seq))
            .all()
        )
        q.session.expunge_all()
        return rows

기준일 보다 이전인 모든 rows 를 제거하는 함수

from datetime import datetime
from sqlalchemy.orm import Session
from sqlalchemy import and_

def delete_rows_before_date(cutoff_date: datetime, session: Session):
    """
    주어진 기준일 이전의 모든 행을 삭제합니다.

    Parameters:
        cutoff_date (datetime): 기준일
        session (Session): SQLAlchemy 세션 객체
    """
    # 기준일 이전의 모든 행을 조회하여 삭제
    session.query(HeatInfoCandidateTable) \
           .filter(HeatInfoCandidateTable.reg_ddtt < cutoff_date) \
           .delete(synchronize_session=False)
    session.commit()

# 사용 예시
with DBManager.instance().Session.begin() as session:
    cutoff_date = datetime(2023, 1, 1)  # 예: 2023년 1월 1일 이전의 데이터 삭제
    delete_rows_before_date(cutoff_date, session)

synchronize_session=False는 성능 향상을 위해 세션 동기화를 피하는 옵션입니다.

LIKE 비교 예제

@staticmethod
def select_charging_info_with_heat_text(heat_no: int):
    with DBManager.instance().query(ChargingInfoTable) as q:
        rows = q.filter(ChargingInfoTable.heat_text.like(f"D_{heat_no}")).all()
        q.session.expunge_all()
        return rows

Troubleshooting

Adaptive Server is unavailable or does not exist

모든 정보가 정상일 때 다음과 같은 에러가 발생:

Traceback (most recent call last):
  File "src/pymssql/_pymssql.pyx", line 650, in pymssql._pymssql.connect
  File "src/pymssql/_mssql.pyx", line 2158, in pymssql._mssql.connect
  File "src/pymssql/_mssql.pyx", line 712, in pymssql._mssql.MSSQLConnection.__init__
  File "src/pymssql/_mssql.pyx", line 1884, in pymssql._mssql.maybe_raise_MSSQLDatabaseException
  File "src/pymssql/_mssql.pyx", line 1901, in pymssql._mssql.raise_MSSQLDatabaseException
pymssql._mssql.MSSQLDatabaseException: (20009, b'DB-Lib error message 20009, severity 9:\nUnable to connect: Adaptive Server is unavailable or does not exist ([email protected])\n')

create_engine 함수 호출시 전달되는 url 인자를 URL ENCODING 했는지 확인하자.

한 예시로 password 에 "@" 문자가 있어 구분에 문제가 발생될 수 있다. (e.g. 비밀번호가 "PassW@rd" 일 경우 mssql+pymssql://user:PassW@[email protected]:1433/db?charset=utf8 같이 전달되면 URL 구문에 문제가 발생된다)

이런 경우 "@"를 "%40" 같은 문자로 바꾸거나 urllib.parse.quote_plus 같은 함수를 사용하자.

from urllib.parse import quote_plus

password = quote_plus("PassW@rd")

Ignoring declarative-like tuple value of attribute '...'

SAWarning: Ignoring declarative-like tuple value of attribute 'if_seq': possibly a copy-and-paste error with a comma accidentally placed at the end of the line?
  class TbEpAiscrapchargeIfTable(Base, BaseTable):

복사-붙여넣기 해서 줄 끝에 쉼표가 붙은듯 함.

class TbEpAiscrapchargeIfTable(Base, BaseTable):
    __tablename__ = "TB_EP_AISCRAPCHARGE_IF"
    __table_args__ = {"extend_existing": True}

    if_seq = Column("IF_SEQ", Integer, nullable=True),  # 여기서 마지막 쉼표 빼자.

could not assemble any primary key columns for mapped table '...'

sqlalchemy.exc.ArgumentError: Mapper Mapper[TbEpAiscrapchargeIfTable(TB_EP_AISCRAPCHARGE_IF)] could not assemble any primary key columns for mapped table 'TB_EP_AISCRAPCHARGE_IF'

기본 키(primary key) 컬럼을 찾지 못했을 때 발생합니다. SQLAlchemy는 매핑된 각 테이블에 기본 키가 정의되어 있어야 객체-관계 매핑(ORM)을 설정할 수 있습니다.

이 문제를 해결하려면 기본 키 정의를 확인하자.

Column 객체에서 primary_key=True가 설정된 속성이 있는지 확인하세요.

See also

Favorite site