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"}, )