huoyan-enterprise/backend/services/department_service.py

"""部门管理"""
from typing import List, Optional
import asyncpg


async def get_subtree_ids(conn: asyncpg.Connection, dept_id: int) -> List[int]:
    """递归查询某部门及其所有子孙部门的 ID 列表。"""
    rows = await conn.fetch(
        """
        WITH RECURSIVE sub AS (
            SELECT id FROM department WHERE id = $1
            UNION ALL
            SELECT d.id FROM department d JOIN sub ON d.parent_id = sub.id
        )
        SELECT id FROM sub
        """,
        dept_id,
    )
    return [r["id"] for r in rows]


class DepartmentService:
    @staticmethod
    async def list_by_enterprise(
        conn: asyncpg.Connection, enterprise_id: int
    ) -> List[dict]:
        rows = await conn.fetch(
            """
            SELECT d.id, d.enterprise_id, d.name, d.parent_id,
                   d.leader_user_id, d.created_at, d.updated_at,
                   COALESCE(NULLIF(TRIM(u.display_name),''), u.username) AS leader_name
            FROM department d
            LEFT JOIN user_list u ON u.id = d.leader_user_id
            WHERE d.enterprise_id = $1
            ORDER BY d.id ASC
            """,
            enterprise_id,
        )
        return [dict(r) for r in rows]

    @staticmethod
    async def get_by_id(
        conn: asyncpg.Connection, dept_id: int, enterprise_id: int
    ) -> Optional[dict]:
        row = await conn.fetchrow(
            """
            SELECT d.id, d.enterprise_id, d.name, d.parent_id,
                   d.leader_user_id, d.created_at, d.updated_at,
                   COALESCE(NULLIF(TRIM(u.display_name),''), u.username) AS leader_name
            FROM department d
            LEFT JOIN user_list u ON u.id = d.leader_user_id
            WHERE d.id = $1 AND d.enterprise_id = $2
            """,
            dept_id,
            enterprise_id,
        )
        return dict(row) if row else None

    @staticmethod
    async def set_leader(
        conn: asyncpg.Connection,
        dept_id: int,
        enterprise_id: int,
        leader_user_id: Optional[int],
    ) -> Optional[dict]:
        """设置或清除部门负责人。leader_user_id=None 表示清除。"""
        row = await conn.fetchrow(
            """
            UPDATE department
            SET leader_user_id = $1, updated_at = CURRENT_TIMESTAMP
            WHERE id = $2 AND enterprise_id = $3
            RETURNING id, enterprise_id, name, parent_id, leader_user_id, created_at, updated_at
            """,
            leader_user_id,
            dept_id,
            enterprise_id,
        )
        return dict(row) if row else None

    @staticmethod
    async def create(
        conn: asyncpg.Connection,
        enterprise_id: int,
        name: str,
        parent_id: Optional[int] = None,
    ) -> dict:
        row = await conn.fetchrow(
            """
            INSERT INTO department (enterprise_id, name, parent_id)
            VALUES ($1, $2, $3)
            RETURNING id, enterprise_id, name, parent_id, created_at, updated_at
            """,
            enterprise_id,
            name,
            parent_id,
        )
        return dict(row)

    @staticmethod
    async def update(
        conn: asyncpg.Connection,
        dept_id: int,
        enterprise_id: int,
        name: Optional[str] = None,
        parent_id: Optional[int] = None,
    ) -> Optional[dict]:
        fields: List[str] = []
        params: List = []
        if name is not None:
            fields.append(f"name = ${len(params) + 1}")
            params.append(name)
        if parent_id is not None:
            fields.append(f"parent_id = ${len(params) + 1}")
            params.append(parent_id)
        if not fields:
            return await DepartmentService.get_by_id(conn, dept_id, enterprise_id)
        wid = len(params) + 1
        we = len(params) + 2
        params.extend([dept_id, enterprise_id])
        q = f"""
            UPDATE department SET {", ".join(fields)}, updated_at = CURRENT_TIMESTAMP
            WHERE id = ${wid} AND enterprise_id = ${we}
            RETURNING id, enterprise_id, name, parent_id, created_at, updated_at
        """
        row = await conn.fetchrow(q, *params)
        return dict(row) if row else None

    @staticmethod
    async def delete(
        conn: asyncpg.Connection, dept_id: int, enterprise_id: int
    ) -> Optional[str]:
        cnt = await conn.fetchval(
            """
            SELECT COUNT(*) FROM user_list
            WHERE department_id = $1 AND enterprise_id = $2
            """,
            dept_id,
            enterprise_id,
        )
        if cnt and int(cnt) > 0:
            return "部门下仍有用户，无法删除"
        row = await conn.fetchrow(
            """
            DELETE FROM department
            WHERE id = $1 AND enterprise_id = $2
            RETURNING id
            """,
            dept_id,
            enterprise_id,
        )
        return None if row else "部门不存在"