fix(expense-db): 앱 부팅 시 DDL 실행 제거 + lazy pool open

PostgreSQL 15+ 는 public 스키마 CREATE 권한을 PUBLIC 역할에서 회수했기 때문에
expense_app (앱 전용 비특권 계정) 으로 CREATE TABLE 시도하면
"permission denied for schema public" 으로 컨테이너 부팅 실패.

스키마 책임 분리:
- DDL: scripts/sql/expense_db_init.sql + expense_db_002_workflow_attachments.sql
       (superuser 가 사전 적용)
- 런타임: SELECT/INSERT/UPDATE/DELETE 만

ConnectionPool 도 open=False + lazy open 으로 변경. DB 가 잠시 끊겨도
컨테이너가 죽지 않게 함.
This commit is contained in:
2026-05-29 03:22:08 +09:00
parent 57ea186061
commit a4eaf9b770
+11 -25
View File
@@ -20,27 +20,17 @@ from psycopg_pool import ConnectionPool
from .store import CATEGORIES, METHODS, STATUSES
_DDL = """
CREATE TABLE IF NOT EXISTS expense_items (
id TEXT PRIMARY KEY,
owner TEXT NOT NULL,
spent_at DATE NOT NULL,
category TEXT NOT NULL,
method TEXT NOT NULL,
merchant TEXT NOT NULL DEFAULT '',
amount BIGINT NOT NULL DEFAULT 0 CHECK (amount >= 0),
memo TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT '작성중',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_expense_owner_spent ON expense_items (owner, spent_at DESC);
CREATE INDEX IF NOT EXISTS idx_expense_status ON expense_items (status);
"""
class ExpenseDBStore:
"""`ExpenseStore` 와 동일한 메서드 시그니처."""
"""`ExpenseStore` 와 동일한 메서드 시그니처.
스키마(테이블/인덱스/트리거)는 앱이 직접 만들지 않는다.
`scripts/sql/expense_db_init.sql` 과 `expense_db_002_*.sql` 을 통해
superuser 가 사전 적용한다. 앱 계정(expense_app)은 SELECT/INSERT/UPDATE/DELETE
권한만 받기 때문에 PostgreSQL 15+ 의 strict public-schema 정책과 충돌하지 않음.
연결 풀은 lazy open — 부팅 시점에 DB 가 잠시 끊겨도 컨테이너가 죽지 않게.
"""
def __init__(self, dsn: str, *, min_size: int = 1, max_size: int = 5):
self._pool = ConnectionPool(
@@ -48,17 +38,13 @@ class ExpenseDBStore:
min_size=min_size,
max_size=max_size,
kwargs={"row_factory": dict_row, "autocommit": True},
open=True,
open=False,
)
self._ensure_schema()
self._pool.open(wait=False)
def close(self) -> None:
self._pool.close()
def _ensure_schema(self) -> None:
with self._pool.connection() as conn:
conn.execute(_DDL)
# ── 조회 ──
def list_for(self, email: str) -> list[dict[str, Any]]:
email = email.lower().strip()