wag-managment-api-service-v.../databases/sql_models/company/company.py

577 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from fastapi.exceptions import HTTPException
from databases.sql_models.core_mixin import CrudCollection
from sqlalchemy import String, Integer, Boolean, ForeignKey, Index, Identity
from sqlalchemy.orm import mapped_column, Mapped
from api_configs import RelationAccess
from databases.extensions import SelectAction
from api_validations.validations_request import (
InsertCompany,
UpdateCompany,
MatchCompany2Company,
)
from api_objects.auth.token_objects import EmployeeTokenObject
class RelationshipDutyCompany(CrudCollection):
"""
CompanyRelationship class based on declarative_base and CrudCollection via session
Company -> Sub Company -> Sub-Sub Company
if owner_id == parent_id: can manipulate data of any record
else: Read-Only
duty_id = if relationship_type == base An organization / not operational / no responsible person
relationship = company_id filter -> Action filter(company_id) relationship_type = Organization
relationship = company_id filter -> Action filter(company_id) relationship_type = Commercial
"""
__tablename__ = "relationship_duty_company"
__exclude__fields__ = []
__access_by__ = RelationAccess.SuperAccessList
owner_id: Mapped[int] = mapped_column(
ForeignKey("companies.id"), nullable=False
) # 1
duties_id: Mapped[int] = mapped_column(
ForeignKey("duties.id"), nullable=False
) # duty -> (n)employee Evyos LTD
member_id: Mapped[int] = mapped_column(
ForeignKey("companies.id"), nullable=False
) # 2, 3, 4
parent_id: Mapped[int] = mapped_column(
ForeignKey("companies.id"), nullable=True
) # None
relationship_type: Mapped[str] = mapped_column(
String, nullable=True, server_default="Commercial"
) # Commercial, Organization # Bulk
child_count: Mapped[int] = mapped_column(Integer) # 0
show_only: Mapped[bool] = mapped_column(Boolean, server_default="0")
# related_company: Mapped[List["Companies"]] = relationship(
# "Companies",
# back_populates="related_companies",
# foreign_keys=[related_company_id],
# )
@classmethod
def match_company_to_company_commercial(cls, data: MatchCompany2Company, token):
from databases import (
Duties,
)
token_duties_id, token_company_id = token.get("duty_id"), token.get(
"company_id"
)
list_match_company_id = []
send_duties = Duties.filter_one(
Duties.uu_id==data.duty_uu_id,
*Duties.valid_record_args(Duties),
)
send_user_duties = Duties.filter_one(
Duties.duties_id==send_duties.id,
Duties.company_id==token_duties_id,
*Duties.valid_record_args(Duties),
)
if not send_user_duties:
raise Exception(
"Send Duty is not found in company. Please check duty uuid and try again."
)
for company_uu_id in list(data.match_company_uu_id):
company = Companies.filter_one(
Companies.uu_id==company_uu_id,
*Companies.valid_record_args(Companies),
)
bulk_company = RelationshipDutyCompany.filter_one(
RelationshipDutyCompany.owner_id==token_company_id,
RelationshipDutyCompany.relationship_type=="Bulk",
RelationshipDutyCompany.member_id==company.id,
*RelationshipDutyCompany.valid_record_args(RelationshipDutyCompany),
)
if not bulk_company:
raise Exception(
f"Bulk Company is not found in company. "
f"Please check company uuid {bulk_company.uu_id} and try again."
)
list_match_company_id.append(bulk_company)
for match_company_id in list_match_company_id:
RelationshipDutyCompany.find_or_create(
owner_id=token_company_id,
duties_id=send_user_duties.id,
member_id=match_company_id.id,
parent_id=match_company_id.parent_id,
relationship_type="Commercial",
show_only=False,
)
@classmethod
def match_company_to_company_organization(cls, data: MatchCompany2Company, token):
from databases import (
Duties,
)
token_duties_id, token_company_id = token.get("duty_id"), token.get(
"company_id"
)
list_match_company_id = []
send_duties = Duties.filter_one(
Duties.uu_id==data.duty_uu_id,
*Duties.valid_record_args(Duties),
)
send_user_duties = Duties.filter_one(
Duties.duties_id==send_duties.id,
Duties.company_id==token_duties_id,
*Duties.valid_record_args(Duties),
)
if not send_user_duties:
raise Exception(
"Send Duty is not found in company. Please check duty uuid and try again."
)
for company_uu_id in list(data.match_company_uu_id):
company = Companies.filter_one(
Companies.uu_id==company_uu_id,
*Companies.valid_record_args(Companies),
)
bulk_company = RelationshipDutyCompany.filter_one(
RelationshipDutyCompany.owner_id==token_company_id,
RelationshipDutyCompany.relationship_type=="Bulk",
RelationshipDutyCompany.member_id==company.id,
*RelationshipDutyCompany.valid_record_args(RelationshipDutyCompany),
)
if not bulk_company:
raise Exception(
f"Bulk Company is not found in company. "
f"Please check company uuid {bulk_company.uu_id} and try again."
)
list_match_company_id.append(bulk_company)
for match_company_id in list_match_company_id:
Duties.init_a_company_default_duties(
company_id=match_company_id.id,
company_uu_id=str(match_company_id.uu_id),
)
RelationshipDutyCompany.find_or_create(
owner_id=token_company_id,
duties_id=send_user_duties.id,
member_id=match_company_id.id,
parent_id=match_company_id.parent_id,
relationship_type="Organization",
show_only=False,
)
__table_args__ = (
Index(
"_company_relationship_ndx_01",
duties_id,
owner_id,
member_id,
relationship_type,
unique=True,
),
{"comment": "Company Relationship Information"},
)
class Companies(CrudCollection, SelectAction):
"""
Company class based on declarative_base and CrudCollection via session
formal_name = Government register name by offical
public_name = Public registered name by User
nick_name = Search by nickname, commercial_type = Tüzel veya birey
"""
__tablename__ = "companies"
__exclude__fields__ = ["is_blacklist", "is_commercial"]
__access_by__ = []
__many__table__ = RelationshipDutyCompany
# __explain__ = AbstractCompany()
formal_name: Mapped[str] = mapped_column(
String, nullable=False, comment="Formal Name"
)
company_type: Mapped[str] = mapped_column(
String, nullable=False, comment="Company Type"
)
commercial_type: Mapped[str] = mapped_column(
String, nullable=False, comment="Commercial Type"
)
tax_no: Mapped[str] = mapped_column(
String, index=True, unique=True, nullable=False, comment="Tax No"
)
public_name: Mapped[str] = mapped_column(String, comment="Public Name of a company")
company_tag: Mapped[str] = mapped_column(String, comment="Company Tag")
default_lang_type: Mapped[str] = mapped_column(String, server_default="TR")
default_money_type: Mapped[str] = mapped_column(String, server_default="TL")
is_commercial: Mapped[bool] = mapped_column(Boolean, server_default="False")
is_blacklist: Mapped[bool] = mapped_column(Boolean, server_default="False")
parent_id = mapped_column(Integer, nullable=True)
workplace_no: Mapped[str] = mapped_column(String, nullable=True)
official_address_id: Mapped[int] = mapped_column(
ForeignKey("addresses.id"), nullable=True
)
official_address_uu_id: Mapped[str] = mapped_column(
String, nullable=True, comment="Official Address UUID"
)
top_responsible_company_id: Mapped[int] = mapped_column(
ForeignKey("companies.id"), nullable=True
)
top_responsible_company_uu_id: Mapped[str] = mapped_column(
String, nullable=True, comment="Top Responsible Company UUID"
)
# buildings: Mapped[List["Build"]] = relationship(
# "Build",
# back_populates="companies",
# foreign_keys="Build.company_id",
# )
__table_args__ = (
Index("_company_ndx_01", tax_no, unique=True),
Index("_company_ndx_02", formal_name, public_name),
{"comment": "Company Information"},
)
@classmethod
def create_action(cls, data: InsertCompany, token: EmployeeTokenObject):
from databases import Addresses, Duties
data_dict = data.model_dump()
if cls.filter_one(cls.tax_no == str(data.tax_no).strip(), system=True).data:
raise HTTPException(
status_code=400,
detail="Company already exists. Please ask supervisor to make company visible for your duty."
)
official_address = Addresses.filter_one(
Addresses.uu_id == data.official_address_uu_id,
*Addresses.valid_record_args(Addresses),
).data
# if not official_address:
# raise HTTPException(
# status_code=400,
# detail="Official address is not found. Please check address uuid and try again.",
# )
bulk_duties = Duties.get_bulk_duties_of_a_company(
company_id=token.selected_company.company_id
)
if official_address:
data_dict["official_address_id"] = official_address.id
data_dict["official_address_uu_id"] = str(official_address.uu_id)
data_dict["parent_id"] = token.selected_company.company_id
data_dict["top_responsible_company_id"] = token.selected_company.company_id
data_dict["top_responsible_company_uu_id"] = (
token.selected_company.company_uu_id
)
company_created = cls.find_or_create(**data_dict)
RelationshipDutyCompany.find_or_create(
owner_id=token.selected_company.company_id,
duties_id=bulk_duties.id,
member_id=company_created.id,
parent_id=company_created.parent_id,
child_count=0,
relationship_type="Bulk",
show_only=False,
)
return company_created
@classmethod
def update_action(cls, data: UpdateCompany, token):
from databases import (
Addresses,
)
data_dict = data.excluded_dump()
duty_id = token.get("duty_id")
company_id = token.get("company_id")
if data.official_address_uu_id:
official_address = Addresses.filter_one(
Addresses.uu_id==data.official_address_uu_id,
*Addresses.valid_record_args(Addresses),
).data
data_dict["official_address_id"] = official_address.id
del data_dict["official_address_uu_id"], data_dict["company_uu_id"]
company_to_update = cls.select_action(
duty_id_list=[duty_id],
filter_expr=[
cls.uu_id == data.company_uu_id,
RelationshipDutyCompany.parent_id == company_id,
],
)
return company_to_update.update(**data_dict)
# parent_id = mapped_column(ForeignKey("companies.id"))
# if data.parent_uu_id:
# company = Companies.find_one(uu_id=data.parent_uu_id)
# data_dict["parent_id"] = company.id
# def is_access_valid(self, endpoint_ext: str):
# try:
# if (
# not arrow.get(self.stop_date)
# > arrow.utcnow()
# > arrow.get(self.start_date)
# ):
# message = f"Kullanıcı yetkileri süresi dolmuştur. {self.endpoint_name} için supervisor ile görüşünüz."
# SystemLogs.create_log(
# log_type="ERROR",
# log_code="ACCESS_EXPIRED",
# log_action=self.__tablename__,
# log_message=message,
# )
# return False
# except Exception as e:
# SystemLogs.create_log(
# log_type="ERROR",
# log_code="ACCESS_EXPIRED",
# log_action=self.__tablename__,
# log_message=e,
# )
# return False
#
# access_dict = {
# "LIST": self.access_read,
# "INSERT": self.access_write,
# "UPDATE": self.access_update,
# "DELETE": self.access_delete,
# "ACTIVE": self.access_update,
# "PRINT": self.report_print,
# "EXPORT": self.report_export,
# }
# return access_dict.get(endpoint_ext.upper(), False)
# official_address: Mapped[List["Address"]] = relationship(
# "Address",
# back_populates="official_companies",
# foreign_keys=[official_address_id],
# )
#
# emails: Mapped[List["UsersEmails"]] = relationship(
# "UsersEmails", back_populates="companies", foreign_keys="UsersEmails.company_id"
# )
# phones: Mapped[List["UsersPhones"]] = relationship(
# "UsersPhones", back_populates="company", foreign_keys="UsersPhones.company_id"
# )
# buildings: Mapped[List["Build"]] = relationship(
# "Build",
# back_populates="companies",
# foreign_keys="Build.company_id",
# )
# response_buildings: Mapped[List["Build"]] = relationship(
# "Build",
# back_populates="response_companies",
# foreign_keys="Build.response_company_id",
# )
# departments: Mapped[List["CompanyDepartments"]] = relationship(
# "CompanyDepartments",
# back_populates="company",
# foreign_keys="CompanyDepartments.company_id",
# )
# budget_records: Mapped[List["CompanyBudgetRecords"]] = relationship(
# "CompanyBudgetRecords",
# back_populates="companies",
# foreign_keys="CompanyBudgetRecords.company_id",
# )
# send_budget_records: Mapped[List["CompanyBudgetRecords"]] = relationship(
# "CompanyBudgetRecords",
# back_populates="send_companies",
# foreign_keys="CompanyBudgetRecords.send_company_id",
# )
# decision_books: Mapped[List["BuildDecisionBook"]] = relationship(
# "BuildDecisionBook",
# back_populates="companies",
# foreign_keys="BuildDecisionBook.resp_company_id",
# )
# decision_book_projects: Mapped[List["BuildDecisionBookProjects"]] = relationship(
# "BuildDecisionBookProjects",
# back_populates="companies",
# foreign_keys="BuildDecisionBookProjects.resp_company_id",
# )
# decision_book_legal: Mapped["BuildDecisionBookLegal"] = relationship(
# "BuildDecisionBookLegal",
# back_populates="attorney_companies",
# foreign_keys="BuildDecisionBookLegal.resp_attorney_company",
# )
#
# company_account_books: Mapped["AccountBooks"] = relationship(
# "AccountBooks",
# back_populates="company",
# foreign_keys="AccountBooks.company_id",
# )
# branch_account_books: Mapped["AccountBooks"] = relationship(
# "AccountBooks",
# back_populates="branch",
# foreign_keys="AccountBooks.branch_id",
# )
# account_codes: Mapped["AccountCodes"] = relationship(
# "AccountCodes", back_populates="company", foreign_keys="AccountCodes.company_id"
# )
# search_iban_description: Mapped["BuildIbanDescription"] = relationship(
# "BuildIbanDescription",
# back_populates="company",
# foreign_keys="BuildIbanDescription.company_id",
# )
# related_companies: Mapped[List["CompanyRelationship"]] = relationship(
# "CompanyRelationship",
# back_populates="related_company",
# foreign_keys="CompanyRelationship.related_company_id",
# )
#
# class AbstractCompany:
# """
# Abstract and explanation of Company class for end-user guide
# """
#
# formal_name = Explanation(
# explanation="Devletin resmi kayıtlarında bulunan şirket ünvanıdır.",
# usage="Devletin resmi kayıtlarında bulunan şirket adı istendiğinde kullanılır.",
# alias="Resmi Ünvan",
# example=["X Şirketi LTD", "Y Şirketi A.Ş."],
# )
# company_type = Explanation(
# explanation="Şirketin türüdür.",
# usage="Şirketin türü istendiğinde kullanılır.",
# alias="Şirket Türü",
# example=[
# "Şahıs",
# "Limited",
# "Anonim",
# "Kolektif",
# "Komandit",
# "Kooperatif",
# "Serbest Meslek",
# "Adi Ortaklık",
# ],
# )
# commercial_type = Explanation(
# explanation="Şirketin ticari türüdür.",
# usage="Şirketin ticari türü istendiğinde kullanılır.",
# alias="Ticari Tür",
# example=["Tüzel", "Birey"],
# )
# tax_no = Explanation(
# explanation="Şirketin vergi numarasıdır.",
# usage="Şirketin vergi numarası istendiğinde kullanılır.",
# alias="Vergi No",
# example=["1234567890"],
# )
# public_name = Explanation(
# explanation="Şirketin kamuoyunda bilinen adıdır.",
# usage="Şirketin kamuoyunda bilinen adı istendiğinde kullanılır.",
# alias="Piyasada Bilinen Adı",
# example=["X Şirketi", "Y Şirketi"],
# )
# company_tag = Explanation(
# explanation="Şirketin takma adı veya etiketidir.",
# usage="Şirketin yöneticisin karar verdiği takma adı veya etiketi istendiğinde kullanılır.",
# alias="Şirket Etiketi veya Takma Adı",
# example=["X", "Y"],
# )
# default_lang_type = Explanation(
# explanation="Şirketin varsayılan dil türüdür.",
# usage="Şirketin varsayılan dil türü istendiğinde kullanılır.",
# alias="Şirketin Dil Türü",
# example=["TR", "EN"],
# )
# default_money_type = Explanation(
# explanation="Şirketin varsayılan para birimi türüdür.",
# usage="Şirketin varsayılan para birimi türü istendiğinde kullanılır.",
# alias="Şirketin Para Birimi Türü",
# example=["TL", "USD", "EUR"],
# )
# is_commercial = Explanation(
# explanation="Şirketin ticari olup olmadığını belirtir.",
# usage="Şirketin ticari olup olmadığını applikasyonun anlaması için kullanılır.",
# condition=lambda commercial_type: True if commercial_type == "Şahıs" else False,
# alias="Şirket Ticari mi?",
# )
# is_blacklist = Explanation(
# explanation="Şirketin kara listeye alınıp alınmadığını belirtir.",
# usage="Şirketin kara listeye alınıp alınmadığını applikasyonun anlaması için kullanılır.",
# alias="Kara Listeye alınsın mı?",
# example=[True, False],
# )
# parent_id = Explanation(
# explanation="Şirketin sorumlu olduğu şirketin ID'sidir.",
# usage="Şirketin sorumlu olduğu şirketin ID'si istendiğinde kullanılır.",
# alias="Sorumlu Şirket",
# example=[
# "Bir şirketin sorumlu şirketi hangisi olduğunu bulmak için kullanılır.",
# ],
# )
# workplace_no = Explanation(
# explanation="Şirketin iş yeri numarasıdır.",
# usage="Şirketin iş yeri numarası istendiğinde kullanılır.",
# alias="İş Yeri No",
# example=["1234567890"],
# )
# official_address_id = Explanation(
# explanation="Şirketin resmi adresidi.",
# usage="Şirketin resmi adresinin ne olduğunu bulmak için kullanılır.",
# alias="Resmi Adres",
# example=[
# "Bu şirketin adresi nedir sorusuna cevap vermek için kullanılır.",
# ],
# )
# top_responsible_company_id = Explanation(
# explanation="Şirketin en üst sorumlu şirketin ID'sidir.",
# usage="Şirketin en üst sorumlu şirketin hangisi olduğunu bulmak için kullanılır.",
# alias="Ana Yetkili Şirket",
# example=[
# "Bölge veya ülke genelinde en üst sorumlu şirketin hangisi olduğunu belirtmek için kullanılır.",
# ],
# )
# buildings = Explanation(
# explanation="Şirketin sahip olduğu binaların listesidir.",
# usage="Şirketin sahip olduğu binaların listesini bulmak için kullanılır.",
# alias="Sorumlu olduğu binalar Binalar",
# example=[
# "Şirketin sahip olduğu binaların listesini bulmak için kullanılır.",
# ],
# )
#
# def wag_create_company(self):
# """
# Er kişiye wag_create_company fonksiyonu = fieldları manipule edebilir?
# 78 ile oluşturulan bir user için wag_create_company fonksiyonu = fieldları manipule edebilir?
# """
# return {
# "commercial_type": self.commercial_type,
# "formal_name": self.formal_name,
# "public_name": self.public_name,
# "company_type": self.company_type,
# "tax_no": self.tax_no,
# "workplace_no": self.workplace_no,
# "company_tag": self.company_tag,
# "default_lang_type": self.default_lang_type,
# "default_money_type": self.default_money_type,
# "official_address_id": self.official_address_id,
# }
#
# def wag_update_company(self):
# return {
# "commercial_type": self.commercial_type,
# "formal_name": self.formal_name,
# "public_name": self.public_name,
# "company_type": self.company_type,
# "tax_no": self.tax_no,
# "workplace_no": self.workplace_no,
# "company_tag": self.company_tag,
# "default_lang_type": self.default_lang_type,
# "default_money_type": self.default_money_type,
# "official_address_id": self.official_address_id,
# }