182 lines
6.8 KiB
Python
182 lines
6.8 KiB
Python
from uuid import UUID
|
|
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Optional, TYPE_CHECKING, List
|
|
|
|
from sqlalchemy import DateTime, func, or_
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlmodel import Field, SQLModel, select
|
|
from sqlmodel import Field, Relationship, SQLModel, select
|
|
from department.plugins import as_uuid, now_utc
|
|
from users.models import Users
|
|
|
|
if TYPE_CHECKING:
|
|
from job.job_post.models import JobPosts
|
|
|
|
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)
|
|
short_code: str = Field(index=True, unique=True, max_length=10)
|
|
|
|
requisitions: List["JobPosts"] = Relationship(back_populates="department", sa_relationship_kwargs={"lazy": "selectin"})
|
|
|
|
subtitle: Optional[str] = Field(default=None)
|
|
description: Optional[str] = Field(default=None)
|
|
is_active: bool = Field(default=True)
|
|
parent_department_id: Optional[uuid.UUID] = Field(
|
|
default=None, index=True, foreign_key="departments.id"
|
|
)
|
|
department_head_id: Optional[uuid.UUID] = Field(default=None, foreign_key="users.id")
|
|
location: list[str] = Field(
|
|
default_factory=list, sa_type=JSONB, sa_column_kwargs={"server_default": "[]"}
|
|
)
|
|
created_at: datetime = Field(default_factory=now_utc, sa_type=DateTime(timezone=True))
|
|
updated_at: datetime = Field(default_factory=now_utc, sa_type=DateTime(timezone=True))
|
|
created_by: Optional[uuid.UUID] = Field(default=None, foreign_key="users.id")
|
|
updated_by: Optional[uuid.UUID] = Field(default=None, foreign_key="users.id")
|
|
|
|
@classmethod
|
|
async def get_department_names(cls, session: AsyncSession, search: str | None):
|
|
statement = select(cls.id, cls.name).where(cls.is_active == True)
|
|
if search:
|
|
pattern = f"%{search}%"
|
|
statement = statement.where(
|
|
or_(
|
|
cls.name.ilike(pattern),
|
|
cls.short_code.ilike(pattern),
|
|
cls.subtitle.ilike(pattern),
|
|
)
|
|
)
|
|
statement = statement.order_by(cls.created_at.desc(),cls.id.desc())
|
|
result = await session.execute(statement)
|
|
return result.all()
|
|
|
|
@classmethod
|
|
def _filters(cls, search: str | None, is_active: bool | None):
|
|
clauses = []
|
|
if search:
|
|
pattern = f"%{search}%"
|
|
clauses.append(
|
|
or_(
|
|
cls.name.ilike(pattern),
|
|
cls.short_code.ilike(pattern),
|
|
cls.subtitle.ilike(pattern),
|
|
)
|
|
)
|
|
if is_active is not None:
|
|
clauses.append(cls.is_active == is_active)
|
|
return clauses
|
|
|
|
@classmethod
|
|
async def get_by_id(cls, session: AsyncSession, record_id) -> "Department | None":
|
|
uid = as_uuid(record_id)
|
|
if uid is None:
|
|
return None
|
|
result = await session.execute(select(cls).where(cls.id == uid))
|
|
return result.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_departments(
|
|
cls,
|
|
session: AsyncSession,
|
|
top: int | None = None,
|
|
skip: int = 0,
|
|
search: str | None = None,
|
|
is_active: bool | None = None,
|
|
) -> list["Department"]:
|
|
statement = select(cls).where(*cls._filters(search, is_active)).order_by(cls.name)
|
|
if skip:
|
|
statement = statement.offset(skip)
|
|
if top is not None:
|
|
statement = statement.limit(top)
|
|
result = await session.execute(statement)
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def count_departments(
|
|
cls,
|
|
session: AsyncSession,
|
|
search: str | None = None,
|
|
is_active: bool | None = None,
|
|
) -> int:
|
|
statement = select(func.count()).select_from(cls).where(*cls._filters(search, is_active))
|
|
result = await session.execute(statement)
|
|
return int(result.scalar_one())
|
|
|
|
@classmethod
|
|
async def _commit(cls, session: AsyncSession, department: "Department") -> "Department":
|
|
"""Name/short-code uniqueness is the DB's unique indexes; IntegrityError propagates."""
|
|
session.add(department)
|
|
try:
|
|
await session.commit()
|
|
except IntegrityError:
|
|
await session.rollback()
|
|
raise
|
|
await session.refresh(department)
|
|
return department
|
|
|
|
@classmethod
|
|
async def insert_department(cls, session: AsyncSession, fields: dict) -> "Department":
|
|
return await cls._commit(session, cls(**fields))
|
|
|
|
@classmethod
|
|
async def update_department(
|
|
cls, session: AsyncSession, record_id, fields: dict
|
|
) -> "Department | None":
|
|
department = await cls.get_by_id(session, record_id)
|
|
if not department:
|
|
return None
|
|
for key, value in fields.items():
|
|
setattr(department, key, value)
|
|
department.updated_at = now_utc()
|
|
return await cls._commit(session, department)
|
|
|
|
# @classmethod
|
|
# async def head_options(cls, session: AsyncSession):
|
|
# """Active users for the Department Head picker. COLUMN select, not the Users entity."""
|
|
# result = await session.execute(
|
|
# select(Users.id, Users.name, Users.email)
|
|
# .where(Users.is_deleted == False, Users.is_active == True) # noqa: E712
|
|
# .order_by(Users.name)
|
|
# )
|
|
# return result.all()
|
|
|
|
@classmethod
|
|
async def job_posts_for(cls, session: AsyncSession, department_ids):
|
|
"""(department_id, job_post_id, requisition_status) for non-deleted job posts linked
|
|
to these departments. job_posts.department is free text, so the link is a
|
|
case-insensitive match on the department's name or short code.
|
|
"""
|
|
from job.job_post.models import JobPosts
|
|
|
|
ids = [i for i in (department_ids or []) if i]
|
|
if not ids:
|
|
return []
|
|
job_department = func.lower(func.btrim(JobPosts.department))
|
|
result = await session.execute(
|
|
select(cls.id, JobPosts.id, JobPosts.requisition_status)
|
|
.join(
|
|
JobPosts,
|
|
or_(
|
|
job_department == func.lower(cls.name),
|
|
job_department == func.lower(cls.short_code),
|
|
),
|
|
)
|
|
.where(cls.id.in_(ids), JobPosts.is_deleted == False) # noqa: E712
|
|
)
|
|
return result.all()
|
|
|
|
@classmethod
|
|
async def names_by_ids(cls, session: AsyncSession, ids) -> dict[uuid.UUID, str]:
|
|
uids = {i for i in (ids or []) if i}
|
|
if not uids:
|
|
return {}
|
|
result = await session.execute(select(cls.id, cls.name).where(cls.id.in_(uids)))
|
|
return {row[0]: row[1] for row in result.all()}
|