74 lines
3.1 KiB
Python
74 lines
3.1 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Optional,List
|
|
from sqlalchemy import DateTime, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlmodel import Field, SQLModel, select
|
|
from fastapi import HTTPException
|
|
from users.models import Users
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
class Department(SQLModel, table=True):
|
|
__tablename__ = "departments"
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
name: str = Field(index=True,unique=True)
|
|
description: Optional[str] = Field(default="")
|
|
short_code: Optional[str] = Field(default="",max_length=10)
|
|
is_active: bool = Field(default=True)
|
|
parent_department_id: Optional[uuid.UUID] = Field(foreign_key="departments.id")
|
|
department_head_id: Optional[uuid.UUID] = Field(foreign_key="users.id")
|
|
location:Optional[List[str]] = Field(default_factory=list)
|
|
subtitle:Optional[str] = Field(default="")
|
|
created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True))
|
|
created_by: Optional[uuid.UUID] = Field(foreign_key="users.id")
|
|
updated_by: Optional[uuid.UUID] = Field(foreign_key="users.id")
|
|
|
|
@staticmethod
|
|
def _as_uuid(record_id) -> uuid.UUID | None:
|
|
if record_id in (None, ""):
|
|
return None
|
|
try:
|
|
return uuid.UUID(str(record_id))
|
|
except ValueError:
|
|
return None
|
|
|
|
@classmethod
|
|
async def create_department(cls,session:AsyncSession,data:dict):
|
|
try:
|
|
uid=cls._as_uuid(data.get("parent_department_id"))
|
|
if data.get("department_head_id"):
|
|
user=await Users.get_by_id(session,data.get("department_head_id"))
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
department_head_id=user.id
|
|
else:
|
|
raise HTTPException(status_code=400, detail="Department head ID is required")
|
|
|
|
department=cls(
|
|
name=data.get("name"),
|
|
description=data.get("description"),
|
|
short_code=data.get("short_code"),
|
|
parent_department_id=uid if data.get("parent_department_id") else None,
|
|
department_head_id=department_head_id,
|
|
location=data.get("location"),
|
|
subtitle=data.get("subtitle"),
|
|
created_by=data.get("created_by"),
|
|
updated_by=data.get("updated_by"),
|
|
)
|
|
session.add(department)
|
|
await session.commit()
|
|
return department
|
|
except Exception as e:
|
|
session.rollback()
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@classmethod
|
|
async def get_department_by_id(cls,session:AsyncSession,department_id:uuid.UUID):
|
|
uid=cls._as_uuid(department_id)
|
|
if uid is None:
|
|
raise ValueError("Invalid department ID")
|
|
result=await session.execute(select(cls).where(cls.id == uid))
|
|
return result.scalars().first() |