Untitled

avatar
unknown
python
a year ago
16 kB
3
Indexable
import uuid
from enum import Enum
from typing import List

from sqlalchemy import ARRAY
from sqlalchemy import Boolean
from sqlalchemy import Column
from sqlalchemy import ForeignKey
from sqlalchemy import Index
from sqlalchemy import Integer
from sqlalchemy import JSON
from sqlalchemy import String
from sqlalchemy import Table
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.types import DateTime

from snakesdk_orm.infrastructure.tables import BaseTable
from snakesdk_orm.infrastructure.tables import UtcNow


class TableName(str, Enum):
    G_LEAD = "pdms_lead"

    # Relations.
    LEAD_HEALTH_PLAN_RELATION = "pdms_lead_health_plan_relation"
    LEAD_BUSINESS_LINE_RELATION = "pdms_lead_business_line_relation"

    LOCATION_COUNTY_REGION_RELATION = "pdms_location_county_region_relation"
    LOCATION_STATE_REGION_RELATION = "pdms_location_state_region_relation"

    LEAD_LOCATION_STATE_RELATION = "pdms_lead_location_state_relation"
    LEAD_LOCATION_REGION_RELATION = "pdms_lead_location_region_relation"
    LEAD_LOCATION_COUNTY_RELATION = "pdms_lead_location_county_relation"

    TP_LEAD = "pdms_tp_lead"
    SP_LEAD = "pdms_sp_lead"
    IMPORT_TASK = "pdms_import_task"
    USER = "pdms_user"
    LEAD_HEALTH_PLAN = "pdms_lead_health_plan"
    LEAD_ADDRESS = "pdms_lead_address"
    TP_LEAD_SERVICE_DETAIL = "pdms_tp_lead_service_detail"
    LEAD_BUSINESS_LINE = "pdms_lead_business_line"
    LEAD_CHANGES = "pdms_lead_changes"
    TP_MOBILITY_TYPE = "pdms_tp_mobility_type"

    LOCATION_STATE = "pdms_location_state"
    LOCATION_LEAD_REGION = "pdms_location_lead_region"
    LOCATION_COUNTY = "pdms_location_county"

    LEAD_SP_SPECIALTY = "pdms_lead_sp_specialty"
    LEAD_SP_SUBSPECIALTY = "pdms_lead_sp_subspecialty"
    LEAD_SP_LEAD_SPECIALTY_RELATION = "pdms_sp_lead_specialty_relation"
    LEAD_SP_LEAD_SUBSPECIALTY_RELATION = "pdms_sp_lead_subspecialty_relation"


lead_health_plan_table = Table(
    TableName.LEAD_HEALTH_PLAN_RELATION.value,
    BaseTable.metadata,
    Column("lead_id", ForeignKey(f"{TableName.G_LEAD}.entity_id"), primary_key=True),
    Column("health_plan_id", ForeignKey(f"{TableName.LEAD_HEALTH_PLAN}.entity_id"), primary_key=True),
)

lead_business_lines_table = Table(
    TableName.LEAD_BUSINESS_LINE_RELATION.value,
    BaseTable.metadata,
    Column("lead_id", ForeignKey(f"{TableName.G_LEAD}.entity_id"), primary_key=True),
    Column("business_line_id", ForeignKey(f"{TableName.LEAD_BUSINESS_LINE}.entity_id"), primary_key=True),
)

lead_location_states_table = Table(
    TableName.LEAD_LOCATION_STATE_RELATION.value,
    BaseTable.metadata,
    Column("lead_id", ForeignKey(f"{TableName.G_LEAD}.entity_id"), primary_key=True),
    Column("location_state_id", ForeignKey(f"{TableName.LOCATION_STATE}.entity_id"), primary_key=True),
)

lead_location_regions_table = Table(
    TableName.LEAD_LOCATION_REGION_RELATION.value,
    BaseTable.metadata,
    Column("lead_id", ForeignKey(f"{TableName.G_LEAD}.entity_id"), primary_key=True),
    Column("location_region_id", ForeignKey(f"{TableName.LOCATION_LEAD_REGION}.entity_id"), primary_key=True),
)

lead_location_counties_table = Table(
    TableName.LEAD_LOCATION_COUNTY_RELATION.value,
    BaseTable.metadata,
    Column("lead_id", ForeignKey(f"{TableName.G_LEAD}.entity_id"), primary_key=True),
    Column("location_county_id", ForeignKey(f"{TableName.LOCATION_COUNTY}.entity_id"), primary_key=True),
)


sp_lead_specialty_table = Table(
    TableName.LEAD_SP_LEAD_SPECIALTY_RELATION.value,
    BaseTable.metadata,
    Column("lead_id", ForeignKey(f"{TableName.G_LEAD}.entity_id"), primary_key=True),
    Column("specialty_id", ForeignKey(f"{TableName.LEAD_SP_SPECIALTY}.entity_id"), primary_key=True),
)


sp_lead_subspecialty_table = Table(
    TableName.LEAD_SP_LEAD_SUBSPECIALTY_RELATION.value,
    BaseTable.metadata,
    Column("lead_id", ForeignKey(f"{TableName.G_LEAD}.entity_id"), primary_key=True),
    Column("sub_specialty_id", ForeignKey(f"{TableName.LEAD_SP_SUBSPECIALTY}.entity_id"), primary_key=True),
)

location_county_region_table = Table(
    TableName.LOCATION_COUNTY_REGION_RELATION.value,
    BaseTable.metadata,
    Column("county_id", ForeignKey(f"{TableName.LOCATION_COUNTY}.entity_id")),
    Column("region_id", ForeignKey(f"{TableName.LOCATION_LEAD_REGION}.entity_id")),
)


location_state_region_table = Table(
    TableName.LOCATION_STATE_REGION_RELATION.value,
    BaseTable.metadata,
    Column("state_id", ForeignKey(f"{TableName.LOCATION_STATE}.entity_id")),
    Column("region_id", ForeignKey(f"{TableName.LOCATION_LEAD_REGION}.entity_id")),
)


class UserTable(BaseTable):
    __tablename__ = TableName.USER.value

    username = Column(String(50), nullable=False, unique=True, comment="Username.")
    first_name = Column(String(50), nullable=False, comment="User firstname.")
    last_name = Column(String(50), nullable=False, comment="User lastname.")

    roles = Column(ARRAY(String), nullable=True, comment="User roles.")

    external_id = Column(String(50), nullable=True, unique=True, comment="User lastname.")

    is_active = Column(Boolean(), nullable=True, comment="True if user is active.")

    created_leads: Mapped[List["LeadTable"]] = relationship(
        back_populates="creator_user", foreign_keys="[LeadTable.creator_user_id]"
    )
    created_import_task_leads: Mapped[List["LeadImportTaskTable"]] = relationship(back_populates="creator_user")


class PdmBaseTable:
    @classmethod
    def get_field_names(cls):
        # TODO: move this to snakesdk_orm.
        return (item.name for item in cls.__table__.columns)


class LocationStateTable(BaseTable):
    __tablename__ = TableName.LOCATION_STATE.value

    name = Column(String(50), nullable=False)
    acronym = Column(String(2), nullable=False)


class LocationCountyTable(BaseTable):
    __tablename__ = TableName.LOCATION_COUNTY.value

    name = Column(String(50), nullable=False)
    state_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LOCATION_STATE.value}.entity_id"), nullable=False
    )
    regions = relationship("LocationLeadRegionTable", secondary=location_county_region_table, backref="counties")


class LocationLeadRegionTable(BaseTable):
    __tablename__ = TableName.LOCATION_LEAD_REGION.value

    name = Column(String(50), nullable=False)
    lead_type = Column(String(50), nullable=False)

    states = relationship("LocationStateTable", secondary=location_state_region_table, backref="regions")


class LeadHealthPlanTable(BaseTable):
    __tablename__ = TableName.LEAD_HEALTH_PLAN.value

    name = Column(String(50), nullable=False)


class LeadBusinessLineTable(BaseTable):
    __tablename__ = TableName.LEAD_BUSINESS_LINE.value

    name = Column(String(50), nullable=False)


class TpLeadMobilityTypeTable(BaseTable):
    __tablename__ = TableName.TP_MOBILITY_TYPE.value

    name = Column(String(50), nullable=False)


class LeadTable(PdmBaseTable, BaseTable):
    __tablename__ = TableName.G_LEAD.value

    company_name = Column(String(100), nullable=True)
    dba = Column(String(100), nullable=True)
    phone = Column(String(100), nullable=True, unique=False)

    tax_id = Column(String(100), nullable=True, unique=True)
    email = Column(String(100), nullable=True)
    contact_person = Column(String(100), nullable=True)

    is_medicaid = Column(Boolean(), nullable=True)
    is_medicare = Column(Boolean(), nullable=True)
    status = Column(String(100), nullable=True)
    status_reason = Column(String(100), nullable=True)

    pr_assigned_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.USER.value}.entity_id"), nullable=True)
    creator_user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.USER.value}.entity_id"), nullable=True)
    last_update_by_user_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.USER.value}.entity_id"), nullable=True
    )

    pr_assigned_user: Mapped["UserTable"] = relationship("UserTable", foreign_keys=[pr_assigned_id])
    creator_user: Mapped["UserTable"] = relationship("UserTable", foreign_keys=[creator_user_id])

    comments = Column(String(1000), nullable=True)

    lead_type = Column(String(50), nullable=True)

    last_status_changed_at = Column(DateTime, server_default=UtcNow())

    contact_title = Column(String(50), nullable=True)
    npi = Column(String(50), nullable=True)

    last_update_by_roles = Column(ARRAY(String), nullable=True, comment="User roles.")

    health_plans: Mapped[List[LeadHealthPlanTable]] = relationship(secondary=lead_health_plan_table)
    business_lines: Mapped[List[LeadBusinessLineTable]] = relationship(secondary=lead_business_lines_table)

    aggregate_groups = Column(ARRAY(String), nullable=True)

    source = Column(String(50), nullable=True)

    addresses: Mapped[List["LeadAddressTable"]] = relationship()

    location_states: Mapped[List[LocationStateTable]] = relationship(secondary=lead_location_states_table)
    location_counties: Mapped[List[LocationCountyTable]] = relationship(secondary=lead_location_counties_table)
    location_regions: Mapped[List[LocationLeadRegionTable]] = relationship(secondary=lead_location_regions_table)

    __mapper_args__ = {
        "polymorphic_identity": "lead",
        "polymorphic_on": "lead_type",
    }


class TPLeadTable(LeadTable):
    __tablename__ = TableName.TP_LEAD.value

    entity_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.G_LEAD.value}.entity_id"), primary_key=True)

    services = Column(ARRAY(String))
    drivers_count = Column(Integer(), nullable=True)
    vehicles_count = Column(Integer(), nullable=True)

    is_mba_qualified = Column(Boolean(), nullable=True, comment="True if TP lead is MBA qualified.")
    is_mba_certified = Column(Boolean(), nullable=True, comment="True if TP lead is MBA certified.")

    working_hours = Column(JSON, nullable=True)
    digital_platform_type = Column(String(50), nullable=True)

    credential_details = Column(JSON, nullable=True)

    service_details: Mapped[List["TPLeadServiceDetailTable"]] = relationship()

    __mapper_args__ = {
        "polymorphic_identity": "TRANPORTATION",
    }


class SpLeadSpecialtyTable(BaseTable):
    __tablename__ = TableName.LEAD_SP_SPECIALTY.value

    name = Column(String(50), nullable=False)


class SpLeadSubSpecialtyTable(BaseTable):
    __tablename__ = TableName.LEAD_SP_SUBSPECIALTY.value

    name = Column(String(50), nullable=False)
    specialty_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LEAD_SP_SPECIALTY.value}.entity_id"), nullable=True
    )


class SPLeadTable(LeadTable):
    __tablename__ = TableName.SP_LEAD.value

    entity_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.G_LEAD.value}.entity_id"), primary_key=True)
    specialties: Mapped[List[SpLeadSpecialtyTable]] = relationship(secondary=sp_lead_specialty_table)
    sub_specialties: Mapped[List[SpLeadSubSpecialtyTable]] = relationship(secondary=sp_lead_subspecialty_table)
    providers_count = Column(Integer(), nullable=True)
    locations_count = Column(Integer(), nullable=True)
    age_range = Column(ARRAY(Integer()), nullable=True)

    __mapper_args__ = {
        "polymorphic_identity": "SPECIALIST",
    }


class LeadAddressTable(PdmBaseTable, BaseTable):
    __tablename__ = TableName.LEAD_ADDRESS.value

    lead_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.G_LEAD.value}.entity_id"))

    address_1 = Column(String(100), nullable=True, comment="Address line 1.")
    address_2 = Column(String(100), nullable=True, comment="Address line 2.")

    city = Column(String(100), nullable=True, comment="Address city.")
    zip_code = Column(String(10), nullable=True, comment="Address zip code.")

    state_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LOCATION_STATE.value}.entity_id"),
        nullable=True,
        comment="Address state identification.",
    )
    county_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LOCATION_COUNTY.value}.entity_id"), nullable=True, comment="Address county."
    )
    region_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LOCATION_LEAD_REGION.value}.entity_id"), nullable=True, comment="Address region."
    )

    website_url = Column(String(100), nullable=True, comment="Address website URL.")
    phone = Column(String(50), nullable=True, comment="Address phone.")
    fax = Column(String(50), nullable=True, comment="Address fax.")

    is_main_location = Column(Boolean(), nullable=True, comment="True if current address is the main location.")

    state: Mapped["LocationStateTable"] = relationship("LocationStateTable", foreign_keys=[state_id])
    county: Mapped["LocationCountyTable"] = relationship("LocationCountyTable", foreign_keys=[county_id])
    region: Mapped["LocationLeadRegionTable"] = relationship("LocationLeadRegionTable", foreign_keys=[region_id])


class TPLeadServiceDetailTable(PdmBaseTable, BaseTable):
    __tablename__ = TableName.TP_LEAD_SERVICE_DETAIL.value

    lead_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.TP_LEAD.value}.entity_id"))

    state_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LOCATION_STATE.value}.entity_id"),
        nullable=True,
        comment="Address state identification.",
    )
    county_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LOCATION_COUNTY.value}.entity_id"), nullable=True, comment="Address county."
    )
    region_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.LOCATION_LEAD_REGION.value}.entity_id"), nullable=True, comment="Address region."
    )

    mobility_type_id: Mapped[uuid.UUID] = mapped_column(
        ForeignKey(f"{TableName.TP_MOBILITY_TYPE.value}.entity_id"), nullable=True
    )

    vehicles_count = Column(Integer(), nullable=True)

    mobility_type: Mapped["TpLeadMobilityTypeTable"] = relationship()


class LeadImportTaskTable(BaseTable):
    __tablename__ = TableName.IMPORT_TASK.value

    file_path = Column(String(100), nullable=False, comment="File location path.")
    file_fails_path = Column(String(100), nullable=True, comment="File path location with unimported records.")

    import_id = Column(String(100), nullable=False, unique=True, comment="Unique Id from creator.")

    separator = Column(String(10), nullable=False, unique=False, comment="Data separator on csv")

    mappings = Column(ARRAY(String), nullable=False, comment="Mapping configuration for imports.")

    lead_type = Column(String(50), nullable=False, unique=False, comment="Lead type content.")

    csv_headers = Column(ARRAY(String), nullable=True, comment="Extracted csv headers.")

    records_total = Column(Integer(), nullable=True, comment="Amount of records into the file.")
    records_analyzed = Column(Integer(), nullable=True, comment="Amount of records analyzed.")
    records_duplicated = Column(Integer(), nullable=True, comment="Amount of duplicated records.")
    records_invalid = Column(Integer(), nullable=True, comment="Amount of invalid records.")

    start_at = Column(DateTime, nullable=True, comment="Date when import tasks was started.")
    end_at = Column(DateTime, nullable=True, comment="Date when import task was finished.")

    is_write = Column(Boolean(), nullable=True, comment="True if lead should be imported.")

    creator_user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.USER.value}.entity_id"), nullable=True)
    creator_user: Mapped["UserTable"] = relationship(back_populates="created_import_task_leads")


class LeadChangeTable(BaseTable):
    __tablename__ = TableName.LEAD_CHANGES.value

    lead_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.G_LEAD.value}.entity_id"))

    changes = Column(JSON, nullable=False)

    user_id: Mapped[uuid.UUID] = mapped_column(ForeignKey(f"{TableName.USER.value}.entity_id"), nullable=True)
    user_fullname = Column(String(50), nullable=True, comment="Change author fullname.")

    user_roles = Column(ARRAY(String), nullable=True, comment="User roles.")


lead_type_index = Index("idx_lead_type", LocationLeadRegionTable.lead_type)
Editor is loading...
Leave a Comment