Naki-404/naki-postgres

PostgreSQL 및 SQLAlchemy 2

사용 시기

• 테이블, 제약 조건 및 인덱스 설계.
• SQLAlchemy 2 모델 및 쿼리 작성 (동기/비동기).
• 쿼리가 느리거나, 잠금이 쌓이거나, 커넥션이 부족할 때.
• 실시간 트래픽이 있는 테이블에 대해 Alembic 마이그레이션을 작성할 때.

파괴적인 데이터 변경의 경우 naki-safety도 적용하세요. 프로젝트에 프로젝트 규칙 스킬이 있다면 (예: naki-myapp) 그것을 따르세요.

스키마 규칙

1. 기본 키: bigint generated always as identity 또는 UUID (가능하면 v7, 인덱스 지역성 향상). 열거형이 중요한 경우 순차적 ID를 공개적으로 노출하지 마세요.
2. 타입: timestamptz (시간대 없는 timestamp는 절대 사용 금지), 길이 제한이 변경될 수 있는 경우 varchar(n) 대신 text와 CHECK 사용, 금액은 numeric, jsonb (json 아님), boolean not null default false.
3. 제약 조건은 최후의 방어선입니다: NOT NULL 기본, UNIQUE, CHECK, 명시적인 ON DELETE 선택이 있는 외래 키.
4. 조인이나 계단식(cascade)에서 사용되는 모든 외래 키에 인덱스를 생성하세요. PostgreSQL은 자동으로 생성하지 않습니다.
5. 제약 조건과 인덱스 이름은 SQLAlchemy MetaData의 명명 규칙을 통해 지정하여 Alembic diff가 안정적이도록 하세요.
6. jsonb는 진정으로 가변적인 데이터에만 사용하고, 모델링을 회피하는 용도로 사용하지 마세요. 자주 쿼리되는 키는 실제 컬럼으로 전환하세요.
7. 소프트 삭제는 반드시 필요한 경우에만 사용하세요. 사용한다면 부분 인덱스 WHERE deleted_at IS NULL을 추가하세요.

인덱스 규칙

쿼리 패턴	인덱스
한 컬럼에 대한 동등 및 범위 조건	B-tree
WHERE a = ? AND b > ? ORDER BY b	복합 (a, b), 동등 조건 컬럼 우선
행의 부분 집합 (오픈 항목)	부분 WHERE status = 'open'
lower(email) = ?	lower(email)에 대한 표현식 인덱스
jsonb 포함 @>	GIN
전문 검색, trigram ILIKE '%x%'	tsvector 또는 pg_trgm을 이용한 GIN
대규모 추가 전용 시계열	타임스탬프에 BRIN

• 모든 인덱스는 쓰기와 디스크 비용이 듭니다. 사용되지 않는 인덱스는 제거하세요 (pg_stat_user_indexes.idx_scan = 0이 장기간 지속).
• 읽기가 많은 경우 인덱스 전용 스캔을 위해 INCLUDE (...)를 사용한 커버링 인덱스를 고려하세요.

쿼리 튜닝 워크플로우

1. 실제 쿼리 (echo 활성화 또는 log_min_duration_statement 설정)와 실제 파라미터를 확인하세요.
2. 프로덕션과 유사한 데이터셋에서 EXPLAIN (ANALYZE, BUFFERS)를 실행하세요. 100행짜리 개발 테이블의 실행 계획은 절대 신뢰하지 마세요.
3. 다음 항목을 찾으세요: 선택적 필터가 있는 대형 테이블의 Seq Scan, 실제와 크게 다른 행 추정치 (오래된 통계: ANALYZE 실행), 많은 행에 대한 중첩 루프, 디스크로 유출되는 정렬, 높은 shared read.
4. 가장 저렴한 원인을 수정하세요: 인덱스 추가 또는 조정, 조건자를 sargable하게 재작성, N+1 제거, keyset 기반 페이지네이션.
5. EXPLAIN ANALYZE를 다시 실행하고 타이밍을 비교하세요. pg_stat_statements를 사용하여 총 시간 기준 상위 쿼리를 찾으세요.

SQLAlchemy 2 규칙

1. 2.0 스타일을 사용하세요: select(), session.execute(), session.scalars(), Mapped[...], mapped_column.
2. 작업 단위(요청, 작업)당 하나의 세션을 사용하세요. 스레드나 태스크 간에 세션을 공유하지 마세요.
3. N+1을 피하세요: 관계형을 명시적으로 selectinload (컬렉션) 또는 joinedload (다대일)로 로드하세요. 비동기 코드에서 지연 로드는 오류를 발생시킵니다. 동기 코드에서도 이를 잡기 위해 관계형에 lazy="raise"를 설정하세요.
4. 바인드 파라미터만 사용하세요. 사용자 입력을 SQL 문자열( text() 포함)에 직접 포맷하지 마세요.
5. 풀 크기와 pool_pre_ping=True를 설정하세요. 모든 앱 인스턴스의 총 커넥션 수는 max_connections 미만이어야 합니다 (워커가 많으면 PgBouncer 사용).
6. 트랜잭션을 짧게 유지하세요. 트랜잭션을 보유한 상태에서 다른 서비스로 네트워크 호출을 하지 마세요.

from datetime import datetime
from sqlalchemy import ForeignKey, Index, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, selectinload

class Base(DeclarativeBase):
    pass

class Incident(Base):
    __tablename__ = "incidents"
    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str] = mapped_column(String(16), index=True)
    opened_at: Mapped[datetime]
    alerts: Mapped[list["Alert"]] = relationship(back_populates="incident", lazy="raise")
    __table_args__ = (Index("ix_incidents_open", "opened_at", postgresql_where="status = 'open'"),)

async def open_incidents(session: AsyncSession, limit: int) -> list[Incident]:
    stmt = (
        select(Incident)
        .where(Incident.status == "open")
        .order_by(Incident.opened_at.desc(), Incident.id.desc())
        .limit(limit)
        .options(selectinload(Incident.alerts))
    )
    return list((await session.scalars(stmt)).all())

트랜잭션 및 잠금

• 기본 READ COMMITTED는 대부분의 작업에 적합합니다. 읽은 내용을 기반으로 수정할 행을 잠그려면 SELECT ... FOR UPDATE를 사용하세요.
• 큐의 경우: FOR UPDATE SKIP LOCKED를 사용하면 여러 워커가 서로 블로킹하지 않고 작업을 가져갈 수 있습니다.
• Upsert: 읽기 후 삽입 경쟁 대신 INSERT ... ON CONFLICT (key) DO UPDATE를 사용하세요.
• 앱 역할에 대해 lock_timeout과 statement_timeout을 설정하여 하나의 잘못된 쿼리가 전체를 멈추지 않도록 하세요.
• 데드락은 서로 다른 순서로 행을 잠글 때 발생합니다. 일관된 순서로 잠그세요 (id로 정렬).

Alembic 및 무중단 변경

자동 생성된 마이그레이션은 항상 검토하세요. 자동 생성은 이름 변경 (drop 후 add를 생성함) 및 일부 타입 변경을 놓칠 수 있습니다.

호환성을 깨는 변경에는 확장-수축(Expand and contract) 방식을 사용하세요:

1. 확장: 새 컬럼을 nullable로 추가 (또는 상수 기본값 사용, PG 11+에서는 즉시 적용됨), 새 테이블 추가, 또는 CONCURRENTLY로 새 인덱스 추가.
2. 이전과 새 코드를 모두 작성하는 코드를 배포합니다.
3. 백필(Backfill) 을 마이그레이션 외부에서 배치 단위로 (예: 트랜잭션당 5천 행) 수행하거나, 진행 상황 로깅과 함께 별도의 데이터 마이그레이션에서 수행합니다.
4. 새 코드를 읽는 코드를 배포합니다.
5. 수축: 이후 릴리스에서 이전 컬럼을 삭제합니다.

대규모 라이브 테이블에서 위험한 작업:

작업	더 안전한 접근 방식
CREATE INDEX	CREATE INDEX CONCURRENTLY (Alembic: with op.get_context().autocommit_block(): 내부에서 postgresql_concurrently=True 사용)
NOT NULL 추가	CHECK (col IS NOT NULL) NOT VALID 추가, VALIDATE CONSTRAINT 실행, 그 후 NOT NULL 설정
외래 키 추가	NOT VALID로 추가, 그 후 VALIDATE CONSTRAINT 실행
컬럼 이름 변경	확장-수축 방식 사용, 이전 코드가 실행 중인 상태에서 제자리 변경 절대 금지
컬럼 타입 변경	새 컬럼 추가 후 백필, 테이블을 재작성하는 ALTER TYPE는 사용 금지

• 마이그레이션에서 SET lock_timeout = '5s'를 설정하여 긴 트랜잭션 뒤에서 대기하지 않고 빠르게 실패하도록 하세요.
• 모든 마이그레이션은 작동하는 downgrade() 또는 불가능한 이유를 설명하는 메모를 포함해야 합니다.
• 실제 데이터 복사본에 대해 마이그레이션 업, 다운, 다시 업을 테스트하세요.

체크리스트

• timestamptz, NOT NULL 기본값, 명명된 제약 조건, 인덱싱된 외래 키.
• 쿼리 파라미터화됨; SQL에 f-문자열 없음.
• N+1 없음; 관계형이 명시적으로 로드됨.
• 느린 쿼리가 실제 데이터로 EXPLAIN (ANALYZE, BUFFERS)를 통해 확인됨.
• 트랜잭션 짧음; 타임아웃 설정됨.
• 마이그레이션 검토됨, 되돌리기 가능, 필요한 경우 CONCURRENTLY / NOT VALID 사용.
• 백필은 배치 처리되고 재개 가능함.

일반적인 실수

• OFFSET 100000 페이지네이션; 인덱싱된 정렬 키에 keyset을 사용하세요.
• WHERE date(created_at) = ?는 인덱스를 무효화합니다. 원시 컬럼에 범위 조건을 사용하세요.
• trigram 인덱스 없이 LIKE '%term%' 사용.
• await session.commit()을 잊거나, 루프 내에서 행마다 커밋하는 경우.
• alembic revision --autogenerate를 실행하고 읽지 않은 채 커밋하는 경우.
• 외부 API를 호출하는 동안 트랜잭션을 열어두는 경우.
• 로컬 시간으로 타임스탬프를 저장하는 경우.