[Python] SQLAlchemy - 파이썬에서 여러 RDB를 하나의 코드로 다루기

[시리즈] Python 데이터베이스 마이그레이션 가이드

SQLAlchemy와 Alembic을 사용하여 데이터베이스 마이그레이션을 쉽게 해보자

• ▶ [Python] SQLAlchemy - 파이썬에서 여러 RDB를 하나의 코드로 다루기
• [Python] Alembic으로 구축하는 DB 마이그레이션
• [Python] ConnectorX - RDB 데이터를 빠르게 데이터프레임으로 읽어들이는 라이브러리

SQLAlchemy

(1) 소개

• Python 기반의 ORM 이자 SQL Toolkit
• SQL과 관련된 여러 작업을 유연하게 처리할 수 있게 도와주는 툴
• DB 종류에 관계 없이 일관된 코드 구조를 적용할 수 있음
• 한마디로 Python으로 DB를 쉽게 다루게 해주는 번역기의 역할

# 코드에서 SQL을 직접 작성하는 방식
query = "SELECT * FROM users WHERE id = 1;"
con.exec(query)
...

# ORM을 적용하면
user = session.get(User, 1)

(2) ORM

• Object Relational Mapper
• 관계형 데이터베이스 모델(RDB)과 Python 코드상의 객체로 매핑하는 방법
• 즉, 애플리케이션 코드의 클래스, 객체, 속성, 관계를 RDB의 테이블, 행, 컬럼, 외래키 관계 등과 대응시키는 기술

객체 지향 모델	관계형 DB 모델
Class	Table
Object instance	Row
Attribute	Column
Object relationship	Foreign Key / Join
Collection	One-to-many / Many-to-many relationship

(3) 장점

장점	설명
ORM과 Core 모두 제공	• 객체 기반 ORM 뿐만 아니라 SQL에 가까운 Core 방식도 사용 가능
복잡한 쿼리에 강함	• 단순 CRUD 뿐 아니라 JOIN, SUB Query, 트랜잭션 관리 등 복잡한 DB 로직 작성 가능
DB 종속성 완화	• Postgres, MySQL, SQLite 등 여러 DB를 같은 코드 구조로 다룰 수 있음 • 단, DB별 고유 기능은 별도 처리 필요
체계적인 트랜잭션 관리	• 컨텍스트 매니저(Session)로 commit과 rollback을 안정적으로 처리
커넥션 풀링 지원	• 애플리케이션에서 DB 연결을 재사용하는 connection pool을 기본적으로 제공
백엔드와 연계	• FastAPI 등 Python 백엔드 생태계와 연계 좋음

사용법

(1) 설치

• SQLAlchemy 설치

# pip
pip install SQLAlchemy

# uv
uv add SQLAlchemy

• 연결하려는 RDB 종류에 맞는 드라이버도 함께 설치해줘야 함

# 실습에서는 Postgres, Mariadb 두 가지를 사용할 것임

# pip
pip install psycopg2-binary pymysql

# uv
uv add psycopg2-binary pymysql

(2) SQLite 다루기 예시

• 본격적으로 사용법을 살펴보기에 앞서, 전체적인 흐름을 이해할 수 있는 간단한 예시를 보자

from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

# 1. DB 연결 설정
engine = create_engine("sqlite:///app.db", echo=True)

# 2. ORM Base 선언
class Base(DeclarativeBase):
    pass

# 3. 모델 선언
class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)

# 4. 테이블 생성
Base.metadata.create_all(engine)

# 5. Session 생성기
SessionLocal = sessionmaker(bind=engine)

# 6. 데이터 추가
with SessionLocal() as session:
    user = User(name="Kim", email="kim@example.com")
    session.add(user)
    session.commit()

# 7. 데이터 조회
with SessionLocal() as session:
    stmt = select(User).where(User.email == "kim@example.com")
    user = session.scalars(stmt).first()

    print("---- Result ----")
    print(user.id, user.name, user.email)

---- Result ----
1 Kim kim@example.com

(3) DB와 연결하기 Engine

• DB와의 연결은 보통 sqlalchemy.create_engine() 을 통해 수행한다.
• 해당 함수에 DB URL 파라미터를 꼭 제공해야 한다.
• 해당 함수는 engine 이라는 객체를 반환하는데, 하나의 engine 은 여러 DB API connection을 관리한다.

from sqlalchemy import create_engine

engine = create_engine("DB_URL", echo=True)

• 각각의 RDB에 따라 파라미터로 제공해야하는 DB_URL 형식이 다르다.

RDB	설치 패키지	SQLAlchemy URL 예시
SQLite	별도 드라이버 불필요	sqlite:///app.db
PostgreSQL	psycopg	postgresql+psycopg://user:pw@host:5432/db
MySQL	pymysql	mysql+pymysql://user:pw@host:3306/db
MariaDB	pymysql	mysql+pymysql://user:pw@host:3306/db
Oracle	oracledb	oracle+oracledb://user:pw@host:1521/?service_name=…
SQL Server	pyodbc	mssql+pyodbc://user:pw@host:1433/db?driver=…

from sqlalchemy import create_engine

# SQLite (파일 기반 작동시)
sqlite_url = "sqlite:///app.db"
engine = create_engine(sqlite_url)

# SQLite (메모리 기반 작동시)
sqlite_url = "sqlite:///:memory:"
engine = create_engine(sqlite_url)

# PostgreSQL
p_url = "postgresql+psycopg://user:password@서버ip:포트번호/DB이름"
engine = create_engine(p_url)

# MySQL / MariaDB
m_url = "mysql+pymysql://user:password@서버ip:포트번호/DB이름?charset=utf8mb4"
engine = create_engine(m_url)

# Oracle
o_url = "oracle+oracledb://user:password@서버ip:포트번호/?service_name=서비스이름"
o_url = "oracle+oracledb://user:password@서버ip:포트번호/SID" # sid 방법

• echo=True 옵션을 주면, 트랜잭션 실행시 발생하는 로그가 출력된다.

from sqlalchemy import create_engine

engine = create_engine("DB_URL", echo=True)
# echo : 트랜잭션 실행에 따른 로그 출력 여부

(4) 데이터 모델

a. 데이터모델의 선언

• SQLAlchemy 그리고 ORM의 핵심적인 부분
• 데이터 모델을 파이썬의 Class 형식으로 표현하는 방법이다.
• 보통 DeclarativeBase 클래스를 상속받아 사용. (SQLAlchemy ORM 모델 클래스들이 상속받는 기준 클래스)
• 즉, Base 는 ORM 모델들을 SQLAlchemy가 인식할 수 있도록 등록하는 중심점이라고 볼 수 있다.

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class User(Base):
    # 테이블 이름
    __tablename__ = "users"
    
    # 컬럼
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)

    # 그 외 테이블 설정
    __table_args__ = {
        "schema": "service_a" # e.g. postgres : 이 테이블이 속한 스키마 지정
        "mysql_collate": "utf8mb4_bin" # e.g. mysql : 이 테이블의 collation
    }

b. 테이블이름

• 테이블 이름은 데이터모델 클래스의 __tablename__ 속성값에 할당한다.

class User(Base):
    # 테이블 이름
    __tablename__ = "users"

c. 컬럼 선언

• RDB의 컬럼은 데이터모델 클래스의 “속성(attribute)”로 표현한다.
• 이 때, 각 속성에 대응해 mapped_column() 함수를 할당함으로써 해당 속성이 컬럼임을 설정한다.
• 또한 Mapped 를 사용해 속성의 데이터타입과, 이와 매핑되는 RDB 컬럼이나 관계를 표시하는 타입힌팅을 할 수 있다.

요소	역할
Mapped[int]	Python 타입 힌트 + ORM 매핑 대상 표시
mapped_column()	실제 DB 컬럼 설정

class User(Base):
    ...
    # 컬럼
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)

d. 컬럼 설정

• mapped_column() 함수 안에서 타입, 제약조건, 기본값, 인덱스, 외래키, 자동증가, 주석 을 설정할 수 있다.

옵션	예시	의미
_type	String(50)	데이터타입
primary_key	primary_key=True	기본키
nullable	nullable=False	NULL 허용 여부
unique	unique=True	중복 금지
index	index=True	인덱스 생성
default	default=datetime.utcnow()	Python 쪽 기본값 (SQLAlchemy 단에서 Insert할 때 적용) DDL에 반영되지 않음. 동적인 값 할당 가능.
server_default	server_default=text(“now()”)	DB 서버 기본값 (DB에 Insert될 때 적용) DDL에 반영됨. 문자열로 전달해야 함.
onupdate	onupdate=datetime.utcnow	UPDATE 시 자동 값
ForeignKey	ForeignKey(“users.id”)	외래키
autoincrement	autoincrement=True	자동 증가
comment	comment=”설명”	DB 컬럼 주석
doc	doc=”설명”	Python 속성 설명
deferred	deferred=True	지연 로딩
info	info={…}	앱 전용 메타 정보
name	mapped_column(“db_name”)	DB 컬럼명 별도 지정
sort_order	sort_order=10	컬럼 정렬 순서

class User(Base):
    ...
    # 컬럼
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100), unique=True)

e. 컬럼 타입

• mapped_column() 에서 설정할 수 있는 데이터타입은 아래와 같다.
• 각 데이터타입은 sqlalchemy 모듈 내 클래스로 선언되어있다.

SQLAlchemy 타입	Python 타입	DB에서의 일반적 의미
Integer	int	정수
BigInteger	int	큰 정수
SmallInteger	int	작은 정수
String(길이)	str	VARCHAR
Text	str	긴 문자열
Boolean	bool	참/거짓
Date	date	날짜
DateTime	datetime	날짜+시간
Time	time	시간
Float	float	부동소수점
Numeric(정밀도, 소수점)	Decimal	정확한 숫자, 금액
JSON	dict, list	JSON 데이터
LargeBinary	bytes	바이너리 데이터

(5) DDL과 Base.metadata

a. Base.metadata

• Base.metadata 는 데이터베이스의 구조 정보를 모아두는 객체
• 즉, 실제 데이터가 아닌 테이블 정의 정보
• 테이블 구조, foreign key, index, constraint 등의 정보를 가짐

b. 테이블들 생성하기

• create_all() 메서드를 통해 Base를 상속받은 모든 데이터모델 테이블을 생성할 수 있다.
• create_all() 메서드를 실행할 때에는, bind 파라미터에 대상 DB 연결(engine)을 할당해줘야 한다.

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

engine = create_engine("DB URL", echo=True)

class Base(DeclarativeBase):
    pass

class Users(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(Integer, comment="user id", primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(200), comment="real name of user")
    email: Mapped[str] = mapped_column(String(200), comment="email address of user")
    age: Mapped[int] = mapped_column(Integer, comment="age of user")

class Plants(Base):
    __tablename__ = "plants"
    id: Mapped[int] = mapped_column(Integer, comment="plant id", primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(200), comment="nickname of plant")
    birthday: Mapped[str] = mapped_column(DateTime, comment="birthday of plant", nullable=True)
    user_id: Mapped[int] = mapped_column(Integer, comment="owner of plant")

Base.metadata.create_all(bind=engine)

아래와 같이 생성됨

c. 단일 테이블 생성하기

• 단일 테이블을 생성할 때에는 데이터모델클래스.__table__.create() 메서드를 사용하면 된다.

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

engine = create_engine("DB URL", echo=True)

class Base(DeclarativeBase):
    pass

class Users(Base):
    __tablename__ = "users"
    ...(생략)

class Plants(Base):
    __tablename__ = "plants"
    ...(생략)

Users.__table__.create(bind=engine) # <-- 이렇게

그러면 users 테이블만 생성된 것을 볼 수 있다.

(6) 세션 Session

• ORM 작업 단위이자, 객체 상태 관리와 트랜잭션 관리 계층을 다루는 인스턴스
• sessionmaker 를 통해 세션(작업) 인스턴스를 생성할 수 있다.
• 파라미터로 bind 에 연결할 DB에 대한 엔진을 제공해줘야 한다.
• Session 은 자원이므로 컨텍스트 매니저 with 와 함께 사용할 수 있다.
• Session 을 통해 CRUD작업을 수행할 수 있다.

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

with Session() as s:
    user = Users(name="Pengoo", email="pengoo@pengpeng.com", age=10)
    plant = Plants(name="EucalyPengoo", user_id=1)
    s.add(user)
    s.add(plant)
    s.commit()

Session 을 통한 작업 결과

(7) CRUD

a. 문법

• Session 을 통해 CRUD 작업을 수행할 수 있다.
• add() : create, 생성 작업을 수행한다.
• get() : 기본키(Primary Key)를 통한 데이터 조회 기능
• select() : SQL의 SELECT 와 동일한 동작을 하며, where절 등도 적용이 가능하다.
• 수정(upadte)는, 조회해온 데이터 인스턴스의 속성값을 수정하고 commit 함으로써 이루어진다.
• delete() : 조회해온 데이터 인스턴스를 파리미터로 하여, 해당 데이터를 삭제한다.

b. CREATE 데이터 추가 : add

• Session.add() 메서드로 수행하며, 데이터 인스턴스를 파라미터로 받는다.

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

names  = ["pingoo", "edward", "elen", "iverson"]
emails = ["pingoo@example.com", "ed@ex.com", "elen@ex.com", "shoot@76s.com"]
ages   = [8, 23, 56, 20]

with Session() as s:
    for name, email, age in zip(names, emails, ages):
        user = Users(name=name, email=email, age=age)
        s.add(user)    
    s.commit()

c. Retrieve 데이터조회 (1) : get

• Session.get(데이터모델, 기본키) 로 데이터 조회를 수행할 수 있다.

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

with Session() as s:
    result = s.get(Users, 1)

print(result.id, result.name, result.email, result.age)

1 Pengoo pengoo@pengpeng.com 10

d. Retrieve 데이터조회 (2) : select

• sqlalchemy.session 메서드를 이용하여 데이터조회를 수행한다.
• Session.scalars : 반환된 결과의 첫 번째 row.
• Session.scalar : 반환된 모든 결과 row들.

from sqlalchemy import select

Session = sessionmaker(bind=engine)

with Session() as s:
    result = select(Users).where(Users.age < 20)
    users = s.scalars(result).all()
    
    for user in users:
        print(user.id, user.name, user.email, user.age)

1 Pengoo pengoo@pengpeng.com 10
2 pingoo pingoo@example.com 8

• 전체 데이터 조회 : 전체 데이터 조회(SQL의 SELECT *)를 하려면 where 조건을 안붙이면 된다.

from sqlalchemy import select

Session = sessionmaker(bind=engine)

with Session() as s:
    result = select(Users)
    users = s.scalars(result)
    for u in users:
        print(u.id, u.name, u.email, u.age)

• 특정 컬럼 조회 : 특정 컬럼만 조회(SQL의 SELECT a, b …)를 하려면 조회하고자 하는 컬럼을 select() 메서드 안에 명시하며, execute() 함수로 쿼리를 실행시킨다.

from sqlalchemy import select

Session = sessionmaker(bind=engine)

with Session() as s:
    result = select(Users.name, Users.age)
    users = s.execute(result).all()
    for u in users:
        print(u.name, u.age)

• Alias : 조회하는 컬럼에 별칭을 붙여 조회하려면 lable() 함수를 사용한다.
• Alias를 적용했다면, 결과에서 꺼내 쓸 때에도 속성명을 별칭으로 호출해줘야 한다.

from sqlalchemy import select
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

with Session() as s:
    result = select(Users.name.label("who"),
                    Users.age.label("years_from_birth"))
    rows = s.execute(result).all()
    for row in rows:
        print(row.who, row.years_from_birth) # 별칭으로 꺼내 씀 (또는 그냥 a, b)

e. Update 데이터수정

• 조회해온 데이터 인스턴스에 대한 속성값을 수정한 뒤, Session.commit 을 수행한다.

from sqlalchemy import select

Session = sessionmaker(bind=engine)

with Session() as s:
    # with get
    pingoo = s.get(Users, 2)
    pingoo.email = "pingping@ex.com"
    # with select
    result = select(Users).where(Users.name == "pengoo")
    pengoo = s.scalar(result)
    pengoo.name = "pengoo"
    pengoo.email = "king-pengoo@ex.com"
    # commit
    s.commit()

# 수정 후 조회 결과
1 pengoo king-pengoo@ex.com 10
2 pingoo pingping@ex.com 8

f. Delete 데이터 삭제

• Session.delete() 메서드로 데이터 삭제가 가능하며, 삭제하려는 데이터 인스턴스를 파라미터로 제공해야 한다.

from sqlalchemy import select

Session = sessionmaker(bind=engine)

with Session() as s:
    # edward 를 삭제
    result = select(Users).where(Users.name == "edward")
    ed = s.scalar(result)
    s.delete(ed)
    s.commit()

# 삭제 후 전체 데이터 조회 결과
1 pengoo king-pengoo@ex.com 10
2 pingoo pingping@ex.com 8
4 elen elen@ex.com 56
5 iverson shoot@76s.com 20

(8) 고급 사용 : JOIN

a. 기본 문법

• select() 메서드와 .join() 메서드를 결합하여 사용한다.
• 또한 JOIN 을 사용하는 경우, 쿼리의 실행은 일반적으로 execute() 메서드를 이용한다.

from sqlalchemy import select

Session = sessionmaker(bind=engine)

with Session() as s:
    result = (
        select(Plants, Users) # 조회할 데이터 (여기선 SELECT Plants.*, Users.*)
        .join(Users,          # 조인할 테이블
              Plants.user_id == Users.id) # JOIN 조건
    )
    rows = s.execute(result).all()
    for plant, user in rows:
        print(user.name, plant.name)

pengoo EucalyPengoo

b. INNER JOIN

• 일반적으로 .join() 은 inner join 이 수행된다.
• 따라서 기본 문법 예시와 같다.

c. LEFT OUTER JOIN

• .outerjoin() 메서드를 사용하거나
• .join(isouter=True) 파라미터를 사용하면 된다.

from sqlalchemy import select
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

with Session() as s:
    result = (
        select(Users.name.label("user_name"),
               Plants.name.label("plant_name")) # 조회할 데이터 (여기선 SELECT Plants.*, Users.*)
        .join(Plants,                     # 조인할 테이블
              Users.id == Plants.user_id, # JOIN 조건
              isouter=True)               # OUTER JOIN
    )
    rows = s.execute(result).all()
    for u, p in rows:
        print(u, p)

pengoo  EucalyPengoo
pingoo  None
elen    None
iverson None

d. RIGHT OUTER JOIN

• SQLAlchemy 는 직접 RIGHT OUTER JOIN 을 수행하지 않는다.
• 대신 테이블 순서를 뒤집고 LEFT OUTER JOIN 을 수행하면 된다.

e. FULL OUTER JOIN

• join() 메서드에서 full=True 파라미터를 추가한다.
• 단, FULL OUTER JOIN 을 지원하는 RDB에 대해서만 실행할 수 있다.

from sqlalchemy import select
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)

with Session() as s:
    result = (
        select(Users.name.label("user_name"),
               Plants.name.label("plant_name"))
        .join(Plants,                    
              Users.id == Plants.user_id,
              full=True)               
    )
    rows = s.execute(result).all()
    for u, p in rows:
        print(u, p)

다중 JOIN

• join() 을 연속ㅇ로 수행하면 된다.

stmt = (
    select(User)
    .join(User.posts)
    .join(Post.comments)
    .where(Comment.content.contains("좋아요"))
)

users = session.scalars(stmt).all()

(9) 쿼리문을 직접 사용하기

• SQLAlchemy 와 함께 직접 텍스트로 된 쿼리문을 사용할 수도 있다.
• 엔진을 생성한 뒤, engine.begin() 으로부터 반환받은 연결 컨텍스트 내에서 처리하면 되며
• 주의할 점은, 쿼리문을 sqlalchemy.text() 함수로 감싸줘야 한다는 것이다.

from sqlalchemy import text # <-- text 함수 import

# 테이블명 변경
with target_engine.begin() as conn:
    # f-string으로 만든 문자열을 text()로 감싸줌
    conn.execute(text(f"RENAME TABLE {tb1} TO {'drop_' + tb1};"))
    conn.execute(text(f"RENAME TABLE {tb2} TO {tb1};"))
    conn.execute(text(f"DROP TABLE {'drop_' + tb1};"))

• 데이터에 대한 CRUD의 경우, 하나의 컨텍스트 안에서 수행되는 작업은 단일 트랜잭션으로 취급된다.
• 단, 위처럼 DDL문인 경우에는 DBMS에 따라 트랜잭션 처리가 다르므로 주의!

Reference

SQLAlchemy

Overview — SQLAlchemy 2.0 Documentation

[시리즈] Python 데이터베이스 마이그레이션 가이드

SQLAlchemy와 Alembic을 사용하여 데이터베이스 마이그레이션을 쉽게 해보자

• ▶ [Python] SQLAlchemy - 파이썬에서 여러 RDB를 하나의 코드로 다루기
• [Python] Alembic으로 구축하는 DB 마이그레이션
• [Python] ConnectorX - RDB 데이터를 빠르게 데이터프레임으로 읽어들이는 라이브러리