king
c192d61c71
- project_db 신규(projects/members/stages/tasks/comments/activity, project_app CRUD) - 프로젝트/서브프로젝트(self-FK), 업무·진행단계(칸반), 멤버 배정 - 메인 뷰 달력(FullCalendar)/타임라인(vis-timeline) 토글 + 보드/리스트 - 진입은 로그인 회사 직원 전원, 생성/배정은 관리자 전용 - 업무 배정·완료 시 관리자 메일 알림(app/mail.py, SMTP_* env, 미설정 시 skip) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
520 lines
23 KiB
Python
520 lines
23 KiB
Python
"""project_db PostgreSQL 저장소.
|
|
|
|
- 드라이버: psycopg 3 (`psycopg[binary,pool]`) — 다른 모듈과 동일 패턴.
|
|
- 연결 정보: 환경변수 `PROJECT_DB_URL`
|
|
(예: postgresql://project_app:<pwd>@postgres-db:5432/project_db)
|
|
- 스키마는 앱이 만들지 않는다. `scripts/sql/project_db_init.sql` 을 superuser 가
|
|
사전 적용한다. 앱 계정(project_app)은 CRUD 권한만 받는다.
|
|
- 연결 풀은 lazy open.
|
|
|
|
순수 검증/상수는 store.py 에 있고, 여기서는 DB I/O 와 직렬화만 담당한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date, datetime
|
|
from typing import Any
|
|
|
|
from psycopg.rows import dict_row
|
|
from psycopg_pool import ConnectionPool
|
|
|
|
from app.timezone import KST, now_kst
|
|
|
|
from . import store
|
|
|
|
logger = logging.getLogger("project.db")
|
|
|
|
|
|
class ProjectStore:
|
|
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()
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 프로젝트 / 서브프로젝트
|
|
# ════════════════════════════════════════════════════════════
|
|
def list_projects(
|
|
self, *, include_archived: bool = False, member_email: str | None = None
|
|
) -> list[dict[str, Any]]:
|
|
"""전체 프로젝트(서브 포함) 평면 목록. member_email 지정 시 그 사용자가
|
|
멤버이거나 owner 인 프로젝트만(관리자는 라우터에서 None 으로 전체 조회).
|
|
"""
|
|
clauses: list[str] = []
|
|
params: list[Any] = []
|
|
if not include_archived:
|
|
clauses.append("p.status <> 'archived'")
|
|
if member_email:
|
|
clauses.append(
|
|
"(p.owner_email = %s OR EXISTS ("
|
|
" SELECT 1 FROM project_members m "
|
|
" WHERE m.project_id = p.id AND m.user_email = %s))"
|
|
)
|
|
params.extend([member_email, member_email])
|
|
where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
|
|
with self._pool.connection() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT p.* FROM projects p {where} "
|
|
"ORDER BY p.parent_id NULLS FIRST, p.sort_order ASC, p.id ASC",
|
|
params,
|
|
).fetchall()
|
|
return [self._serialize(r) for r in rows]
|
|
|
|
def get_project(self, *, project_id: int) -> dict[str, Any] | None:
|
|
with self._pool.connection() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM projects WHERE id = %s", (project_id,)
|
|
).fetchone()
|
|
return self._serialize(row) if row else None
|
|
|
|
def create_project(
|
|
self,
|
|
*,
|
|
name: str,
|
|
parent_id: int | None = None,
|
|
description: str = "",
|
|
color: str = "",
|
|
owner_email: str = "",
|
|
start_date: str | None = None,
|
|
due_date: str | None = None,
|
|
created_by: str = "",
|
|
seed_stages: bool = True,
|
|
) -> dict[str, Any]:
|
|
nm = store.validate_project_name(name)
|
|
col = store.validate_color(color)
|
|
sd = store.validate_date(start_date)
|
|
dd = store.validate_date(due_date)
|
|
with self._pool.connection() as conn:
|
|
with conn.transaction():
|
|
if parent_id is not None:
|
|
parent = conn.execute(
|
|
"SELECT id FROM projects WHERE id = %s", (parent_id,)
|
|
).fetchone()
|
|
if not parent:
|
|
raise KeyError(f"상위 프로젝트를 찾을 수 없습니다: {parent_id}")
|
|
row = conn.execute(
|
|
"""
|
|
INSERT INTO projects
|
|
(parent_id, name, description, color, owner_email,
|
|
start_date, due_date, created_by)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s)
|
|
RETURNING *
|
|
""",
|
|
(
|
|
parent_id,
|
|
nm,
|
|
store.norm_str(description),
|
|
col,
|
|
store.norm_str(owner_email, lower=True),
|
|
sd,
|
|
dd,
|
|
store.norm_str(created_by, lower=True),
|
|
),
|
|
).fetchone()
|
|
pid = row["id"]
|
|
if seed_stages:
|
|
for i, (stage_name, is_done) in enumerate(store.DEFAULT_STAGES):
|
|
conn.execute(
|
|
"INSERT INTO project_stages "
|
|
"(project_id, name, sort_order, is_done_stage) "
|
|
"VALUES (%s,%s,%s,%s)",
|
|
(pid, stage_name, i, is_done),
|
|
)
|
|
self._log(conn, project_id=pid, actor=created_by,
|
|
action=store.ACTION_CREATED, detail=nm)
|
|
return self._serialize(row)
|
|
|
|
def update_project(self, *, project_id: int, fields: dict[str, Any]) -> dict[str, Any]:
|
|
allowed = {
|
|
"name": lambda v: store.validate_project_name(v),
|
|
"description": lambda v: store.norm_str(v),
|
|
"color": lambda v: store.validate_color(v),
|
|
"owner_email": lambda v: store.norm_str(v, lower=True),
|
|
"start_date": lambda v: store.validate_date(v),
|
|
"due_date": lambda v: store.validate_date(v),
|
|
"status": lambda v: (store.norm_str(v, lower=True)
|
|
if store.norm_str(v, lower=True) in store.PROJECT_STATUSES
|
|
else "active"),
|
|
}
|
|
sets: list[str] = []
|
|
params: list[Any] = []
|
|
for key, fn in allowed.items():
|
|
if key in fields:
|
|
sets.append(f"{key} = %s")
|
|
params.append(fn(fields[key]))
|
|
if not sets:
|
|
existing = self.get_project(project_id=project_id)
|
|
if existing is None:
|
|
raise KeyError(project_id)
|
|
return existing
|
|
params.append(project_id)
|
|
with self._pool.connection() as conn:
|
|
row = conn.execute(
|
|
f"UPDATE projects SET {', '.join(sets)} WHERE id = %s RETURNING *",
|
|
params,
|
|
).fetchone()
|
|
if not row:
|
|
raise KeyError(project_id)
|
|
return self._serialize(row)
|
|
|
|
def delete_project(self, *, project_id: int) -> None:
|
|
"""프로젝트(서브프로젝트·단계·업무 CASCADE) 삭제."""
|
|
with self._pool.connection() as conn:
|
|
cur = conn.execute("DELETE FROM projects WHERE id = %s", (project_id,))
|
|
if cur.rowcount == 0:
|
|
raise KeyError(project_id)
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 멤버
|
|
# ════════════════════════════════════════════════════════════
|
|
def list_members(self, *, project_id: int) -> list[dict[str, Any]]:
|
|
with self._pool.connection() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM project_members WHERE project_id = %s "
|
|
"ORDER BY role DESC, user_email ASC",
|
|
(project_id,),
|
|
).fetchall()
|
|
return [self._serialize(r) for r in rows]
|
|
|
|
def is_member(self, *, project_id: int, user_email: str) -> bool:
|
|
email = store.norm_str(user_email, lower=True)
|
|
with self._pool.connection() as conn:
|
|
row = conn.execute(
|
|
"SELECT 1 FROM projects p "
|
|
"LEFT JOIN project_members m "
|
|
" ON m.project_id = p.id AND m.user_email = %s "
|
|
"WHERE p.id = %s AND (p.owner_email = %s OR m.id IS NOT NULL) LIMIT 1",
|
|
(email, project_id, email),
|
|
).fetchone()
|
|
return bool(row)
|
|
|
|
def add_member(
|
|
self, *, project_id: int, user_email: str, user_name: str = "", role: str = "member"
|
|
) -> dict[str, Any]:
|
|
email = store.norm_str(user_email, lower=True)
|
|
if not email or "@" not in email:
|
|
raise ValueError("올바른 이메일이 아닙니다.")
|
|
rl = store.validate_role(role)
|
|
with self._pool.connection() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
INSERT INTO project_members (project_id, user_email, user_name, role)
|
|
VALUES (%s,%s,%s,%s)
|
|
ON CONFLICT (project_id, user_email) DO UPDATE
|
|
SET role = EXCLUDED.role, user_name = EXCLUDED.user_name
|
|
RETURNING *
|
|
""",
|
|
(project_id, email, store.norm_str(user_name), rl),
|
|
).fetchone()
|
|
return self._serialize(row)
|
|
|
|
def remove_member(self, *, project_id: int, user_email: str) -> None:
|
|
email = store.norm_str(user_email, lower=True)
|
|
with self._pool.connection() as conn:
|
|
conn.execute(
|
|
"DELETE FROM project_members WHERE project_id = %s AND user_email = %s",
|
|
(project_id, email),
|
|
)
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 단계 (stages)
|
|
# ════════════════════════════════════════════════════════════
|
|
def list_stages(self, *, project_id: int) -> list[dict[str, Any]]:
|
|
with self._pool.connection() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM project_stages WHERE project_id = %s "
|
|
"ORDER BY sort_order ASC, id ASC",
|
|
(project_id,),
|
|
).fetchall()
|
|
return [self._serialize(r) for r in rows]
|
|
|
|
def add_stage(
|
|
self, *, project_id: int, name: str, is_done_stage: bool = False
|
|
) -> dict[str, Any]:
|
|
nm = store.norm_str(name)
|
|
if not nm:
|
|
raise ValueError("단계 이름은 필수입니다.")
|
|
with self._pool.connection() as conn:
|
|
nxt = conn.execute(
|
|
"SELECT COALESCE(MAX(sort_order), -1) + 1 AS n "
|
|
"FROM project_stages WHERE project_id = %s",
|
|
(project_id,),
|
|
).fetchone()
|
|
row = conn.execute(
|
|
"INSERT INTO project_stages (project_id, name, sort_order, is_done_stage) "
|
|
"VALUES (%s,%s,%s,%s) RETURNING *",
|
|
(project_id, nm, int(nxt["n"]), bool(is_done_stage)),
|
|
).fetchone()
|
|
return self._serialize(row)
|
|
|
|
def delete_stage(self, *, stage_id: int) -> None:
|
|
with self._pool.connection() as conn:
|
|
cur = conn.execute("DELETE FROM project_stages WHERE id = %s", (stage_id,))
|
|
if cur.rowcount == 0:
|
|
raise KeyError(stage_id)
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 업무 (tasks)
|
|
# ════════════════════════════════════════════════════════════
|
|
def list_tasks(
|
|
self,
|
|
*,
|
|
project_id: int | None = None,
|
|
assignee_email: str | None = None,
|
|
include_subprojects: bool = False,
|
|
limit: int = 1000,
|
|
) -> list[dict[str, Any]]:
|
|
clauses: list[str] = []
|
|
params: list[Any] = []
|
|
if project_id is not None:
|
|
if include_subprojects:
|
|
clauses.append(
|
|
"t.project_id IN ("
|
|
" WITH RECURSIVE tree AS ("
|
|
" SELECT id FROM projects WHERE id = %s "
|
|
" UNION ALL "
|
|
" SELECT p.id FROM projects p JOIN tree ON p.parent_id = tree.id"
|
|
" ) SELECT id FROM tree)"
|
|
)
|
|
params.append(project_id)
|
|
else:
|
|
clauses.append("t.project_id = %s")
|
|
params.append(project_id)
|
|
if assignee_email:
|
|
clauses.append("t.assignee_email = %s")
|
|
params.append(store.norm_str(assignee_email, lower=True))
|
|
where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
|
|
params.append(int(limit))
|
|
with self._pool.connection() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT t.*, p.name AS project_name, p.color AS project_color, "
|
|
f" s.name AS stage_name, s.is_done_stage "
|
|
f"FROM tasks t "
|
|
f"JOIN projects p ON p.id = t.project_id "
|
|
f"LEFT JOIN project_stages s ON s.id = t.stage_id "
|
|
f"{where} "
|
|
"ORDER BY t.sort_order ASC, t.id ASC LIMIT %s",
|
|
params,
|
|
).fetchall()
|
|
return [self._serialize(r) for r in rows]
|
|
|
|
def get_task(self, *, task_id: int) -> dict[str, Any] | None:
|
|
with self._pool.connection() as conn:
|
|
row = conn.execute(
|
|
"SELECT t.*, p.name AS project_name, p.color AS project_color, "
|
|
" s.name AS stage_name, s.is_done_stage "
|
|
"FROM tasks t JOIN projects p ON p.id = t.project_id "
|
|
"LEFT JOIN project_stages s ON s.id = t.stage_id "
|
|
"WHERE t.id = %s",
|
|
(task_id,),
|
|
).fetchone()
|
|
return self._serialize(row) if row else None
|
|
|
|
def create_task(
|
|
self,
|
|
*,
|
|
project_id: int,
|
|
title: str,
|
|
stage_id: int | None = None,
|
|
description: str = "",
|
|
assignee_email: str = "",
|
|
assignee_name: str = "",
|
|
priority: str = "normal",
|
|
start_date: str | None = None,
|
|
due_date: str | None = None,
|
|
created_by: str = "",
|
|
) -> dict[str, Any]:
|
|
ttl = store.validate_task_title(title)
|
|
pr = store.validate_priority(priority)
|
|
sd = store.validate_date(start_date)
|
|
dd = store.validate_date(due_date)
|
|
assignee = store.norm_str(assignee_email, lower=True)
|
|
with self._pool.connection() as conn:
|
|
with conn.transaction():
|
|
# stage 미지정 시 프로젝트의 첫 단계로
|
|
if stage_id is None:
|
|
st = conn.execute(
|
|
"SELECT id FROM project_stages WHERE project_id = %s "
|
|
"ORDER BY sort_order ASC, id ASC LIMIT 1",
|
|
(project_id,),
|
|
).fetchone()
|
|
stage_id = st["id"] if st else None
|
|
nxt = conn.execute(
|
|
"SELECT COALESCE(MAX(sort_order), -1) + 1 AS n FROM tasks "
|
|
"WHERE project_id = %s",
|
|
(project_id,),
|
|
).fetchone()
|
|
row = conn.execute(
|
|
"""
|
|
INSERT INTO tasks
|
|
(project_id, stage_id, title, description, assignee_email,
|
|
assignee_name, priority, start_date, due_date, sort_order, created_by)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
|
RETURNING *
|
|
""",
|
|
(
|
|
project_id, stage_id, ttl, store.norm_str(description),
|
|
assignee, store.norm_str(assignee_name), pr, sd, dd,
|
|
int(nxt["n"]), store.norm_str(created_by, lower=True),
|
|
),
|
|
).fetchone()
|
|
if assignee:
|
|
self._log(conn, project_id=project_id, task_id=row["id"],
|
|
actor=created_by, action=store.ACTION_ASSIGNED,
|
|
detail=f"{ttl} → {assignee}")
|
|
return self.get_task(task_id=row["id"])
|
|
|
|
def update_task(
|
|
self, *, task_id: int, fields: dict[str, Any], actor: str = ""
|
|
) -> dict[str, Any]:
|
|
"""업무 수정. 반환에 알림 힌트를 담는다(라우터가 메일 발송 판단).
|
|
|
|
반환 dict 에 추가되는 키:
|
|
_newly_assigned: 새 담당자 이메일 or None
|
|
_newly_completed: 완료로 전환됐으면 True
|
|
"""
|
|
prev = self.get_task(task_id=task_id)
|
|
if prev is None:
|
|
raise KeyError(task_id)
|
|
|
|
sets: list[str] = []
|
|
params: list[Any] = []
|
|
newly_assigned: str | None = None
|
|
newly_completed = False
|
|
|
|
if "title" in fields:
|
|
sets.append("title = %s")
|
|
params.append(store.validate_task_title(fields["title"]))
|
|
if "description" in fields:
|
|
sets.append("description = %s")
|
|
params.append(store.norm_str(fields["description"]))
|
|
if "priority" in fields:
|
|
sets.append("priority = %s")
|
|
params.append(store.validate_priority(fields["priority"]))
|
|
if "start_date" in fields:
|
|
sets.append("start_date = %s")
|
|
params.append(store.validate_date(fields["start_date"]))
|
|
if "due_date" in fields:
|
|
sets.append("due_date = %s")
|
|
params.append(store.validate_date(fields["due_date"]))
|
|
if "assignee_email" in fields:
|
|
new_assignee = store.norm_str(fields["assignee_email"], lower=True)
|
|
sets.append("assignee_email = %s")
|
|
params.append(new_assignee)
|
|
if "assignee_name" in fields:
|
|
sets.append("assignee_name = %s")
|
|
params.append(store.norm_str(fields["assignee_name"]))
|
|
if new_assignee and new_assignee != (prev.get("assignee_email") or ""):
|
|
newly_assigned = new_assignee
|
|
elif "assignee_name" in fields:
|
|
sets.append("assignee_name = %s")
|
|
params.append(store.norm_str(fields["assignee_name"]))
|
|
|
|
# 단계 변경 → 완료단계 진입 시 completed_at 자동 세팅
|
|
if "stage_id" in fields:
|
|
new_stage = fields["stage_id"]
|
|
sets.append("stage_id = %s")
|
|
params.append(new_stage)
|
|
done = self._stage_is_done(new_stage)
|
|
if done and not prev.get("completed_at"):
|
|
sets.append("completed_at = %s")
|
|
params.append(now_kst())
|
|
newly_completed = True
|
|
elif not done and prev.get("completed_at"):
|
|
sets.append("completed_at = NULL")
|
|
|
|
if not sets:
|
|
return {**prev, "_newly_assigned": None, "_newly_completed": False}
|
|
|
|
params.append(task_id)
|
|
with self._pool.connection() as conn:
|
|
with conn.transaction():
|
|
conn.execute(
|
|
f"UPDATE tasks SET {', '.join(sets)} WHERE id = %s", params
|
|
)
|
|
if newly_assigned:
|
|
self._log(conn, project_id=prev["project_id"], task_id=task_id,
|
|
actor=actor, action=store.ACTION_ASSIGNED,
|
|
detail=f"{prev['title']} → {newly_assigned}")
|
|
if newly_completed:
|
|
self._log(conn, project_id=prev["project_id"], task_id=task_id,
|
|
actor=actor, action=store.ACTION_COMPLETED,
|
|
detail=prev["title"])
|
|
out = self.get_task(task_id=task_id)
|
|
out["_newly_assigned"] = newly_assigned
|
|
out["_newly_completed"] = newly_completed
|
|
return out
|
|
|
|
def delete_task(self, *, task_id: int) -> None:
|
|
with self._pool.connection() as conn:
|
|
cur = conn.execute("DELETE FROM tasks WHERE id = %s", (task_id,))
|
|
if cur.rowcount == 0:
|
|
raise KeyError(task_id)
|
|
|
|
def _stage_is_done(self, stage_id: Any) -> bool:
|
|
if stage_id is None:
|
|
return False
|
|
with self._pool.connection() as conn:
|
|
row = conn.execute(
|
|
"SELECT is_done_stage FROM project_stages WHERE id = %s", (stage_id,)
|
|
).fetchone()
|
|
return bool(row and row["is_done_stage"])
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 활동 이력
|
|
# ════════════════════════════════════════════════════════════
|
|
def list_activity(self, *, project_id: int, limit: int = 100) -> list[dict[str, Any]]:
|
|
with self._pool.connection() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM project_activity WHERE project_id = %s "
|
|
"ORDER BY created_at DESC, id DESC LIMIT %s",
|
|
(project_id, int(limit)),
|
|
).fetchall()
|
|
return [self._serialize(r) for r in rows]
|
|
|
|
@staticmethod
|
|
def _log(
|
|
conn: Any, *, project_id: int | None = None, task_id: int | None = None,
|
|
actor: str = "", action: str = "", detail: str = "",
|
|
) -> None:
|
|
conn.execute(
|
|
"INSERT INTO project_activity "
|
|
"(project_id, task_id, actor_email, action, detail) "
|
|
"VALUES (%s,%s,%s,%s,%s)",
|
|
(project_id, task_id, store.norm_str(actor, lower=True), action,
|
|
store.norm_str(detail)),
|
|
)
|
|
|
|
# ════════════════════════════════════════════════════════════
|
|
# 직렬화
|
|
# ════════════════════════════════════════════════════════════
|
|
@staticmethod
|
|
def _serialize(row: dict[str, Any] | None) -> dict[str, Any] | None:
|
|
if row is None:
|
|
return None
|
|
out = dict(row)
|
|
for k, v in list(out.items()):
|
|
if isinstance(v, datetime):
|
|
out[k] = v.astimezone(KST).isoformat(timespec="seconds")
|
|
elif isinstance(v, date):
|
|
out[k] = v.isoformat()
|
|
for k in ("id", "parent_id", "project_id", "stage_id", "task_id", "sort_order"):
|
|
if k in out and out[k] is not None:
|
|
try:
|
|
out[k] = int(out[k])
|
|
except (TypeError, ValueError):
|
|
pass
|
|
for k in ("is_done_stage",):
|
|
if k in out and out[k] is not None:
|
|
out[k] = bool(out[k])
|
|
return out
|