한양과기고 에듀테크 수업도구 대시보드

psycopg asyncpg pool 파라미터 바인딩 SQLAlchemy

파이썬에서 PostgreSQL 을 쓰는 두 주력 드라이버 — 동기 psycopg(v3), 비동기 asyncpg.

1. psycopg 3 (동기·비동기)

import psycopg
from psycopg.rows import dict_row

with psycopg.connect("postgresql://school:pw@localhost/school", row_factory=dict_row) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM students WHERE grade = %s", (1,))  # 항상 파라미터로
        rows = cur.fetchall()
    conn.commit()

# 커넥션 풀
from psycopg_pool import ConnectionPool
pool = ConnectionPool("postgresql://...", min_size=2, max_size=10)
with pool.connection() as conn:
    conn.execute("INSERT INTO logs (msg) VALUES (%s)", ["hello"])

2. asyncpg

import asyncpg

pool = await asyncpg.create_pool(dsn, min_size=2, max_size=10)

async with pool.acquire() as conn:
    rows = await conn.fetch("SELECT * FROM students WHERE grade = $1", 1)  # $1, $2 ...
    one  = await conn.fetchrow("SELECT * FROM students WHERE id = $1", 5)
    n    = await conn.fetchval("SELECT count(*) FROM students")
    await conn.execute("UPDATE students SET grade = grade + 1 WHERE id = $1", 5)

PgBouncer(transaction 모드)와 함께면

asyncpg.create_pool(..., statement_cache_size=0) — 서버측 prepared statement 캐시 비활성화

3. 이 서버에서

fastapi-school(offset0xf.com)은 projects/db.py 에서 asyncpg 풀을 쓰고, lifespan 에서 생성/종료합니다. 트랜잭션은 async with conn.transaction():.

async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute("...")
        await conn.execute("...")   # 둘 중 하나 실패하면 자동 롤백

4. SQLAlchemy / FastAPI

더 보기 (FastAPI 레퍼런스)

SQLAlchemy 2.0 비동기는 내부적으로 asyncpg 를 드라이버로 씁니다 (postgresql+asyncpg://).

"정규화로 시작하고, 측정한 뒤에 비정규화하라"