233 lines
8.4 KiB
Python
233 lines
8.4 KiB
Python
from sqlalchemy import String, Integer, ForeignKey, Index, Boolean, Identity
|
|
from sqlalchemy.orm import mapped_column, Mapped
|
|
|
|
from Services.PostgresService.controllers.mixin_controllers import CrudCollection
|
|
|
|
|
|
class Departments(CrudCollection):
|
|
|
|
__tablename__ = "departments"
|
|
__exclude__fields__ = []
|
|
|
|
parent_department_id = mapped_column(Integer, server_default="0")
|
|
department_code = mapped_column(
|
|
String(16), nullable=False, index=True, comment="Department Code"
|
|
)
|
|
department_name: Mapped[str] = mapped_column(
|
|
String(128), nullable=False, comment="Department Name"
|
|
)
|
|
department_description: Mapped[str] = mapped_column(String, server_default="")
|
|
|
|
company_id: Mapped[int] = mapped_column(ForeignKey("companies.id"), nullable=False)
|
|
company_uu_id: Mapped[str] = mapped_column(
|
|
String, nullable=False, comment="Company UUID"
|
|
)
|
|
|
|
# @classmethod
|
|
# def create_action(cls, data: DepartmentsPydantic, token):
|
|
# data_dict = data.model_dump()
|
|
# data_dict["company_id"] = token.selected_company.company_id
|
|
# return cls.find_or_create(**data_dict)
|
|
|
|
__table_args__ = {"comment": "Departments Information"}
|
|
|
|
|
|
class Duty(CrudCollection):
|
|
|
|
__tablename__ = "duty"
|
|
__exclude__fields__ = []
|
|
|
|
duty_name: Mapped[str] = mapped_column(
|
|
String, unique=True, nullable=False, comment="Duty Name"
|
|
)
|
|
duty_code: Mapped[str] = mapped_column(String, nullable=False, comment="Duty Code")
|
|
duty_description: Mapped[str] = mapped_column(String, comment="Duty Description")
|
|
|
|
# @classmethod
|
|
# def create_action(cls, data: InsertCompanyDuty, token):
|
|
# # if not cls.__is_super__:
|
|
# # raise HTTPException(
|
|
# # status_code=401, detail="You are not authorized to create a duty."
|
|
# # )
|
|
# data_dict = data.model_dump()
|
|
#
|
|
# return cls.find_or_create(**data_dict)
|
|
|
|
__table_args__ = ({"comment": "Duty Information"},)
|
|
|
|
|
|
class Duties(CrudCollection):
|
|
|
|
__tablename__ = "duties"
|
|
__exclude__fields__ = []
|
|
|
|
users_default_duty = mapped_column(
|
|
ForeignKey("duty.id"), nullable=True, comment="Default Duty for Users"
|
|
)
|
|
company_id: Mapped[int] = mapped_column(Integer)
|
|
company_uu_id: Mapped[str] = mapped_column(
|
|
String, nullable=False, comment="Company UUID"
|
|
)
|
|
duties_id: Mapped[int] = mapped_column(ForeignKey("duty.id"), nullable=False)
|
|
duties_uu_id: Mapped[str] = mapped_column(
|
|
String, nullable=False, comment="Duty UUID"
|
|
)
|
|
department_id = mapped_column(
|
|
ForeignKey("departments.id"), nullable=False, comment="Department ID"
|
|
)
|
|
department_uu_id: Mapped[str] = mapped_column(
|
|
String, nullable=False, comment="Department UUID"
|
|
)
|
|
# priority_id: Mapped[int] = mapped_column(ForeignKey("priority.id"), nullable=True)
|
|
management_duty = mapped_column(
|
|
Boolean, server_default="0"
|
|
) # is this a prime Company Duty ???
|
|
|
|
@classmethod
|
|
def init_a_company_default_duties(cls, company_id, company_uu_id):
|
|
__default_init__ = ["Execution Office", "IT Department"]
|
|
|
|
active_row = dict(
|
|
is_confirmed=True, active=True, deleted=False, is_notification_send=True
|
|
)
|
|
list_of_created = []
|
|
|
|
execution = Departments.find_or_create(
|
|
department_name="Execution Office",
|
|
department_code="EO001",
|
|
company_id=company_id,
|
|
company_uu_id=str(company_uu_id),
|
|
**active_row,
|
|
)
|
|
list_of_created.append(execution)
|
|
it_dept = Departments.find_or_create(
|
|
department_name="IT Department",
|
|
department_code="ITD001",
|
|
company_id=company_id,
|
|
company_uu_id=str(company_uu_id),
|
|
**active_row,
|
|
)
|
|
list_of_created.append(it_dept)
|
|
bm_duty = Duty.find_or_create(
|
|
duty_name="Business Manager",
|
|
duty_code="BM0001",
|
|
duty_description="Business Manager",
|
|
**active_row,
|
|
)
|
|
list_of_created.append(bm_duty)
|
|
it_duty = Duty.find_or_create(
|
|
duty_name="IT Manager",
|
|
duty_code="IT0001",
|
|
duty_description="IT Manager",
|
|
**active_row,
|
|
)
|
|
list_of_created.append(it_duty)
|
|
bulk_duty = Duty.find_or_create(
|
|
duty_name="BULK",
|
|
duty_code="BULK",
|
|
duty_description="BULK RECORDS OF THE COMPANY",
|
|
**active_row,
|
|
)
|
|
list_of_created.append(bulk_duty)
|
|
occu_duty = Duty.find_or_create(
|
|
duty_name="OCCUPANT",
|
|
duty_code="OCCUPANT",
|
|
duty_description="OCCUPANT RECORDS OF THE COMPANY",
|
|
**active_row,
|
|
)
|
|
list_of_created.append(occu_duty)
|
|
duties_created_bm = cls.find_or_create(
|
|
company_id=company_id,
|
|
company_uu_id=str(company_uu_id),
|
|
duties_id=bm_duty.id,
|
|
duties_uu_id=str(bm_duty.uu_id),
|
|
department_id=execution.id,
|
|
department_uu_id=str(execution.uu_id),
|
|
**active_row,
|
|
)
|
|
list_of_created.append(duties_created_bm)
|
|
duties_created_it = cls.find_or_create(
|
|
company_id=company_id,
|
|
company_uu_id=str(company_uu_id),
|
|
duties_id=it_duty.id,
|
|
duties_uu_id=str(it_duty.uu_id),
|
|
department_id=it_dept.id,
|
|
department_uu_id=str(it_dept.uu_id),
|
|
**active_row,
|
|
)
|
|
list_of_created.append(duties_created_it)
|
|
duties_created__ex = cls.find_or_create(
|
|
company_id=company_id,
|
|
company_uu_id=str(company_uu_id),
|
|
duties_id=bulk_duty.id,
|
|
duties_uu_id=str(bulk_duty.uu_id),
|
|
department_id=execution.id,
|
|
department_uu_id=str(execution.uu_id),
|
|
**active_row,
|
|
)
|
|
list_of_created.append(duties_created__ex)
|
|
duties_created_at = cls.find_or_create(
|
|
company_id=company_id,
|
|
company_uu_id=str(company_uu_id),
|
|
duties_id=occu_duty.id,
|
|
duties_uu_id=str(occu_duty.uu_id),
|
|
department_id=execution.id,
|
|
department_uu_id=str(execution.uu_id),
|
|
**active_row,
|
|
)
|
|
list_of_created.append(duties_created_at)
|
|
return list_of_created
|
|
|
|
@classmethod
|
|
def get_bulk_duties_of_a_company(cls, company_id):
|
|
duties_id = Duty.filter_by_one(system=True, duty_code="BULK").data
|
|
if bulk_duties := Duties.filter_by_one(
|
|
duties_id=getattr(duties_id, "id", None),
|
|
company_id=company_id,
|
|
**Duties.valid_record_dict,
|
|
).data:
|
|
return bulk_duties
|
|
raise Exception("Bulk Duty not found. Please contact with supervisor.")
|
|
|
|
# @classmethod
|
|
# def create_action(cls, data: InsertCompanyDuty):
|
|
# data_dict = data.model_dump()
|
|
# if department := Departments.find_one(uu_id=data.department_uu_id):
|
|
# data_dict["department_id"] = department.id
|
|
# del data_dict["department_uu_id"]
|
|
# return cls.find_or_create(**data_dict)
|
|
|
|
__table_args__ = (
|
|
Index("duty_ndx_00", company_id, duties_id, department_id, unique=True),
|
|
{"comment": "Duty & Company & Department Information"},
|
|
)
|
|
|
|
# department: Mapped[List["CompanyDepartments"]] = relationship(
|
|
# "CompanyDepartments", back_populates="duties", foreign_keys=[department_id]
|
|
# )
|
|
# employees: Mapped[List["CompanyEmployees"]] = relationship(
|
|
# "CompanyEmployees",
|
|
# back_populates="duty",
|
|
# foreign_keys="CompanyEmployees.duty_id",
|
|
# )
|
|
# duty_app: Mapped["CompanyDutyApp"] = relationship(
|
|
# "CompanyDutyApp", back_populates="duties", foreign_keys="CompanyDutyApp.company_duty_id"
|
|
# )
|
|
|
|
# def get_language_of_duty(self, lang):
|
|
# if erp_text := ErpText.find_one(lang=lang, text_code=self.duty_code):
|
|
# return erp_text.text_name, erp_text.text_description
|
|
# return None, None
|
|
|
|
# company: Mapped["Companies"] = relationship(
|
|
# "Company", back_populates="departments", foreign_keys=[company_id]
|
|
# )
|
|
# duties: Mapped[List["CompanyDuty"]] = relationship(
|
|
# "CompanyDuty",
|
|
# back_populates="department",
|
|
# foreign_keys="CompanyDuty.department_id",
|
|
# )
|
|
# app_item: Mapped["AppItems"] = relationship(
|
|
# "AppItems", back_populates="department", foreign_keys="AppItems.department_id"
|
|
# )
|