mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
switch Task back to use SQLModel because SQLAlchemy doesn't auto parse and validate response using pydantic
This commit is contained in:
Daniel Roth
parent
2afbf05ce0
commit
36f5ad6532
1 changed files with 109 additions and 109 deletions
|
|
@ -1,130 +1,130 @@
|
|||
# import enum
|
||||
# from typing import Optional
|
||||
# from datetime import datetime
|
||||
# from uuid import UUID, uuid4
|
||||
|
||||
# from sqlalchemy import Column, Enum
|
||||
# from sqlmodel import SQLModel, Field, Relationship
|
||||
|
||||
|
||||
# class SourceEnum(enum.Enum): # TODO: move to domain?
|
||||
# PORTFOLIO = "portfolio_id"
|
||||
|
||||
|
||||
# class Task(SQLModel, table=True):
|
||||
# __tablename__ = "tasks"
|
||||
|
||||
# id: UUID = Field(
|
||||
# default_factory=uuid4,
|
||||
# primary_key=True,
|
||||
# index=True,
|
||||
# )
|
||||
# task_source: str
|
||||
# job_started: Optional[datetime] = None
|
||||
# job_completed: Optional[datetime] = None
|
||||
# status: str = Field(default="In Progress")
|
||||
# service: Optional[str] = None
|
||||
# updated_at: datetime = Field(default_factory=datetime.utcnow)
|
||||
|
||||
# # source: Mapped[Optional[SourceEnum]] = mapped_column(Enum(SourceEnum)) <- SQLAlchemy not SQLModel
|
||||
|
||||
# source: Optional[SourceEnum] = Field(
|
||||
# default=None,
|
||||
# sa_column=Column(
|
||||
# Enum(
|
||||
# SourceEnum,
|
||||
# name="source",
|
||||
# values_callable=lambda e: [m.value for m in e],
|
||||
# ),
|
||||
# nullable=True,
|
||||
# ),
|
||||
# )
|
||||
# source_id: Optional[str] = None
|
||||
|
||||
# sub_tasks: list["SubTask"] = Relationship(back_populates="task")
|
||||
|
||||
|
||||
# class SubTask(SQLModel, table=True):
|
||||
# __tablename__ = "sub_task"
|
||||
|
||||
# id: UUID = Field(
|
||||
# default_factory=uuid4,
|
||||
# primary_key=True,
|
||||
# index=True,
|
||||
# )
|
||||
|
||||
# task_id: UUID = Field(foreign_key="tasks.id")
|
||||
# job_started: Optional[datetime] = None
|
||||
# job_completed: Optional[datetime] = None
|
||||
# status: str = Field(default="In Progress")
|
||||
# inputs: Optional[str] = None
|
||||
# outputs: Optional[str] = None
|
||||
# cloud_logs_url: Optional[str] = None
|
||||
# updated_at: datetime = Field(default_factory=datetime.utcnow)
|
||||
|
||||
# task: Optional["Task"] = Relationship(back_populates="sub_tasks")
|
||||
|
||||
|
||||
import enum
|
||||
from typing import Optional, List
|
||||
from typing import Optional
|
||||
from datetime import datetime
|
||||
from uuid import UUID, uuid4
|
||||
|
||||
from sqlalchemy import Enum, String, ForeignKey
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
from sqlalchemy.dialects.postgresql import UUID as PG_UUID, TIMESTAMP
|
||||
|
||||
from backend.app.db.base import Base
|
||||
from sqlalchemy import Column, Enum
|
||||
from sqlmodel import SQLModel, Field, Relationship
|
||||
|
||||
|
||||
class SourceEnum(enum.Enum):
|
||||
class SourceEnum(enum.Enum): # TODO: move to domain?
|
||||
PORTFOLIO = "portfolio_id"
|
||||
|
||||
|
||||
class Task(Base):
|
||||
class Task(SQLModel, table=True):
|
||||
__tablename__ = "tasks"
|
||||
|
||||
id: Mapped[UUID] = mapped_column(
|
||||
PG_UUID(as_uuid=True), primary_key=True, default=uuid4, index=True
|
||||
id: UUID = Field(
|
||||
default_factory=uuid4,
|
||||
primary_key=True,
|
||||
index=True,
|
||||
)
|
||||
task_source: Mapped[str] = mapped_column(String, nullable=False)
|
||||
job_started: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
job_completed: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
status: Mapped[str] = mapped_column(String, nullable=False, default="In Progress")
|
||||
service: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
TIMESTAMP, nullable=False, default=datetime.utcnow
|
||||
)
|
||||
source: Mapped[Optional[SourceEnum]] = mapped_column(
|
||||
Enum(
|
||||
SourceEnum,
|
||||
name="source",
|
||||
values_callable=lambda e: [m.value for m in e],
|
||||
task_source: str
|
||||
job_started: Optional[datetime] = None
|
||||
job_completed: Optional[datetime] = None
|
||||
status: str = Field(default="In Progress")
|
||||
service: Optional[str] = None
|
||||
updated_at: datetime = Field(default_factory=datetime.utcnow)
|
||||
|
||||
# source: Mapped[Optional[SourceEnum]] = mapped_column(Enum(SourceEnum)) <- SQLAlchemy not SQLModel
|
||||
|
||||
source: Optional[SourceEnum] = Field(
|
||||
default=None,
|
||||
sa_column=Column(
|
||||
Enum(
|
||||
SourceEnum,
|
||||
name="source",
|
||||
values_callable=lambda e: [m.value for m in e],
|
||||
),
|
||||
nullable=True,
|
||||
),
|
||||
nullable=True,
|
||||
)
|
||||
source_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
source_id: Optional[str] = None
|
||||
|
||||
sub_tasks: Mapped[List["SubTask"]] = relationship("SubTask", back_populates="task")
|
||||
sub_tasks: list["SubTask"] = Relationship(back_populates="task")
|
||||
|
||||
|
||||
class SubTask(Base):
|
||||
class SubTask(SQLModel, table=True):
|
||||
__tablename__ = "sub_task"
|
||||
|
||||
id: Mapped[UUID] = mapped_column(
|
||||
PG_UUID(as_uuid=True), primary_key=True, default=uuid4, index=True
|
||||
)
|
||||
task_id: Mapped[UUID] = mapped_column(
|
||||
PG_UUID(as_uuid=True), ForeignKey("tasks.id"), nullable=False
|
||||
)
|
||||
job_started: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
job_completed: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
status: Mapped[str] = mapped_column(String, nullable=False, default="In Progress")
|
||||
inputs: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
outputs: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
cloud_logs_url: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
TIMESTAMP, nullable=False, default=datetime.utcnow
|
||||
id: UUID = Field(
|
||||
default_factory=uuid4,
|
||||
primary_key=True,
|
||||
index=True,
|
||||
)
|
||||
|
||||
task: Mapped[Optional["Task"]] = relationship("Task", back_populates="sub_tasks")
|
||||
task_id: UUID = Field(foreign_key="tasks.id")
|
||||
job_started: Optional[datetime] = None
|
||||
job_completed: Optional[datetime] = None
|
||||
status: str = Field(default="In Progress")
|
||||
inputs: Optional[str] = None
|
||||
outputs: Optional[str] = None
|
||||
cloud_logs_url: Optional[str] = None
|
||||
updated_at: datetime = Field(default_factory=datetime.utcnow)
|
||||
|
||||
task: Optional["Task"] = Relationship(back_populates="sub_tasks")
|
||||
|
||||
|
||||
# import enum
|
||||
# from typing import Optional, List
|
||||
# from datetime import datetime
|
||||
# from uuid import UUID, uuid4
|
||||
|
||||
# from sqlalchemy import Enum, String, ForeignKey
|
||||
# from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
# from sqlalchemy.dialects.postgresql import UUID as PG_UUID, TIMESTAMP
|
||||
|
||||
# from backend.app.db.base import Base
|
||||
|
||||
|
||||
# class SourceEnum(enum.Enum):
|
||||
# PORTFOLIO = "portfolio_id"
|
||||
|
||||
|
||||
# class Task(Base):
|
||||
# __tablename__ = "tasks"
|
||||
|
||||
# id: Mapped[UUID] = mapped_column(
|
||||
# PG_UUID(as_uuid=True), primary_key=True, default=uuid4, index=True
|
||||
# )
|
||||
# task_source: Mapped[str] = mapped_column(String, nullable=False)
|
||||
# job_started: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
# job_completed: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
# status: Mapped[str] = mapped_column(String, nullable=False, default="In Progress")
|
||||
# service: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
# updated_at: Mapped[datetime] = mapped_column(
|
||||
# TIMESTAMP, nullable=False, default=datetime.utcnow
|
||||
# )
|
||||
# source: Mapped[Optional[SourceEnum]] = mapped_column(
|
||||
# Enum(
|
||||
# SourceEnum,
|
||||
# name="source",
|
||||
# values_callable=lambda e: [m.value for m in e],
|
||||
# ),
|
||||
# nullable=True,
|
||||
# )
|
||||
# source_id: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
|
||||
# sub_tasks: Mapped[List["SubTask"]] = relationship("SubTask", back_populates="task")
|
||||
|
||||
|
||||
# class SubTask(Base):
|
||||
# __tablename__ = "sub_task"
|
||||
|
||||
# id: Mapped[UUID] = mapped_column(
|
||||
# PG_UUID(as_uuid=True), primary_key=True, default=uuid4, index=True
|
||||
# )
|
||||
# task_id: Mapped[UUID] = mapped_column(
|
||||
# PG_UUID(as_uuid=True), ForeignKey("tasks.id"), nullable=False
|
||||
# )
|
||||
# job_started: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
# job_completed: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP, nullable=True)
|
||||
# status: Mapped[str] = mapped_column(String, nullable=False, default="In Progress")
|
||||
# inputs: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
# outputs: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
# cloud_logs_url: Mapped[Optional[str]] = mapped_column(String, nullable=True)
|
||||
# updated_at: Mapped[datetime] = mapped_column(
|
||||
# TIMESTAMP, nullable=False, default=datetime.utcnow
|
||||
# )
|
||||
|
||||
# task: Mapped[Optional["Task"]] = relationship("Task", back_populates="sub_tasks")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue