mirror of
https://github.com/Hestia-Homes/survey-extraction.git
synced 2026-06-08 11:17:29 +00:00
84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
from sqlmodel import Field, SQLModel, Relationship, Column, text
|
|
import uuid
|
|
from typing import Optional, List
|
|
from datetime import datetime
|
|
from pydantic import EmailStr
|
|
from sqlalchemy import Enum as SAEnum
|
|
from sqlalchemy import Column
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from etl.fileReader.reportType import ReportType
|
|
from sqlalchemy import DateTime
|
|
from sqlalchemy.dialects.postgresql import JSON
|
|
from sqlalchemy import Text
|
|
|
|
class BaseModel(SQLModel):
|
|
# Put primary_key=True in Column; don't pass primary_key to Field
|
|
id: uuid.UUID = Field(
|
|
sa_column=Column(
|
|
UUID(as_uuid=True),
|
|
primary_key=True,
|
|
nullable=False,
|
|
server_default=text("gen_random_uuid()"), # requires pgcrypto extension
|
|
)
|
|
)
|
|
|
|
# class Buildings(BaseModel, table=True):
|
|
# address: str
|
|
# postcode: str
|
|
# UPRN: str
|
|
# landlord_id: str
|
|
# domna_id: str
|
|
|
|
# documents: List["Documents"] = Relationship(back_populates="building")
|
|
|
|
# class Documents(BaseModel, table=True):
|
|
# assessor_id: uuid.UUID = Field(
|
|
# foreign_key="assessorinfo.id",
|
|
# nullable=False
|
|
# )
|
|
# author: Optional["AssessorInfo"] = Relationship(back_populates="documents")
|
|
# created_at: datetime
|
|
# document_type: ReportType
|
|
|
|
# building_id: uuid.UUID = Field(foreign_key="buildings.id", nullable=False)
|
|
# building: Optional["Buildings"] = Relationship(back_populates="documents")
|
|
|
|
# target_table: str
|
|
# target_id: uuid.UUID
|
|
|
|
# class ReportType(str, Enum):
|
|
# QUIDOS_PRESITE_NOTE = "QUIDOS_PRESITE_NOTE"
|
|
# CHARTED_SURVEYOR_REPORT = "CHARTED_SURVEYOR_REPORT"
|
|
# ENERGY_PERFORMANCE_REPORT = "ENERGY_PERFORMANCE_REPORT"
|
|
# U_VALUE_CALCULATOR_REPORT = "U_VALUE_CALCULATOR_REPORT"
|
|
# OVERWRITING_U_VALUE_DECLARATION_FORM = "OVERWRITING_U_VALUE_DECLARATION_FORM"
|
|
# OSMOSIS_CONDITION_PAS_2035_REPORT = "OSMOSIS_CONDITION_PAS_2035_REPORT"
|
|
# DOMNA_CONDITION_PAS_2035_REPORT = "DOMNA_CONDITION_PAS_2035_REPORT"
|
|
|
|
class uploaded_files(BaseModel, table=True):
|
|
__tablename__ = "uploaded_files"
|
|
|
|
s3_json_uri: Optional[str] = Field(
|
|
sa_column=Column(Text, nullable=True)
|
|
)
|
|
s3_file_uri: str = Field(index=True)
|
|
|
|
doc_type: ReportType = Field(
|
|
sa_column=Column(
|
|
SAEnum(ReportType, name="reporttype", native_enum=True, create_type=False, validate_strings=True),
|
|
nullable=False,
|
|
)
|
|
)
|
|
|
|
s3_file_upload_timestamp: datetime = Field(
|
|
sa_column=Column(
|
|
DateTime(timezone=True),
|
|
server_default=text("NOW() AT TIME ZONE 'utc'"),
|
|
nullable=False,
|
|
)
|
|
)
|
|
s3_json_upload_timestamp: Optional[datetime] = Field(
|
|
sa_column=Column(DateTime(timezone=True), nullable=True)
|
|
)
|
|
|
|
uprn: str = Field(index=True)
|