"""ORM model for the Execution table (one record per tracked function call)."""
import uuid
from datetime import datetime
from pydantic import BaseModel
from sqlalchemy import DateTime, Float, ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from .. import models
from .base import Base
def _uuid7() -> uuid.UUID:
import uuid_utils
return uuid.UUID(str(uuid_utils.uuid7()))
[docs]
class ExecutionTable(Base):
__tablename__ = "execution"
[docs]
id_: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=_uuid7)
[docs]
function_node_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("node.id_"), nullable=True,
)
[docs]
status: Mapped[str] = mapped_column(String(20), default="pending")
[docs]
start_time: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
[docs]
end_time: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
[docs]
duration_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)
[docs]
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
error_traceback: Mapped[str | None] = mapped_column(Text, nullable=True)
@classmethod
[docs]
def pydantic_create_class(cls) -> type[BaseModel]:
return models.ExecutionCreate
@classmethod
[docs]
def pydantic_model_class(cls) -> type[BaseModel]:
return models.Execution
@classmethod
[docs]
def class_string(cls) -> str:
return cls.__tablename__