238 lines
8.8 KiB
Python
238 lines
8.8 KiB
Python
from typing import Any
|
|
|
|
from sqlalchemy import (
|
|
String,
|
|
Integer,
|
|
Boolean,
|
|
ForeignKey,
|
|
Index,
|
|
Identity,
|
|
TIMESTAMP,
|
|
func,
|
|
)
|
|
from sqlalchemy.orm import mapped_column, relationship, Mapped
|
|
|
|
from Controllers.Postgres.mixin import CrudCollection
|
|
|
|
|
|
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__ = []
|
|
|
|
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: Any, token):
|
|
from Schemas import Duties
|
|
|
|
with cls.new_session() as db_session:
|
|
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, db=db_session
|
|
)
|
|
send_user_duties = Duties.filter_one(
|
|
Duties.duties_id == send_duties.id,
|
|
Duties.company_id == token_duties_id,
|
|
db=db_session,
|
|
)
|
|
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, db=db_session
|
|
)
|
|
bulk_company = RelationshipDutyCompany.filter_one(
|
|
RelationshipDutyCompany.owner_id == token_company_id,
|
|
RelationshipDutyCompany.relationship_type == "Bulk",
|
|
RelationshipDutyCompany.member_id == company.id,
|
|
db=db_session,
|
|
)
|
|
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,
|
|
db=db_session,
|
|
)
|
|
|
|
@classmethod
|
|
def match_company_to_company_organization(cls, data: Any, token):
|
|
from Schemas import Duties
|
|
|
|
with cls.new_session() as db_session:
|
|
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, db=db_session
|
|
)
|
|
send_user_duties = Duties.filter_one(
|
|
Duties.duties_id == send_duties.id,
|
|
Duties.company_id == token_duties_id,
|
|
db=db_session,
|
|
)
|
|
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, db=db_session
|
|
)
|
|
bulk_company = RelationshipDutyCompany.filter_one(
|
|
RelationshipDutyCompany.owner_id == token_company_id,
|
|
RelationshipDutyCompany.relationship_type == "Bulk",
|
|
RelationshipDutyCompany.member_id == company.id,
|
|
db=db_session,
|
|
)
|
|
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),
|
|
db=db_session,
|
|
)
|
|
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,
|
|
db=db_session,
|
|
)
|
|
|
|
__table_args__ = (
|
|
Index(
|
|
"_company_relationship_ndx_01",
|
|
duties_id,
|
|
owner_id,
|
|
member_id,
|
|
relationship_type,
|
|
unique=True,
|
|
),
|
|
{"comment": "Company Relationship Information"},
|
|
)
|
|
|
|
|
|
class Companies(CrudCollection):
|
|
"""
|
|
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
|
|
|
|
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"},
|
|
)
|