146 lines
4.4 KiB
Python
146 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, time, timedelta
|
|
from typing import Callable
|
|
|
|
from fastapi import HTTPException
|
|
|
|
|
|
def fetch_members(get_conn) -> list[dict[str, object]]:
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, name, employee_id, company, rank, role, department, grp, division, team, cell,
|
|
work_status, work_time, phone, email, seat_label, photo_url,
|
|
sort_order, created_at, updated_at
|
|
FROM members
|
|
ORDER BY sort_order ASC, id ASC
|
|
"""
|
|
)
|
|
return cur.fetchall()
|
|
|
|
|
|
def fetch_history_revision(cur, revision_id: int) -> dict[str, object] | None:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, revision_label, created_at, note
|
|
FROM history_revisions
|
|
WHERE scope = 'organization'
|
|
AND id = %s
|
|
""",
|
|
(revision_id,),
|
|
)
|
|
return cur.fetchone()
|
|
|
|
|
|
def fetch_history_revisions(
|
|
cur,
|
|
*,
|
|
app_timezone,
|
|
day: date | None = None,
|
|
limit: int = 100,
|
|
) -> list[dict[str, object]]:
|
|
safe_limit = max(1, min(int(limit), 500))
|
|
if day is None:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, revision_label, created_at, note
|
|
FROM history_revisions
|
|
WHERE scope = 'organization'
|
|
ORDER BY created_at DESC, id DESC
|
|
LIMIT %s
|
|
""",
|
|
(safe_limit,),
|
|
)
|
|
return cur.fetchall()
|
|
|
|
day_start = datetime.combine(day, time.min, tzinfo=app_timezone)
|
|
day_end = day_start + timedelta(days=1)
|
|
cur.execute(
|
|
"""
|
|
SELECT id, revision_label, created_at, note
|
|
FROM history_revisions
|
|
WHERE scope = 'organization'
|
|
AND created_at >= %s
|
|
AND created_at < %s
|
|
ORDER BY created_at ASC, id ASC
|
|
LIMIT %s
|
|
""",
|
|
(day_start, day_end, safe_limit),
|
|
)
|
|
return cur.fetchall()
|
|
|
|
|
|
def fetch_history_revision_created_at(cur, revision_id: int) -> datetime:
|
|
revision = fetch_history_revision(cur, revision_id)
|
|
if revision is None:
|
|
raise HTTPException(status_code=404, detail="History revision not found.")
|
|
return revision["created_at"]
|
|
|
|
|
|
def fetch_current_member_state(cur) -> dict[int, dict[str, object]]:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, name, employee_id, company, rank, role, department, grp, division, team, cell,
|
|
work_status, work_time, phone, email, seat_label, photo_url, sort_order,
|
|
created_at, updated_at
|
|
FROM members
|
|
"""
|
|
)
|
|
return {int(row["id"]): row for row in cur.fetchall()}
|
|
|
|
|
|
def fetch_current_seat_assignments(cur) -> dict[int, dict[str, object]]:
|
|
cur.execute(
|
|
"""
|
|
SELECT member_id, seat_map_id, seat_slot_id, seat_label, updated_at
|
|
FROM seat_positions
|
|
"""
|
|
)
|
|
return {int(row["member_id"]): row for row in cur.fetchall()}
|
|
|
|
|
|
def fetch_members_as_of(cur, as_of: datetime) -> list[dict[str, object]]:
|
|
cur.execute(
|
|
"""
|
|
SELECT mv.member_id AS id,
|
|
mv.name,
|
|
COALESCE(m.employee_id, '') AS employee_id,
|
|
mv.company,
|
|
mv.rank,
|
|
mv.role,
|
|
mv.department,
|
|
mv.grp,
|
|
mv.division,
|
|
mv.team,
|
|
mv.cell,
|
|
mv.work_status,
|
|
mv.work_time,
|
|
mv.phone,
|
|
mv.email,
|
|
COALESCE(sav.seat_label, '') AS seat_label,
|
|
mv.photo_url,
|
|
COALESCE(m.sort_order, 2147483647) AS sort_order,
|
|
mv.created_at,
|
|
mv.valid_from AS updated_at,
|
|
mv.valid_to AS history_valid_to,
|
|
mv.revision_no,
|
|
hr.created_at AS revision_created_at
|
|
FROM member_versions mv
|
|
LEFT JOIN members m
|
|
ON m.id = mv.member_id
|
|
LEFT JOIN history_revisions hr
|
|
ON hr.id = mv.revision_no
|
|
LEFT JOIN seat_assignment_versions sav
|
|
ON sav.member_id = mv.member_id
|
|
AND sav.valid_from <= %s
|
|
AND (sav.valid_to IS NULL OR sav.valid_to > %s)
|
|
WHERE mv.valid_from <= %s
|
|
AND (mv.valid_to IS NULL OR mv.valid_to > %s)
|
|
ORDER BY COALESCE(m.sort_order, 2147483647) ASC, mv.member_id ASC
|
|
""",
|
|
(as_of, as_of, as_of, as_of),
|
|
)
|
|
return cur.fetchall()
|