SQLAlchemy
Install
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
INFORMATION |
Content |
Emit CREATE TABLE DDL
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_base
는 SQLAlchemy에서 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 같은 함수를 사용하자.
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
- SQLModel
- Flask-SQLAlchemy (Flask)
- sqlacodegen - 자동으로 테이블 생성.