feat(vacation): 휴가 관리 모듈 — 월간 달력 bar/연차·반차 신청/승인 워크플로/공휴일·연차 설정
- vacation_db 전용 저장소(VacationDBStore), VACATION_DB_URL 미설정 시 '설정 필요' 안내 - 월간 달력: 구글식 휴가 bar(주 단위 lane 배정), 일/공휴일 빨강·토요일 파랑 - 신청 폼: 종류/기간/시작·종료 구분(종일/오전/오후)/사유 + 예상 일수 미리보기 - 일수 계산은 서버(compute_days)에서 주말+공휴일(is_red) 제외 재계산 - 워크플로: 작성중/반려→제출→승인|반려, 취소 soft delete, 본인만 수정(작성중/반려) - 권한: vacation(접근)/vacation_approver(승인), admin 통과. 설정은 admin 전용 - 공휴일(vacation_holidays) DB 기반 provider + settings CRUD, 연차잔여(vacation_balances) - 엑셀 내보내기(openpyxl), scripts/sql/vacation_db_init.sql(멱등, 2026 공휴일 seed) - main.py 등록, 메뉴 ready 전환, docs/.env.example/CLAUDE.md 갱신 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,485 @@
|
||||
"""vacation_db PostgreSQL 저장소.
|
||||
|
||||
- 드라이버: psycopg 3 (`psycopg[binary,pool]`) — expense/cupang 모듈과 동일 패턴.
|
||||
- 연결 정보: 환경변수 `VACATION_DB_URL`
|
||||
(예: postgresql://vacation_app:<pwd>@postgres-db:5432/vacation_db)
|
||||
- 스키마(테이블/인덱스/트리거/seed)는 앱이 만들지 않는다.
|
||||
`scripts/sql/vacation_db_init.sql` 을 superuser 가 사전 적용한다.
|
||||
앱 계정(vacation_app)은 SELECT/INSERT/UPDATE/DELETE 권한만 받는다.
|
||||
- 연결 풀은 lazy open — 부팅 시 DB 가 잠시 끊겨도 컨테이너가 죽지 않게.
|
||||
|
||||
휴가 일수는 서버에서 `store.compute_days` 로 재계산하여 저장한다.
|
||||
공휴일 집합은 vacation_holidays(is_red=TRUE)에서 읽어 계산에 넘긴다(holiday provider).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import date, datetime, timezone
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from psycopg.rows import dict_row
|
||||
from psycopg_pool import ConnectionPool
|
||||
|
||||
from .store import EDITABLE_STATUSES, STATUSES, VACATION_TYPES, compute_days, normalize_half
|
||||
|
||||
|
||||
class VacationDBStore:
|
||||
def __init__(self, dsn: str, *, min_size: int = 1, max_size: int = 5):
|
||||
self._pool = ConnectionPool(
|
||||
conninfo=dsn,
|
||||
min_size=min_size,
|
||||
max_size=max_size,
|
||||
kwargs={"row_factory": dict_row, "autocommit": True},
|
||||
open=False,
|
||||
)
|
||||
self._pool.open(wait=False)
|
||||
|
||||
def close(self) -> None:
|
||||
self._pool.close()
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 공휴일 (vacation_holidays) — holiday provider
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def list_holidays(self, *, year: int | None = None) -> list[dict[str, Any]]:
|
||||
clause = ""
|
||||
params: list[Any] = []
|
||||
if year:
|
||||
clause = "WHERE EXTRACT(YEAR FROM holiday_date) = %s"
|
||||
params.append(year)
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
f"SELECT * FROM vacation_holidays {clause} ORDER BY holiday_date ASC",
|
||||
params,
|
||||
).fetchall()
|
||||
return [self._holiday_serialize(r) for r in rows]
|
||||
|
||||
def red_holiday_set(self, *, date_from: str, date_to: str) -> set[str]:
|
||||
"""[date_from, date_to] 범위의 is_red=TRUE 공휴일 ISO 날짜 집합.
|
||||
|
||||
휴가일수 계산/달력 색상의 단일 진실 공급원.
|
||||
"""
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT holiday_date FROM vacation_holidays "
|
||||
"WHERE is_red = TRUE AND holiday_date BETWEEN %s AND %s",
|
||||
(date_from, date_to),
|
||||
).fetchall()
|
||||
out: set[str] = set()
|
||||
for r in rows:
|
||||
d = r["holiday_date"]
|
||||
out.add(d.isoformat() if isinstance(d, date) else str(d))
|
||||
return out
|
||||
|
||||
def upsert_holiday(
|
||||
self,
|
||||
*,
|
||||
holiday_date: str,
|
||||
name: str,
|
||||
kind: str = "public",
|
||||
is_red: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
hd = (holiday_date or "").strip()
|
||||
nm = (name or "").strip()
|
||||
if not hd or not nm:
|
||||
raise ValueError("공휴일 날짜와 이름은 필수입니다.")
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
INSERT INTO vacation_holidays (holiday_date, name, kind, is_red)
|
||||
VALUES (%s, %s, %s, %s)
|
||||
ON CONFLICT (holiday_date) DO UPDATE
|
||||
SET name = EXCLUDED.name,
|
||||
kind = EXCLUDED.kind,
|
||||
is_red = EXCLUDED.is_red
|
||||
RETURNING *
|
||||
""",
|
||||
(hd, nm, (kind or "public").strip(), bool(is_red)),
|
||||
).fetchone()
|
||||
return self._holiday_serialize(row)
|
||||
|
||||
def delete_holiday(self, *, holiday_id: int) -> None:
|
||||
with self._pool.connection() as conn:
|
||||
cur = conn.execute(
|
||||
"DELETE FROM vacation_holidays WHERE id = %s", (holiday_id,)
|
||||
)
|
||||
if cur.rowcount == 0:
|
||||
raise KeyError(holiday_id)
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 잔여 연차 (vacation_balances)
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def get_balance(self, *, user_email: str, year: int) -> dict[str, Any] | None:
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT * FROM vacation_balances WHERE user_email = %s AND year = %s",
|
||||
(user_email.lower().strip(), year),
|
||||
).fetchone()
|
||||
return self._balance_serialize(row) if row else None
|
||||
|
||||
def list_balances(self, *, year: int) -> list[dict[str, Any]]:
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM vacation_balances WHERE year = %s ORDER BY user_email ASC",
|
||||
(year,),
|
||||
).fetchall()
|
||||
return [self._balance_serialize(r) for r in rows]
|
||||
|
||||
def upsert_balance(
|
||||
self, *, user_email: str, year: int, total_days: float, memo: str = ""
|
||||
) -> dict[str, Any]:
|
||||
email = (user_email or "").lower().strip()
|
||||
if not email or "@" not in email:
|
||||
raise ValueError("올바른 이메일이 필요합니다.")
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
INSERT INTO vacation_balances (user_email, year, total_days, memo)
|
||||
VALUES (%s, %s, %s, %s)
|
||||
ON CONFLICT (user_email, year) DO UPDATE
|
||||
SET total_days = EXCLUDED.total_days,
|
||||
memo = EXCLUDED.memo
|
||||
RETURNING *
|
||||
""",
|
||||
(email, year, total_days, (memo or "").strip()),
|
||||
).fetchone()
|
||||
return self._balance_serialize(row)
|
||||
|
||||
def used_days(self, *, user_email: str, year: int) -> float:
|
||||
"""해당 연도 승인된 휴가의 합계 일수(사용 연차). start_date 연도 기준."""
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT COALESCE(SUM(days), 0) AS s FROM vacation_requests "
|
||||
"WHERE owner = %s AND status = '승인' "
|
||||
"AND EXTRACT(YEAR FROM start_date) = %s",
|
||||
(user_email.lower().strip(), year),
|
||||
).fetchone()
|
||||
return float(row["s"]) if row else 0.0
|
||||
|
||||
def balance_summary(self, *, user_email: str, year: int) -> dict[str, Any]:
|
||||
bal = self.get_balance(user_email=user_email, year=year)
|
||||
total = float(bal["total_days"]) if bal else 0.0
|
||||
used = self.used_days(user_email=user_email, year=year)
|
||||
return {
|
||||
"year": year,
|
||||
"total_days": total,
|
||||
"used_days": used,
|
||||
"remaining_days": round(total - used, 2),
|
||||
"memo": bal["memo"] if bal else "",
|
||||
}
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 휴가 신청 (vacation_requests)
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def list_for(self, owner: str) -> list[dict[str, Any]]:
|
||||
owner = owner.lower().strip()
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM vacation_requests WHERE owner = %s "
|
||||
"ORDER BY start_date DESC, created_at DESC",
|
||||
(owner,),
|
||||
).fetchall()
|
||||
return [self._req_serialize(r) for r in rows]
|
||||
|
||||
def list_overlapping(self, *, date_from: str, date_to: str) -> list[dict[str, Any]]:
|
||||
"""[date_from, date_to] 와 겹치는 모든 신청(달력 bar 표시용).
|
||||
|
||||
취소 포함(흐리게 표시). 범위가 한 칸이라도 겹치면 포함.
|
||||
"""
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM vacation_requests "
|
||||
"WHERE start_date <= %s AND end_date >= %s "
|
||||
"ORDER BY start_date ASC, owner ASC",
|
||||
(date_to, date_from),
|
||||
).fetchall()
|
||||
return [self._req_serialize(r) for r in rows]
|
||||
|
||||
def get_request(self, *, request_id: str) -> dict[str, Any] | None:
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT * FROM vacation_requests WHERE id = %s", (request_id,)
|
||||
).fetchone()
|
||||
return self._req_serialize(row) if row else None
|
||||
|
||||
def create_request(
|
||||
self, *, owner: str, owner_name: str, payload: dict[str, Any], status: str = "작성중"
|
||||
) -> dict[str, Any]:
|
||||
h = self._normalize_payload(payload)
|
||||
if status not in STATUSES:
|
||||
status = "작성중"
|
||||
days = self._calc_days(h)
|
||||
req_id = uuid.uuid4().hex[:12]
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
INSERT INTO vacation_requests
|
||||
(id, owner, owner_name, vacation_type, start_date, end_date,
|
||||
start_half, end_half, days, reason, status)
|
||||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||||
RETURNING *
|
||||
""",
|
||||
(
|
||||
req_id,
|
||||
owner.lower().strip(),
|
||||
(owner_name or "").strip(),
|
||||
h["vacation_type"],
|
||||
h["start_date"],
|
||||
h["end_date"],
|
||||
h["start_half"],
|
||||
h["end_half"],
|
||||
days,
|
||||
h["reason"],
|
||||
status,
|
||||
),
|
||||
).fetchone()
|
||||
return self._req_serialize(row)
|
||||
|
||||
def update_request(
|
||||
self, *, request_id: str, owner: str, payload: dict[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
"""작성중/반려 상태에서만 본인이 수정 가능."""
|
||||
owner = owner.lower().strip()
|
||||
current = self.get_request(request_id=request_id)
|
||||
if not current:
|
||||
raise KeyError(request_id)
|
||||
if current["owner"] != owner:
|
||||
raise PermissionError("본인 신청만 수정할 수 있습니다.")
|
||||
if current["status"] not in EDITABLE_STATUSES:
|
||||
raise ValueError(f"{current['status']} 상태에서는 수정할 수 없습니다.")
|
||||
h = self._normalize_payload(payload)
|
||||
days = self._calc_days(h)
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
UPDATE vacation_requests
|
||||
SET vacation_type = %s, start_date = %s, end_date = %s,
|
||||
start_half = %s, end_half = %s, days = %s, reason = %s,
|
||||
status = '작성중', reject_reason = '',
|
||||
approver_email = NULL, decided_at = NULL
|
||||
WHERE id = %s AND owner = %s
|
||||
RETURNING *
|
||||
""",
|
||||
(
|
||||
h["vacation_type"],
|
||||
h["start_date"],
|
||||
h["end_date"],
|
||||
h["start_half"],
|
||||
h["end_half"],
|
||||
days,
|
||||
h["reason"],
|
||||
request_id,
|
||||
owner,
|
||||
),
|
||||
).fetchone()
|
||||
if not row:
|
||||
raise KeyError(request_id)
|
||||
return self._req_serialize(row)
|
||||
|
||||
# ── 워크플로 ──
|
||||
def submit(self, *, request_id: str, owner: str) -> dict[str, Any]:
|
||||
"""작성중/반려 → 제출 (본인)."""
|
||||
owner = owner.lower().strip()
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
UPDATE vacation_requests
|
||||
SET status = '제출', reject_reason = '',
|
||||
approver_email = NULL, decided_at = NULL
|
||||
WHERE id = %s AND owner = %s AND status IN ('작성중', '반려')
|
||||
RETURNING *
|
||||
""",
|
||||
(request_id, owner),
|
||||
).fetchone()
|
||||
if not row:
|
||||
raise ValueError("작성중/반려 상태에서만 제출할 수 있습니다.")
|
||||
return self._req_serialize(row)
|
||||
|
||||
def approve(self, *, request_id: str, approver_email: str) -> dict[str, Any]:
|
||||
approver_email = approver_email.lower().strip()
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
UPDATE vacation_requests
|
||||
SET status = '승인', approver_email = %s, decided_at = now(),
|
||||
reject_reason = ''
|
||||
WHERE id = %s AND status = '제출'
|
||||
RETURNING *
|
||||
""",
|
||||
(approver_email, request_id),
|
||||
).fetchone()
|
||||
if not row:
|
||||
raise ValueError("제출 상태가 아니거나 신청이 없습니다.")
|
||||
return self._req_serialize(row)
|
||||
|
||||
def reject(
|
||||
self, *, request_id: str, approver_email: str, reason: str
|
||||
) -> dict[str, Any]:
|
||||
if not (reason or "").strip():
|
||||
raise ValueError("반려 사유는 필수입니다.")
|
||||
approver_email = approver_email.lower().strip()
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"""
|
||||
UPDATE vacation_requests
|
||||
SET status = '반려', approver_email = %s, decided_at = now(),
|
||||
reject_reason = %s
|
||||
WHERE id = %s AND status = '제출'
|
||||
RETURNING *
|
||||
""",
|
||||
(approver_email, reason.strip(), request_id),
|
||||
).fetchone()
|
||||
if not row:
|
||||
raise ValueError("제출 상태가 아니거나 신청이 없습니다.")
|
||||
return self._req_serialize(row)
|
||||
|
||||
def cancel(self, *, request_id: str, user_email: str, is_admin: bool) -> dict[str, Any]:
|
||||
"""soft delete — status='취소'. owner 또는 admin 만."""
|
||||
user_email = user_email.lower().strip()
|
||||
current = self.get_request(request_id=request_id)
|
||||
if not current:
|
||||
raise KeyError(request_id)
|
||||
if not is_admin and current["owner"] != user_email:
|
||||
raise PermissionError("본인 신청만 취소할 수 있습니다.")
|
||||
with self._pool.connection() as conn:
|
||||
row = conn.execute(
|
||||
"UPDATE vacation_requests SET status = '취소' WHERE id = %s RETURNING *",
|
||||
(request_id,),
|
||||
).fetchone()
|
||||
return self._req_serialize(row)
|
||||
|
||||
def list_pending(self) -> list[dict[str, Any]]:
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM vacation_requests WHERE status = '제출' "
|
||||
"ORDER BY start_date ASC, created_at ASC"
|
||||
).fetchall()
|
||||
return [self._req_serialize(r) for r in rows]
|
||||
|
||||
def list_for_export(
|
||||
self, *, date_from: str | None = None, date_to: str | None = None
|
||||
) -> list[dict[str, Any]]:
|
||||
clauses: list[str] = []
|
||||
params: list[Any] = []
|
||||
if date_from:
|
||||
clauses.append("end_date >= %s")
|
||||
params.append(date_from)
|
||||
if date_to:
|
||||
clauses.append("start_date <= %s")
|
||||
params.append(date_to)
|
||||
where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
|
||||
with self._pool.connection() as conn:
|
||||
rows = conn.execute(
|
||||
f"SELECT * FROM vacation_requests {where} "
|
||||
"ORDER BY start_date ASC, owner ASC",
|
||||
params,
|
||||
).fetchall()
|
||||
return [self._req_serialize(r) for r in rows]
|
||||
|
||||
# ════════════════════════════════════════════════════════════
|
||||
# 정규화 / 계산 / 직렬화 helpers
|
||||
# ════════════════════════════════════════════════════════════
|
||||
def _calc_days(self, h: dict[str, Any]) -> float:
|
||||
holidays = self.red_holiday_set(
|
||||
date_from=h["start_date"], date_to=h["end_date"]
|
||||
)
|
||||
return compute_days(
|
||||
start_date=date.fromisoformat(h["start_date"]),
|
||||
end_date=date.fromisoformat(h["end_date"]),
|
||||
start_half=h["start_half"],
|
||||
end_half=h["end_half"],
|
||||
vacation_type=h["vacation_type"],
|
||||
holidays=holidays,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _normalize_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
vtype = str(payload.get("vacation_type") or "연차").strip()
|
||||
if vtype not in VACATION_TYPES:
|
||||
vtype = "연차"
|
||||
start_date = str(payload.get("start_date") or "").strip()
|
||||
end_date = str(payload.get("end_date") or "").strip()
|
||||
if not start_date:
|
||||
raise ValueError("시작일은 필수입니다.")
|
||||
if not end_date:
|
||||
end_date = start_date
|
||||
# 날짜 형식/순서 검증
|
||||
try:
|
||||
sd = date.fromisoformat(start_date)
|
||||
ed = date.fromisoformat(end_date)
|
||||
except ValueError:
|
||||
raise ValueError("날짜 형식이 올바르지 않습니다 (YYYY-MM-DD).")
|
||||
if ed < sd:
|
||||
raise ValueError("종료일이 시작일보다 빠를 수 없습니다.")
|
||||
start_half = normalize_half(payload.get("start_half"))
|
||||
end_half = normalize_half(payload.get("end_half"))
|
||||
# 반차 종류면 단일 일자로 강제
|
||||
if vtype in ("오전반차", "오후반차"):
|
||||
end_date = start_date
|
||||
start_half = "am" if vtype == "오전반차" else "pm"
|
||||
end_half = start_half
|
||||
return {
|
||||
"vacation_type": vtype,
|
||||
"start_date": start_date,
|
||||
"end_date": end_date,
|
||||
"start_half": start_half,
|
||||
"end_half": end_half,
|
||||
"reason": str(payload.get("reason") or "").strip(),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _num(v: Any) -> float:
|
||||
if isinstance(v, Decimal):
|
||||
return float(v)
|
||||
try:
|
||||
return float(v)
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
@classmethod
|
||||
def _req_serialize(cls, row: dict[str, Any] | None) -> dict[str, Any] | None:
|
||||
if row is None:
|
||||
return None
|
||||
out = dict(row)
|
||||
for k in ("start_date", "end_date"):
|
||||
v = out.get(k)
|
||||
if isinstance(v, date):
|
||||
out[k] = v.isoformat()
|
||||
for k in ("created_at", "updated_at", "decided_at"):
|
||||
v = out.get(k)
|
||||
if isinstance(v, datetime):
|
||||
out[k] = v.astimezone(timezone.utc).isoformat(timespec="seconds")
|
||||
out["days"] = cls._num(out.get("days", 0))
|
||||
return out
|
||||
|
||||
@staticmethod
|
||||
def _holiday_serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
|
||||
if row is None:
|
||||
return None
|
||||
out = dict(row)
|
||||
out["id"] = int(out["id"])
|
||||
v = out.get("holiday_date")
|
||||
if isinstance(v, date):
|
||||
out["holiday_date"] = v.isoformat()
|
||||
out["is_red"] = bool(out.get("is_red", True))
|
||||
for k in ("created_at", "updated_at"):
|
||||
v = out.get(k)
|
||||
if isinstance(v, datetime):
|
||||
out[k] = v.astimezone(timezone.utc).isoformat(timespec="seconds")
|
||||
return out
|
||||
|
||||
@classmethod
|
||||
def _balance_serialize(cls, row: dict[str, Any] | None) -> dict[str, Any] | None:
|
||||
if row is None:
|
||||
return None
|
||||
out = dict(row)
|
||||
out["id"] = int(out["id"])
|
||||
out["year"] = int(out["year"])
|
||||
out["total_days"] = cls._num(out.get("total_days", 0))
|
||||
out["used_days"] = cls._num(out.get("used_days", 0))
|
||||
for k in ("created_at", "updated_at"):
|
||||
v = out.get(k)
|
||||
if isinstance(v, datetime):
|
||||
out[k] = v.astimezone(timezone.utc).isoformat(timespec="seconds")
|
||||
return out
|
||||
Reference in New Issue
Block a user