File size: 4,822 Bytes
d99646e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | from sqlalchemy import func
from sqlalchemy.orm import Session, selectinload
from app.models.department import Department
from app.models.user import User
from app.schemas.admin import AdminDepartmentUpdate, AdminUserUpdate
from app.services.auth_service import get_department_by_name, resolve_signup_target
def _department_exists_with_name_or_code(
db: Session,
*,
name: str | None = None,
code: str | None = None,
exclude_id: int | None = None,
) -> bool:
query = db.query(Department)
if exclude_id is not None:
query = query.filter(Department.id != exclude_id)
if name is not None:
query = query.filter(func.lower(Department.name) == name.lower())
if code is not None:
query = query.filter(func.lower(Department.code) == code.lower())
return query.first() is not None
def list_departments(db: Session) -> list[Department]:
return (
db.query(Department)
.filter(Department.is_active.is_(True))
.order_by(Department.name)
.all()
)
def create_department(
db: Session,
*,
name: str,
code: str,
description: str | None = None,
keywords: str | None = None,
) -> Department:
if _department_exists_with_name_or_code(db, name=name):
raise ValueError("Department name is already registered")
if _department_exists_with_name_or_code(db, code=code):
raise ValueError("Department code is already registered")
department = Department(
name=name,
code=code,
description=description,
keywords=keywords,
)
db.add(department)
db.commit()
db.refresh(department)
return department
def update_department(
db: Session,
*,
department_id: int,
payload: AdminDepartmentUpdate,
) -> Department:
department = db.get(Department, department_id)
if department is None:
raise ValueError("Department not found")
if payload.name is not None and _department_exists_with_name_or_code(
db,
name=payload.name,
exclude_id=department_id,
):
raise ValueError("Department name is already registered")
if payload.code is not None and _department_exists_with_name_or_code(
db,
code=payload.code,
exclude_id=department_id,
):
raise ValueError("Department code is already registered")
if payload.name is not None:
department.name = payload.name
if payload.code is not None:
department.code = payload.code
if payload.description is not None:
department.description = payload.description
if payload.keywords is not None:
department.keywords = payload.keywords
if payload.is_active is not None:
department.is_active = payload.is_active
db.commit()
db.refresh(department)
return department
def deactivate_department(db: Session, *, department_id: int) -> Department:
return update_department(
db,
department_id=department_id,
payload=AdminDepartmentUpdate(is_active=False),
)
def get_department(db: Session, *, department_id: int) -> Department | None:
return db.get(Department, department_id)
def list_users(db: Session) -> list[User]:
return (
db.query(User)
.options(selectinload(User.department))
.order_by(User.id)
.all()
)
def get_user(db: Session, *, user_id: int) -> User | None:
return (
db.query(User)
.options(selectinload(User.department))
.filter(User.id == user_id)
.first()
)
def update_user(
db: Session,
*,
user_id: int,
payload: AdminUserUpdate,
) -> User:
user = (
db.query(User)
.options(selectinload(User.department))
.filter(User.id == user_id)
.first()
)
if user is None:
raise ValueError("User not found")
if payload.email is not None:
existing_user = (
db.query(User)
.filter(func.lower(User.email) == payload.email.lower(), User.id != user_id)
.first()
)
if existing_user is not None:
raise ValueError("Email is already registered")
user.email = payload.email.lower()
if payload.full_name is not None:
user.full_name = payload.full_name
if payload.department_name is not None:
role, department_id = resolve_signup_target(db, payload.department_name)
user.role = role
user.department_id = department_id
if payload.is_active is not None:
user.is_active = payload.is_active
db.commit()
db.refresh(user)
return user
def deactivate_user(db: Session, *, user_id: int) -> User:
return update_user(
db,
user_id=user_id,
payload=AdminUserUpdate(is_active=False),
)
|