1. SQLAlchemy Model / Entity 생성

SQLAlchemy 모델 및 엔티티를 생성

• SQLAlchemy로 모델을 생성하는 방법에는 두가지 방법이 있습니다.

1. SQLAlchemy ORM
2. SQLAlchemy Core

2. SQLAlchemy ORM 방식

주로 사용되는 방식으로 객체지향적으로 제공되는 문법을 통해 쿼리를 작성하고 트랜잭션 및 연결 풀링이 기본적으로 제공된다.

• declarative_base를 통한 모델 선언

• sessionmaker 객체를 활용한 session 객체 생성

• session 객체 하위의 add, commit(), scalars(), all() 등 사용

2.1 Example1

sqlalchemy 1.4 버전 이상에서 사용하는 방법으로, Declarative Mapping을 사용해 타입 힌트를 제공하고 더 명확한 타입 정의를 지원한다.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import Integer, String

class Base(DeclarativeBase):
    pass

class UserEntity(Base):
    __tablename__ = 'tbl_user'

    tbl_user_id: Mapped[int] = mapped_column(Integer, primary_key=True)
    user_id: Mapped[str] = mapped_column(String(45), nullable=False)
    user_name: Mapped[str] = mapped_column(String(45), nullable=False)
    user_group: Mapped[str] = mapped_column(String(45), nullable=False)

2.2 Example2

전통적인 방식인 Declarative Base를 사용하여 Column 클래스를 통해 필드를 정의한다.

from sqlalchemy import Column, String, Integer, SmallInteger
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class UserEntity(Base):
    __tablename__ = 'tbl_user'

    tbl_user_id = Column(Integer, primary_key=True)
    user_id = Column(String(45), nullable=False)
    user_name = Column(String(45), nullable=False)
    user_group = Column(String(45), nullable=False)

3. SQLAlchemy Core 방식

SQL 추상화 계층을 제공하고, 데이터베이스와 직접적인 SQL 쿼리 작성을 통해 표현하는 방식이다. 트랜잭션 및 연결 풀링에 대해 수동으로 처리해야한다.

• 저는 ORM 방식을 사용하여, 아래 예시는 example 예시입니다.

3.1 Example1

from sqlalchemy import create_engine, MetaData, Table, select

# engine 생성
engine = create_engine('sqlite:///example.db')
metadata = MetaData()

# Table, metadata, engine 객체 활용
user = Table('User', metadata, autoload_with=engine)
stmt = select(user)

3.2 Example2

from sqlalchemy import create_engine, MetaData, Table, select

# engine 생성
engine = create_engine('sqlite:///example.db')
metadata = MetaData()

# conn.execute()를 통한 실행
with engine.connect() as conn:
    result = conn.execute(stmt)
    for row in result:
        print(row)

4. References

1. yubi5050 blog