[Python] SQLAlchemy - 파이썬에서 여러 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)

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 쪽 기본값
`server_default`	server_default=text(“now()”)	DB 서버 기본값
`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

Base.metadata 는 데이터베이스의 구조 정보를 모아두는 객체
즉, 실제 데이터가 아닌 테이블 정의 정보
테이블 구조, foreign key, index, constraint 등의 정보를 가짐
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)

아래와 같이 생성됨

(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()

# 삭제 후 전체 데이터 조회 결과
pengoo king-pengoo@ex.com 10
pingoo pingping@ex.com 8
elen elen@ex.com 56
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()

Reference

SQLAlchemy

Overview — SQLAlchemy 2.0 Documentation

Twitter Facebook LinkedIn

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

Jongya

SQLAlchemy

(1) 소개

(2) ORM

(3) 장점

사용법

(1) 설치

(2) SQLite 다루기 예시

(3) DB와 연결하기 Engine

(4) 데이터 모델

a. 데이터모델의 선언

b. 테이블이름

c. 컬럼 선언

d. 컬럼 설정

e. 컬럼 타입

(5) DDL과 Base.metadata

(6) 세션 Session

(7) CRUD

a. 문법

b. CREATE 데이터 추가 : add

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

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

e. Update 데이터수정

f. Delete 데이터 삭제

(8) 고급 사용 : JOIN

a. 기본 문법

b. INNER JOIN

c. LEFT OUTER JOIN

d. RIGHT OUTER JOIN

e. FULL OUTER JOIN

다중 JOIN

Reference

Comments

You May Also Enjoy

[LangGraph] 카페 질문 답변 RAG 챗봇 만들기 실습

[Vector DB] Chroma 빠르게 구축 가능한 벡터 DB

[LangGraph] 그래프의 생성, 컴파일(연결), 실행 (with invoke/stream)

[LangGraph] Edge 엣지의 개념과 종류