from datetime import timedelta from fastapi import HTTPException from sqlalchemy import ( String, Integer, Boolean, ForeignKey, Index, TIMESTAMP, func, Text, BigInteger, Numeric, or_, ) from sqlalchemy.orm import mapped_column, relationship, Mapped from api_library.date_time_actions.date_functions import system_arrow from api_configs import Auth, ApiStatic, RelationAccess from databases.sql_models.core_mixin import CrudCollection from databases.extensions import SelectAction, SelectActionWithEmployee from databases.extensions.auth import UserLoginModule from api_validations.validations_request import InsertUsers, InsertPerson from databases.language_models.identity.identity import ( UsersTokensLanguageModel, UsersLanguageModel, PeopleLanguageModel, RelationshipDutyPeopleLanguageModel, RelationshipEmployee2PostCodeLanguageModel, AddressPostcodeLanguageModel, AddressesLanguageModel, AddressGeographicLocationsLanguageModel, AddressCountryLanguageModel, AddressStateLanguageModel, AddressCityLanguageModel, AddressDistrictLanguageModel, AddressLocalityLanguageModel, AddressNeighborhoodLanguageModel, AddressStreetLanguageModel, OccupantTypesLanguageModel, ContractsLanguageModel, ) class UsersTokens(CrudCollection): __tablename__ = "users_tokens" __exclude__fields__ = [] __language_model__ = UsersTokensLanguageModel 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(system_arrow.shift(date=system_arrow.now(), days=3)), ) # users = relationship("Users", back_populates="tokens", foreign_keys=[user_id]) class Users(CrudCollection, UserLoginModule, SelectAction): """ 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", ] __language_model__ = UsersLanguageModel 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(Auth.PASSWORD_EXPIRE_DAY.days), 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 ) 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 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() ] ) ) ) @property def is_super_user(self): """Checks if the user is a superuser based on priority code""" return getattr(self.priority, "priority_code", 0) == 78 @property def is_user(self): """Checks if the user is a regular user based on priority code""" return getattr(self.priority, "priority_code", 0) == 0 @classmethod def create_action(cls, create_user: InsertUsers, token_dict): found_person = People.filter_one( People.uu_id == create_user.people_uu_id, ).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 @classmethod def credentials(cls): person_object = People.filter_by_one(system=True, id=cls.person_id).data # if not person_object: # raise HTTPException( # status_code=401, # detail="Person not found. Please contact the admin.", # ) 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, } def get_employee_and_duty_details(self): from databases import Employees, Duties found_person = People.filter_one( People.id == self.person_id, ) found_employees = Employees.filter_by_active( people_id=found_person.id, is_confirmed=True ) found_duties = Duties.filter_all( Duties.is_confirmed == True, Duties.id.in_( list(found_employee.duty_id for found_employee in found_employees.data) ), ) 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 databases 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__ = [] __access_by__ = RelationAccess.SuperAccessList __language_model__ = RelationshipDutyPeopleLanguageModel 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, SelectAction): """ 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", ] __language_model__ = PeopleLanguageModel 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" ) # 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}" @classmethod def create_action(cls, data: InsertPerson, token): from databases import Duties token_duties_id, token_company_id = ( token.selected_company.duty_id, token.selected_company.company_id, ) bulk_duty = Duties.get_bulk_duties_of_a_company(company_id=token_company_id) if str(data.country_code) == "TR": if not len(data.national_identity_id) == 11: raise HTTPException( status_code=400, detail="Please enter a valid national identity number.", ) if data.tax_no and not len(str(data.tax_no)) == 10: raise HTTPException( status_code=400, detail="Please enter a valid tax number.", ) create_dict = data.model_dump() create_dict["firstname"] = str(create_dict["firstname"]).capitalize() create_dict["middle_name"] = str(create_dict["middle_name"]).capitalize() create_dict["surname"] = str(create_dict["surname"]).upper() create_dict["birth_place"] = str(create_dict["birth_place"]).upper() created_people = cls.find_or_create(**create_dict) created_people.update(is_confirmed=True) duty_people = RelationshipDutyPeople.find_or_create( company_id=token.selected_company.company_id, duties_id=bulk_duty.id, member_id=created_people.id, ) duty_people.update(is_confirmed=True) return created_people class RelationshipEmployee2PostCode(CrudCollection): """ Build2EmployeeRelationship class based on declarative_base and BaseMixin via session """ __tablename__ = "relationship_employee2postcode" __exclude__fields__ = [] __include__fields__ = [] __language_model__ = RelationshipEmployee2PostCodeLanguageModel company_id: Mapped[int] = mapped_column( ForeignKey("companies.id"), nullable=True ) # 1, 2, 3 employee_id: Mapped[int] = mapped_column(ForeignKey("employees.id"), nullable=False) member_id: Mapped[int] = mapped_column( ForeignKey("address_postcode.id"), nullable=False ) relationship_type: Mapped[str] = mapped_column( String, nullable=True, server_default="Employee" ) # Commercial show_only: Mapped[bool] = mapped_column(Boolean, server_default="0") __table_args__ = ({"comment": "Build2Employee Relationship Information"},) class AddressPostcode(CrudCollection, SelectActionWithEmployee): """ Postcode class based on declarative_base and BaseMixin via session """ __tablename__ = "address_postcode" __exclude__fields__ = [] __access_by__ = [] __many__table__ = RelationshipEmployee2PostCode __language_model__ = AddressPostcodeLanguageModel street_id: Mapped[int] = mapped_column(ForeignKey("address_street.id")) street_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="Street UUID" ) postcode: Mapped[str] = mapped_column( String(32), nullable=False, comment="Postcode" ) __table_args__ = ({"comment": "Postcode Information"},) class Addresses(CrudCollection): """ Address class based on declarative_base and BaseMixin via session """ __tablename__ = "addresses" __exclude__fields__ = [] __language_model__ = AddressesLanguageModel build_number: Mapped[str] = mapped_column( String(24), nullable=False, comment="Build Number" ) door_number: Mapped[str] = mapped_column( String(24), nullable=True, comment="Door Number" ) floor_number: Mapped[str] = mapped_column( String(24), nullable=True, comment="Floor Number" ) comment_address: Mapped[str] = mapped_column( String, nullable=False, comment="Address" ) letter_address: Mapped[str] = mapped_column( String, nullable=False, comment="Address" ) short_letter_address: Mapped[str] = mapped_column( String, nullable=False, comment="Address" ) latitude: Mapped[float] = mapped_column(Numeric(20, 12), server_default="0") longitude: Mapped[float] = mapped_column(Numeric(20, 12), server_default="0") street_id: Mapped[int] = mapped_column( ForeignKey("address_street.id"), nullable=False ) street_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="Street UUID" ) @classmethod def list_via_employee(cls, token_dict, filter_expr=None): post_code_list = RelationshipEmployee2PostCode.filter_all( RelationshipEmployee2PostCode.employee_id == token_dict.selected_company.employee_id, ).data post_code_id_list = [post_code.member_id for post_code in post_code_list] if not post_code_id_list: raise HTTPException( status_code=404, detail="User has no post code registered. User can not list addresses.", ) cls.pre_query = cls.filter_all(cls.post_code_id.in_(post_code_id_list)).query filter_cls = cls.filter_all(*filter_expr or []) cls.pre_query = None return filter_cls.data # buildings: Mapped["Build"] = relationship( # "Build", back_populates="addresses", foreign_keys="Build.address_id" # ) # site: Mapped["BuildSites"] = relationship( # "BuildSites", back_populates="addresses", foreign_keys="BuildSites.address_id" # ) # official_companies: Mapped["Companies"] = relationship( # "Company", # back_populates="official_address", # foreign_keys="Company.official_address_id", # ) # @classmethod # def create_action(cls, request, create_address: InsertAddress): # from services.redis.auth_actions.token import parse_token_object_to_dict # # token_dict = parse_token_object_to_dict(request=request) # data_dict = create_address.model_dump() # post_code = AddressPostcode.find_one(uu_id=create_address.post_code_uu_id) # if not post_code: # raise HTTPException( # status_code=404, # detail="Post code not found.", # ) # if Employee2AddressRelationship.post_code_id.find_one( # employee_id=token_dict.selected_company.employee_id, # post_code_id=post_code.id, # ): # data_dict["post_code_id"] = post_code.id # del data_dict["post_code_uu_id"] # return cls.find_or_create(**create_address.model_dump()) # raise HTTPException( # status_code=401, # detail=f"User is not qualified to create address at this post code {post_code.postcode}", # ) # __table_args__ = ( # Index("_address_ndx_00", country_code, b_state, city, district), # {"comment": "Address Information"}, # ) class AddressGeographicLocations(CrudCollection): """ Country class based on declarative_base and BaseMixin via session """ __tablename__ = "address_geographic_locations" __exclude__fields__ = [] __language_model__ = AddressGeographicLocationsLanguageModel geo_table: Mapped[str] = mapped_column( String, nullable=False, comment="Address Table Name" ) geo_id: Mapped[int] = mapped_column( Integer, nullable=False, comment="Address Table ID" ) geo_name: Mapped[str] = mapped_column( String, nullable=False, comment="Geographic Location Name" ) geo_latitude: Mapped[float] = mapped_column( Numeric(20, 6), server_default="0", comment="Geographic Location Name" ) geo_longitude: Mapped[float] = mapped_column( Numeric(20, 6), server_default="0", comment="Geographic Location Latitude" ) geo_altitude: Mapped[float] = mapped_column( Numeric(20, 6), server_default="0", comment="Geographic Location Longitude" ) geo_description: Mapped[str] = mapped_column( Text, nullable=False, comment="Geographic Location Description" ) geo_area_size: Mapped[float] = mapped_column( Numeric(20, 2), nullable=True, server_default="0", comment="Geographic Location Area Size", ) geo_population: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Geographic Location Population" ) # geo_geom_point = mapped_column(Geometry('POINT', srid=4326), nullable=True, comment="Geographic Location Points") # geo_geom_polygon = mapped_column(Geometry('POLYGON', srid=4326), nullable=True, # comment="Geographic Location Vector geographic information (polygon)") # geo_centroid = mapped_column( GEOMETRY(POINT, 4326), nullable=True, # comment="Geographic Location center of gravity of the region(points)") __table_args__ = ( Index("_address_geographic_locations_ndx_00", geo_table, geo_id), Index("_address_geographic_locations_ndx_01", geo_latitude, geo_longitude), {"comment": "Geographic Location Information"}, ) class AddressCountry(CrudCollection): """ Country class based on declarative_base and BaseMixin via session """ __tablename__ = "address_country" __exclude__fields__ = [] __language_model__ = AddressCountryLanguageModel country_code: Mapped[str] = mapped_column( String(16), nullable=False, comment="Country Code" ) country_name: Mapped[str] = mapped_column( String, nullable=False, comment="Country Name" ) money_code: Mapped[str] = mapped_column( String(12), nullable=True, comment="Money Code" ) language: Mapped[str] = mapped_column( String, nullable=True, comment="Language Code" ) address_geographic_id: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Address Geographic Id" ) __table_args__ = ( Index("_address_country_ndx_00", money_code), Index("_address_country_ndx_01", country_code, unique=True), {"comment": "Country Information"}, ) class AddressState(CrudCollection): """ State class based on declarative_base and BaseMixin via session """ __tablename__ = "address_state" __exclude__fields__ = [] __language_model__ = AddressStateLanguageModel state_code: Mapped[str] = mapped_column( String(16), nullable=False, comment="State Code" ) state_name: Mapped[str] = mapped_column( String, nullable=False, comment="State Name" ) licence_plate: Mapped[str] = mapped_column( String(24), nullable=True, comment="Sign Code" ) phone_code: Mapped[str] = mapped_column( String(36), nullable=True, comment="Phone Code" ) gov_code: Mapped[str] = mapped_column( String(128), nullable=True, comment="Government Code" ) address_geographic_id: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Address Geographic Id" ) country_id: Mapped[int] = mapped_column(ForeignKey("address_country.id")) country_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="Country UUID" ) __table_args__ = ( Index( "_address_state_ndx_01", country_id, state_code, unique=True, ), {"comment": "State Information"}, ) class AddressCity(CrudCollection): """ City class based on declarative_base and BaseMixin via session """ __tablename__ = "address_city" __exclude__fields__ = [] __language_model__ = AddressCityLanguageModel city_code: Mapped[str] = mapped_column( String(24), nullable=False, comment="City Code" ) city_name: Mapped[str] = mapped_column(String, nullable=False, comment="City Name") licence_plate: Mapped[str] = mapped_column( String(24), nullable=True, comment="Sign Code" ) phone_code: Mapped[str] = mapped_column( String(36), nullable=True, comment="Phone Code" ) gov_code: Mapped[str] = mapped_column( String(128), nullable=True, comment="Government Code" ) address_geographic_id: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Address Geographic Id" ) state_id: Mapped[int] = mapped_column(ForeignKey("address_state.id")) state_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="State UUID" ) __table_args__ = ( Index( "_address_city_ndx_01", state_id, city_code, unique=True, ), {"comment": "City Information"}, ) class AddressDistrict(CrudCollection): """ District class based on declarative_base and BaseMixin via session """ __tablename__ = "address_district" __exclude__fields__ = [] __language_model__ = AddressDistrictLanguageModel district_code: Mapped[str] = mapped_column( String(16), nullable=False, comment="District Code" ) district_name: Mapped[str] = mapped_column( String, nullable=False, comment="District Name" ) phone_code: Mapped[str] = mapped_column( String(36), nullable=True, comment="Phone Code" ) gov_code: Mapped[str] = mapped_column( String(128), nullable=True, comment="Government Code" ) address_geographic_id: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Address Geographic Id" ) city_id: Mapped[int] = mapped_column( ForeignKey("address_city.id"), nullable=False, comment="City ID" ) city_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="City UUID" ) __table_args__ = ( Index( "_address_district_ndx_01", city_id, district_code, unique=True, ), {"comment": "District Information"}, ) class AddressLocality(CrudCollection): """ Locality class based on declarative_base and BaseMixin via session """ __tablename__ = "address_locality" __exclude__fields__ = [] __language_model__ = AddressLocalityLanguageModel locality_code: Mapped[str] = mapped_column( String(16), nullable=False, comment="Locality Code" ) locality_name: Mapped[str] = mapped_column( String, nullable=False, comment="Locality Name" ) type_code: Mapped[str] = mapped_column(String, nullable=True, comment="Type Name") type_description: Mapped[str] = mapped_column( String, nullable=True, comment="Type Name" ) gov_code: Mapped[str] = mapped_column( String(128), nullable=True, comment="Government Code" ) address_show: Mapped[bool] = mapped_column(Boolean, server_default="1") address_geographic_id: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Address Geographic Id" ) district_id: Mapped[int] = mapped_column( ForeignKey("address_district.id"), nullable=False, comment="District ID" ) district_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="District UUID" ) __table_args__ = ( Index( "_address_locality_ndx_01", district_id, locality_code, unique=True, ), {"comment": "Locality Information"}, ) class AddressNeighborhood(CrudCollection): """ Neighborhood class based on declarative_base and BaseMixin via session """ __tablename__ = "address_neighborhood" __exclude__fields__ = [] __language_model__ = AddressNeighborhoodLanguageModel neighborhood_code: Mapped[str] = mapped_column( String(16), nullable=False, comment="Neighborhood Code" ) neighborhood_name: Mapped[str] = mapped_column( String, nullable=False, comment="Neighborhood Name" ) type_code: Mapped[str] = mapped_column(String, nullable=True, comment="Type Name") type_description: Mapped[str] = mapped_column( String, nullable=True, comment="Type Name" ) gov_code: Mapped[str] = mapped_column( String(128), nullable=True, comment="Government Code" ) address_show: Mapped[bool] = mapped_column(Boolean, server_default="1") address_geographic_id: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Address Geographic Id" ) district_id: Mapped[int] = mapped_column( ForeignKey("address_district.id"), nullable=True, comment="District ID" ) district_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="District UUID" ) locality_id: Mapped[int] = mapped_column( ForeignKey("address_locality.id"), nullable=True, comment="Locality ID" ) locality_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="Locality UUID" ) __table_args__ = ( Index( "_address_neighborhood_ndx_01", locality_id, neighborhood_code, unique=True, ), {"comment": "Neighborhood Information"}, ) class AddressStreet(CrudCollection): """ Street class based on declarative_base and BaseMixin via session """ __tablename__ = "address_street" __exclude__fields__ = [] __language_model__ = AddressStreetLanguageModel street_code: Mapped[str] = mapped_column( String(16), nullable=False, comment="Street Code" ) street_name: Mapped[str] = mapped_column( String, nullable=False, comment="Street Name" ) type_code: Mapped[str] = mapped_column(String, nullable=True, comment="Type Name") type_description: Mapped[str] = mapped_column( String, nullable=True, comment="Type Name" ) gov_code: Mapped[str] = mapped_column( String(128), nullable=True, comment="Government Code" ) address_geographic_id: Mapped[int] = mapped_column( BigInteger, nullable=True, comment="Address Geographic Id" ) neighborhood_id: Mapped[int] = mapped_column( ForeignKey("address_neighborhood.id"), nullable=False, comment="Neighborhood ID" ) neighborhood_uu_id: Mapped[str] = mapped_column( String, server_default="", comment="Neighborhood UUID" ) __table_args__ = ( Index("_address_street_ndx_01", neighborhood_id, street_code, unique=True), {"comment": "Street Information"}, ) @classmethod def search_address_text(cls, search_text, token_dict=None): field_dict = { "AddressStreet.uu_id": cls.uu_id, "AddressCountry.uu_id": AddressCountry.uu_id, "AddressState.uu_id": AddressState.uu_id, "AddressCity.uu_id": AddressCity.uu_id, "AddressDistrict.uu_id": AddressDistrict.uu_id, "AddressLocality.uu_id": AddressLocality.uu_id, "AddressNeighborhood.uu_id": AddressNeighborhood.uu_id, "AddressCountry.country_name": AddressCountry.country_name, "AddressState.state_name": AddressState.state_name, "AddressCity.city_name": AddressCity.city_name, "AddressDistrict.district_name": AddressDistrict.district_name, "AddressLocality.locality_name": AddressLocality.locality_name, "AddressNeighborhood.neighborhood_name": AddressNeighborhood.neighborhood_name, "AddressStreet.street_name": cls.street_name, } joined_data = ( cls.session.query(*list(field_dict.values())) .select_from(cls) .join(AddressNeighborhood, AddressNeighborhood.id == cls.neighborhood_id) .join( AddressLocality, AddressLocality.id == AddressNeighborhood.locality_id ) .join(AddressDistrict, AddressDistrict.id == AddressLocality.district_id) .join(AddressCity, AddressCity.id == AddressDistrict.city_id) .join(AddressState, AddressState.id == AddressCity.state_id) .join(AddressCountry, AddressCountry.id == AddressState.country_id) .filter( or_( AddressNeighborhood.neighborhood_name.ilike( f"%{str(search_text).upper()}%" ), AddressLocality.locality_name.ilike( f"%{str(search_text).upper()}%" ), AddressDistrict.district_name.ilike( f"%{str(search_text).upper()}%" ), # AddressCity.city_name.ilike(f"%{str(search_text).upper()}%"), # AddressState.state_name.ilike(f"%{str(search_text).upper()}%"), # AddressCountry.country_name.ilike(f"%{str(search_text).upper()}%"), cls.street_name.ilike(f"%{str(search_text).upper()}%"), ), ) ) # select([mytable.c.id]).where( # func.to_tsvector('english', mytable.c.title) \ # .match('somestring', postgresql_regconfig='english') # ) joined_statement = joined_data joined_data = joined_data.first() if not joined_data: raise HTTPException( status_code=404, detail="No address found with the given search text.", ) return dict( query=joined_statement, schema=list(field_dict.keys()), ) class OccupantTypes(CrudCollection): """ Occupant Types class based on declarative_base and BaseMixin via session """ __tablename__ = "occupant_types" __exclude__fields__ = [] __language_model__ = OccupantTypesLanguageModel 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"},) @classmethod def get_manager_occupant_type(cls): if occupant_types := cls.filter_all( cls.occupant_is_unique == True, cls.occupant_category_type == "MT" ).data: return [occupant.uu_id.__str__() for occupant in occupant_types] raise HTTPException( status_code=404, detail="No manager type found.", ) class Contracts(CrudCollection): """ Contract class based on declarative_base and BaseMixin via session """ __tablename__ = "contracts" __exclude__fields__ = [] __language_model__ = ContractsLanguageModel 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): # from api_library.date_time_actions.date_functions import system_arrow # 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"}, )