483 lines
16 KiB
Python
483 lines
16 KiB
Python
import arrow
|
|
|
|
from sqlalchemy import (
|
|
String,
|
|
Integer,
|
|
Boolean,
|
|
ForeignKey,
|
|
Index,
|
|
TIMESTAMP,
|
|
Text,
|
|
func,
|
|
BigInteger,
|
|
Numeric,
|
|
or_,
|
|
)
|
|
from sqlalchemy.orm import mapped_column, relationship, Mapped
|
|
from Controllers.Postgres.mixin import CrudCollection
|
|
|
|
|
|
class UsersTokens(CrudCollection):
|
|
|
|
__tablename__ = "users_tokens"
|
|
__exclude__fields__ = []
|
|
|
|
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
|
|
|
|
token_type: Mapped[str] = mapped_column(String(16), server_default="RememberMe")
|
|
token: Mapped[str] = mapped_column(String, server_default="")
|
|
domain: Mapped[str] = mapped_column(String, server_default="")
|
|
expires_at: Mapped[TIMESTAMP] = mapped_column(
|
|
TIMESTAMP(timezone=True),
|
|
default=str(arrow.now().shift(days=3)),
|
|
)
|
|
|
|
# users = relationship("Users", back_populates="tokens", foreign_keys=[user_id])
|
|
|
|
|
|
class Credentials(CrudCollection):
|
|
"""
|
|
Credentials class to store user credentials
|
|
"""
|
|
|
|
__tablename__ = "credentials"
|
|
__exclude__fields__ = []
|
|
|
|
credential_token: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Credential token for authentication"
|
|
)
|
|
user_id: Mapped[int] = mapped_column(
|
|
ForeignKey("users.id"), nullable=False, comment="Foreign key to users table"
|
|
)
|
|
user_uu_id: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="User UUID", index=True
|
|
)
|
|
person_id: Mapped[int] = mapped_column(
|
|
ForeignKey("people.id"), nullable=False, comment="Foreign key to person table"
|
|
)
|
|
person_uu_id: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Person UUID", index=True
|
|
)
|
|
name: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Name of the user", index=True
|
|
)
|
|
surname: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Surname of the user", index=True
|
|
)
|
|
email: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Email address of the user", index=True
|
|
)
|
|
phone: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Phone number of the user", index=True
|
|
)
|
|
is_verified: Mapped[bool] = mapped_column(
|
|
Boolean, server_default="0", comment="Flag to check if user is verified"
|
|
)
|
|
|
|
def generate_token(self) -> str:
|
|
"""
|
|
Generate a unique token for the user
|
|
"""
|
|
name_token, rest_of_token = "", ""
|
|
if self.name and self.surname:
|
|
name_token = f"{self.name[0].upper()}{self.surname[0].upper()}"
|
|
return ""
|
|
|
|
|
|
class Users(CrudCollection):
|
|
"""
|
|
Application User frame to connect to API with assigned token-based HTTP connection
|
|
"""
|
|
|
|
__tablename__ = "users"
|
|
__exclude__fields__ = [
|
|
"hash_password",
|
|
"password_token",
|
|
"expiry_begins",
|
|
"related_company",
|
|
]
|
|
|
|
user_tag: Mapped[str] = mapped_column(
|
|
String(64), server_default="", comment="Unique tag for the user", index=True
|
|
)
|
|
email: Mapped[str] = mapped_column(
|
|
String(128), server_default="", comment="Email address of the user", index=True
|
|
)
|
|
phone_number: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Phone number of the user", index=True
|
|
)
|
|
via: Mapped[str] = mapped_column(
|
|
String,
|
|
server_default="111",
|
|
comment="Email 1/ Phone 2/ User Tag 3 All 111 Only 100",
|
|
)
|
|
|
|
avatar: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Avatar URL for the user"
|
|
)
|
|
hash_password: Mapped[str] = mapped_column(
|
|
String(256), server_default="", comment="Hashed password for security"
|
|
)
|
|
password_token: Mapped[str] = mapped_column(
|
|
String(256), server_default="", comment="Token for password reset"
|
|
)
|
|
remember_me: Mapped[bool] = mapped_column(
|
|
Boolean, server_default="0", comment="Flag to remember user login"
|
|
)
|
|
|
|
password_expires_day: Mapped[int] = mapped_column(
|
|
Integer,
|
|
server_default=str(30),
|
|
comment="Password expires in days",
|
|
)
|
|
password_expiry_begins: Mapped[TIMESTAMP] = mapped_column(
|
|
TIMESTAMP(timezone=True),
|
|
server_default=func.now(),
|
|
comment="Timestamp when password expiry begins",
|
|
)
|
|
related_company: Mapped[str] = mapped_column(String, comment="Related Company UUID")
|
|
|
|
person_id: Mapped[int] = mapped_column(
|
|
ForeignKey("people.id"), nullable=False, comment="Foreign key to person table"
|
|
)
|
|
person_uu_id: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Person UUID", index=True
|
|
)
|
|
local_timezone = mapped_column(
|
|
String, server_default="GMT+3", comment="Local timezone of user"
|
|
)
|
|
person = relationship("People", back_populates="user", foreign_keys=[person_id])
|
|
#
|
|
# @property
|
|
# def is_occupant(self):
|
|
# return not str(self.email).split("@")[1] == Auth.ACCESS_EMAIL_EXT
|
|
#
|
|
# @property
|
|
# def is_employee(self):
|
|
# return str(self.email).split("@")[1] == Auth.ACCESS_EMAIL_EXT
|
|
#
|
|
# @property
|
|
# def user_type(self):
|
|
# return "Occupant" if self.is_occupant else "Employee"
|
|
#
|
|
# @classmethod
|
|
# def credentials(cls):
|
|
# db_session = cls.new_session()
|
|
# person_object: People = People.filter_by_one(
|
|
# db=db_session, system=True, id=cls.person_id
|
|
# ).data
|
|
# if person_object:
|
|
# return {
|
|
# "person_id": person_object.id,
|
|
# "person_uu_id": str(person_object.uu_id),
|
|
# }
|
|
# return {
|
|
# "person_id": None,
|
|
# "person_uu_id": None,
|
|
# }
|
|
#
|
|
# @property
|
|
# def password_expiry_ends(self):
|
|
# """Calculates the expiry end date based on expiry begins and expires day"""
|
|
# return self.password_expiry_begins + timedelta(
|
|
# days=int(
|
|
# "".join(
|
|
# [
|
|
# _
|
|
# for _ in str(self.password_expires_day).split(",")[0]
|
|
# if _.isdigit()
|
|
# ]
|
|
# )
|
|
# )
|
|
# )
|
|
#
|
|
# @classmethod
|
|
# def create_action(cls, create_user: InsertUsers, token_dict):
|
|
# db_session = cls.new_session()
|
|
# found_person = People.filter_one(
|
|
# People.uu_id == create_user.people_uu_id,
|
|
# db=db_session,
|
|
# ).data
|
|
#
|
|
# if not found_person:
|
|
# raise HTTPException(status_code=400, detail="Person not found.")
|
|
# if (
|
|
# not any(i in str(create_user.email) for i in ["@", "."])
|
|
# and not len(str(create_user.phone_number)) >= 10
|
|
# ):
|
|
# raise HTTPException(
|
|
# status_code=400,
|
|
# detail="Please enter at least one valid email or phone number.",
|
|
# )
|
|
# if not create_user.avatar:
|
|
# create_user.avatar = ApiStatic.PLACEHOLDER
|
|
# create_dict = create_user.model_dump()
|
|
# del create_dict["people_uu_id"]
|
|
# create_dict["person_id"] = found_person.id
|
|
# create_dict["person_uu_id"] = str(found_person.uu_id)
|
|
# create_dict["related_company"] = token_dict.selected_company.company_uu_id
|
|
# created_user = cls.find_or_create(**create_dict)
|
|
# created_user.reset_password_token(found_user=created_user)
|
|
# return created_user
|
|
#
|
|
# def get_employee_and_duty_details(self):
|
|
# from ApiLayers.Schemas import Employees, Duties
|
|
#
|
|
# db_session = self.new_session()
|
|
# found_person = People.filter_one(
|
|
# People.id == self.person_id,
|
|
# db=db_session,
|
|
# )
|
|
# found_employees = Employees.filter_by_active(
|
|
# people_id=found_person.id, is_confirmed=True, db=db_session
|
|
# )
|
|
# found_duties = Duties.filter_all(
|
|
# Duties.is_confirmed == True,
|
|
# Duties.id.in_(
|
|
# list(found_employee.duty_id for found_employee in found_employees.data)
|
|
# ),
|
|
# db=db_session,
|
|
# )
|
|
# if not found_employees.count:
|
|
# raise HTTPException(
|
|
# status_code=401,
|
|
# detail={
|
|
# "message": "Person has no confirmed duty. No employee match please register "
|
|
# "your super admin",
|
|
# "completed": False,
|
|
# },
|
|
# )
|
|
# return {
|
|
# "duty_list": [
|
|
# {
|
|
# "duty_id": duty.id,
|
|
# "duty_uu_id": duty.uu_id.__str__(),
|
|
# "duty_code": duty.duty_code,
|
|
# "duty_name": duty.duty_name,
|
|
# "duty_description": duty.duty_description,
|
|
# }
|
|
# for duty in found_duties.data
|
|
# ],
|
|
# }
|
|
#
|
|
# def get_main_domain_and_other_domains(self, get_main_domain: bool = True):
|
|
# from ApiLayers.Schemas import MongoQueryIdentity
|
|
#
|
|
# query_engine = MongoQueryIdentity(company_uuid=self.related_company)
|
|
# domain_via_user = query_engine.get_domain_via_user(user_uu_id=str(self.uu_id))
|
|
# if not domain_via_user:
|
|
# raise HTTPException(
|
|
# status_code=401,
|
|
# detail="Domain not found. Please contact the admin.",
|
|
# )
|
|
# domain_via_user = domain_via_user[0]
|
|
# if get_main_domain:
|
|
# return domain_via_user.get("main_domain", None)
|
|
# return domain_via_user.get("other_domains_list", None)
|
|
|
|
|
|
class RelationshipDutyPeople(CrudCollection):
|
|
|
|
__tablename__ = "relationship_duty_people"
|
|
__exclude__fields__ = []
|
|
|
|
company_id: Mapped[int] = mapped_column(
|
|
ForeignKey("companies.id"), nullable=False
|
|
) # 1, 2, 3
|
|
duties_id: Mapped[int] = mapped_column(
|
|
ForeignKey("duties.id"), nullable=False
|
|
) # duty -> (n)person Evyos LTD
|
|
member_id: Mapped[int] = mapped_column(
|
|
ForeignKey("people.id"), nullable=False
|
|
) # 2, 3, 4
|
|
|
|
relationship_type: Mapped[str] = mapped_column(
|
|
String, nullable=True, server_default="Employee"
|
|
) # Commercial
|
|
show_only: Mapped[bool] = mapped_column(Boolean, server_default="0")
|
|
|
|
# related_company: Mapped[List["Company"]] = relationship(
|
|
# "Company",
|
|
# back_populates="related_companies",
|
|
# foreign_keys=[related_company_id],
|
|
# )
|
|
|
|
__table_args__ = (
|
|
Index(
|
|
"person_relationship_ndx_01",
|
|
company_id,
|
|
duties_id,
|
|
member_id,
|
|
relationship_type,
|
|
unique=True,
|
|
),
|
|
{"comment": "Person Relationship Information"},
|
|
)
|
|
|
|
|
|
class People(CrudCollection):
|
|
"""
|
|
People that are related to users in the application
|
|
"""
|
|
|
|
__tablename__ = "people"
|
|
__exclude__fields__ = []
|
|
__many__table__ = RelationshipDutyPeople
|
|
__encrypt_list__ = [
|
|
"father_name",
|
|
"mother_name",
|
|
"country_code",
|
|
"national_identity_id",
|
|
"birth_place",
|
|
"birth_date",
|
|
"tax_no",
|
|
]
|
|
|
|
firstname: Mapped[str] = mapped_column(
|
|
String, nullable=False, comment="First name of the person"
|
|
)
|
|
surname: Mapped[str] = mapped_column(
|
|
String(24), nullable=False, comment="Surname of the person"
|
|
)
|
|
middle_name: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Middle name of the person"
|
|
)
|
|
sex_code: Mapped[str] = mapped_column(
|
|
String(1), nullable=False, comment="Sex code of the person (e.g., M/F)"
|
|
)
|
|
person_ref: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Reference ID for the person"
|
|
)
|
|
person_tag: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Unique tag for the person"
|
|
)
|
|
|
|
# ENCRYPT DATA
|
|
father_name: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Father's name of the person"
|
|
)
|
|
mother_name: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Mother's name of the person"
|
|
)
|
|
country_code: Mapped[str] = mapped_column(
|
|
String(4), server_default="TR", comment="Country code of the person"
|
|
)
|
|
national_identity_id: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="National identity ID of the person"
|
|
)
|
|
birth_place: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Birth place of the person"
|
|
)
|
|
birth_date: Mapped[TIMESTAMP] = mapped_column(
|
|
TIMESTAMP(timezone=True),
|
|
server_default="1900-01-01",
|
|
comment="Birth date of the person",
|
|
)
|
|
tax_no: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Tax number of the person"
|
|
)
|
|
# Receive at Create person
|
|
# language = mapped_column(
|
|
# String, comment="Language code of the person"
|
|
# )
|
|
# currency = mapped_column(
|
|
# String, comment="Currency code of the person"
|
|
# )
|
|
|
|
# ENCRYPT DATA
|
|
user = relationship(
|
|
"Users", back_populates="person", foreign_keys="Users.person_id"
|
|
)
|
|
|
|
__table_args__ = (
|
|
Index(
|
|
"person_ndx_001",
|
|
national_identity_id,
|
|
unique=True,
|
|
),
|
|
{"comment": "Person Information"},
|
|
)
|
|
|
|
@property
|
|
def full_name(self):
|
|
if self.middle_name:
|
|
return f"{self.firstname} {self.middle_name} {self.surname}"
|
|
return f"{self.firstname} {self.surname}"
|
|
|
|
|
|
class OccupantTypes(CrudCollection):
|
|
"""
|
|
Occupant Types class based on declarative_base and BaseMixin via session
|
|
"""
|
|
|
|
__tablename__ = "occupant_types"
|
|
__exclude__fields__ = []
|
|
|
|
occupant_type: Mapped[str] = mapped_column(
|
|
String, nullable=False, comment="Occupant Type"
|
|
)
|
|
occupant_description: Mapped[str] = mapped_column(String, server_default="")
|
|
occupant_code: Mapped[str] = mapped_column(String, server_default="")
|
|
occupant_category: Mapped[str] = mapped_column(String, server_default="")
|
|
occupant_category_type: Mapped[str] = mapped_column(String, server_default="")
|
|
occupant_is_unique: Mapped[bool] = mapped_column(Boolean, server_default="0")
|
|
|
|
__table_args__ = ({"comment": "Occupant Types Information"},)
|
|
|
|
|
|
class Contracts(CrudCollection):
|
|
"""
|
|
Contract class based on declarative_base and BaseMixin via session
|
|
"""
|
|
|
|
__tablename__ = "contracts"
|
|
__exclude__fields__ = []
|
|
|
|
contract_type: Mapped[str] = mapped_column(
|
|
String(5),
|
|
nullable=False,
|
|
comment="The code for personnel is P and the code for companies is C.",
|
|
)
|
|
contract_title: Mapped[str] = mapped_column(String(255))
|
|
contract_details: Mapped[str] = mapped_column(Text)
|
|
contract_terms: Mapped[str] = mapped_column(Text)
|
|
|
|
contract_code: Mapped[str] = mapped_column(
|
|
String(100),
|
|
nullable=False,
|
|
comment="contract_code is the unique code given by the system.",
|
|
)
|
|
contract_date: Mapped[TIMESTAMP] = mapped_column(
|
|
TIMESTAMP(timezone=True),
|
|
server_default="2099-12-31 23:59:59",
|
|
comment="contract date is the date the contract is made. "
|
|
"expire start is the start date of the contract, expire en is the end date of the contract.",
|
|
)
|
|
|
|
company_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("companies.id"), nullable=True
|
|
)
|
|
company_uu_id: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Company UUID"
|
|
)
|
|
|
|
person_id: Mapped[int] = mapped_column(
|
|
Integer, ForeignKey("people.id"), nullable=True
|
|
)
|
|
person_uu_id: Mapped[str] = mapped_column(
|
|
String, server_default="", comment="Person UUID"
|
|
)
|
|
|
|
@classmethod
|
|
def retrieve_contact_no(cls):
|
|
# todo When create record contract_code == below string
|
|
related_date, counter = Contracts.client_arrow.now(), 1
|
|
return (
|
|
f"{related_date.date().year}{str(cls.contract_type)}{str(counter).zfill(6)}"
|
|
)
|
|
|
|
__table_args__ = (
|
|
Index("_contract_ndx_01", contract_code, unique=True),
|
|
{"comment": "Contract Information"},
|
|
)
|