feat: add employee management, backend health check, and UI improvements

server/src/app/repositories/employee.py

@@ -1,17 +1,93 @@
-from sqlalchemy.orm import Session
+from __future__ import annotations
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session, selectinload
 
 from app.models.employee import Employee
+from app.models.organization import OrganizationUnit
+from app.models.role import Role
 
 
 class EmployeeRepository:
     def __init__(self, db: Session) -> None:
         self.db = db
 
-    def list(self) -> list[Employee]:
-        return self.db.query(Employee).order_by(Employee.created_at.desc()).all()
+    def list(self, status: str | None = None, keyword: str | None = None) -> list[Employee]:
+        stmt = (
+            select(Employee)
+            .options(
+                selectinload(Employee.organization_unit),
+                selectinload(Employee.manager),
+                selectinload(Employee.roles),
+                selectinload(Employee.change_logs),
+            )
+            .order_by(Employee.updated_at.desc(), Employee.name.asc())
+        )
+
+        if status and status != "全部员工":
+            stmt = stmt.where(Employee.employment_status == status)
+
+        if keyword:
+            pattern = f"%{keyword.strip()}%"
+            stmt = stmt.where(
+                or_(
+                    Employee.name.ilike(pattern),
+                    Employee.employee_no.ilike(pattern),
+                    Employee.email.ilike(pattern),
+                    Employee.position.ilike(pattern),
+                )
+            )
+
+        return list(self.db.execute(stmt).scalars().unique().all())
 
     def get(self, employee_id: str) -> Employee | None:
-        return self.db.query(Employee).filter(Employee.id == employee_id).first()
+        stmt = (
+            select(Employee)
+            .options(
+                selectinload(Employee.organization_unit),
+                selectinload(Employee.manager),
+                selectinload(Employee.roles),
+                selectinload(Employee.change_logs),
+            )
+            .where(Employee.id == employee_id)
+        )
+        return self.db.execute(stmt).scalars().unique().first()
+
+    def get_by_employee_no(self, employee_no: str) -> Employee | None:
+        stmt = select(Employee).where(Employee.employee_no == employee_no)
+        return self.db.execute(stmt).scalars().first()
+
+    def get_by_email(self, email: str) -> Employee | None:
+        stmt = select(Employee).where(Employee.email == email)
+        return self.db.execute(stmt).scalars().first()
+
+    def list_roles(self) -> list[Role]:
+        stmt = select(Role)
+        return list(self.db.execute(stmt).scalars().all())
+
+    def get_role_by_code(self, role_code: str) -> Role | None:
+        stmt = select(Role).where(Role.role_code == role_code)
+        return self.db.execute(stmt).scalars().first()
+
+    def list_organization_units(self) -> list[OrganizationUnit]:
+        stmt = select(OrganizationUnit)
+        return list(self.db.execute(stmt).scalars().all())
+
+    def get_organization_by_code(self, unit_code: str) -> OrganizationUnit | None:
+        stmt = select(OrganizationUnit).where(OrganizationUnit.unit_code == unit_code)
+        return self.db.execute(stmt).scalars().first()
+
+    def count_employees(self) -> int:
+        stmt = select(func.count()).select_from(Employee)
+        return int(self.db.execute(stmt).scalar_one())
+
+    def count_roles(self) -> int:
+        stmt = select(func.count()).select_from(Role)
+        return int(self.db.execute(stmt).scalar_one())
+
+    def count_organization_units(self) -> int:
+        stmt = select(func.count()).select_from(OrganizationUnit)
+        return int(self.db.execute(stmt).scalar_one())
 
     def create(self, employee: Employee) -> Employee:
         self.db.add(employee)