import uuid from datetime import UTC, datetime from typing import Any from pydantic import EmailStr from sqlalchemy import JSON, Column from sqlmodel import Field, SQLModel from fluksio.core.types import UTCDateTime def get_datetime_utc() -> datetime: return datetime.now(UTC) # Shared properties class UserBase(SQLModel): email: EmailStr = Field(unique=True, index=True, max_length=255) is_active: bool = True is_superuser: bool = False full_name: str | None = Field(default=None, max_length=255) # Properties to receive via API on creation class UserCreate(UserBase): password: str = Field(min_length=8, max_length=128) class UserRegister(SQLModel): email: EmailStr = Field(max_length=255) password: str = Field(min_length=8, max_length=128) full_name: str | None = Field(default=None, max_length=255) # Properties to receive via API on update, all are optional class UserUpdate(UserBase): email: EmailStr | None = Field(default=None, max_length=255) # type: ignore password: str | None = Field(default=None, min_length=8, max_length=128) class UserUpdateMe(SQLModel): full_name: str | None = Field(default=None, max_length=255) email: EmailStr | None = Field(default=None, max_length=255) class UpdatePassword(SQLModel): current_password: str = Field(min_length=8, max_length=128) new_password: str = Field(min_length=8, max_length=128) # Database model, database table inferred from class name class User(UserBase, table=True): id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) hashed_password: str #: The portal account this local user stands for, if any. Set for the #: superuser who enrolled this installation and for every remote user one #: of them admitted; None for a purely local account, which is what an #: installation nobody enrolled has only. portal_sub: str | None = Field(default=None, max_length=64, unique=True, index=True) created_at: datetime | None = Field( default_factory=get_datetime_utc, sa_type=UTCDateTime, ) # Properties to return via API, id is always required class UserPublic(UserBase): id: uuid.UUID portal_sub: str | None = None created_at: datetime | None = None class UsersPublic(SQLModel): data: list[UserPublic] count: int # Generic message class Message(SQLModel): message: str # JSON payload containing access token class Token(SQLModel): access_token: str token_type: str = "bearer" # Contents of JWT token class TokenPayload(SQLModel): sub: str | None = None #: Set instead of ``sub`` by a token the portal minted for a person: it #: names them on the portal, and the local account it stands for is looked #: up from it. A portal identity nobody here was mapped to resolves to no #: user at all, which is the refusal. portal_sub: str | None = None class NewPassword(SQLModel): token: str new_password: str = Field(min_length=8, max_length=128) # ----------------------------------------------------------------------------- # OAuth 2.1, for the MCP endpoint # # Agents cannot be handed a password, so they get their own authorization flow: # the client registers itself, a human approves it in the browser, and the code # that comes back is exchanged for a token. Only the hash of a code or a refresh # token is stored, so a copy of this table is not a set of credentials. # ----------------------------------------------------------------------------- class OAuthClient(SQLModel, table=True): """A registered agent. Registration alone grants nothing.""" __tablename__ = "oauth_client" id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) client_name: str = Field(max_length=128) redirect_uris: list[str] = Field(sa_column=Column(JSON), default_factory=list) created_at: datetime = Field( default_factory=lambda: datetime.now(UTC), nullable=False, sa_type=UTCDateTime, ) class OAuthAuthorizationCode(SQLModel, table=True): """One approved authorization, waiting to be exchanged for a token.""" __tablename__ = "oauth_authorization_code" id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) code_hash: str = Field(max_length=64, unique=True, index=True) client_id: uuid.UUID = Field( foreign_key="oauth_client.id", nullable=False, ondelete="CASCADE" ) user_id: uuid.UUID = Field( foreign_key="user.id", nullable=False, ondelete="CASCADE" ) redirect_uri: str = Field(max_length=2048) code_challenge: str = Field(max_length=128) resource: str | None = Field(default=None, max_length=2048) expires_at: datetime = Field(sa_type=UTCDateTime) used_at: datetime | None = Field(default=None, sa_type=UTCDateTime) #: The refresh token this code produced, so replaying the code can revoke it. refresh_token_id: uuid.UUID | None = None created_at: datetime = Field( default_factory=lambda: datetime.now(UTC), nullable=False, sa_type=UTCDateTime, ) class OAuthRefreshToken(SQLModel, table=True): """A rotating refresh token, one family per authorization.""" __tablename__ = "oauth_refresh_token" id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) token_hash: str = Field(max_length=64, unique=True, index=True) client_id: uuid.UUID = Field( foreign_key="oauth_client.id", nullable=False, ondelete="CASCADE" ) user_id: uuid.UUID = Field( foreign_key="user.id", nullable=False, ondelete="CASCADE" ) #: Every token rotated out of one authorization shares this, so reusing an #: old one can revoke the whole line rather than just itself. family_id: uuid.UUID expires_at: datetime = Field(sa_type=UTCDateTime) revoked: bool = False created_at: datetime = Field( default_factory=lambda: datetime.now(UTC), nullable=False, sa_type=UTCDateTime, ) class OAuthClientRegister(SQLModel): """RFC 7591 dynamic client registration request.""" client_name: str = Field(default="MCP client", max_length=128) redirect_uris: list[str] grant_types: list[str] | None = None response_types: list[str] | None = None token_endpoint_auth_method: str | None = None class OAuthClientInfo(SQLModel): client_id: str client_name: str redirect_uris: list[str] client_id_issued_at: int grant_types: list[str] = ["authorization_code", "refresh_token"] response_types: list[str] = ["code"] token_endpoint_auth_method: str = "none" class OAuthAuthorizeInfo(SQLModel): """What the consent page shows, all of it validated server-side.""" client_name: str redirect_uri: str scope: str class OAuthAuthorizeRequest(SQLModel): client_id: str redirect_uri: str code_challenge: str code_challenge_method: str = "S256" state: str | None = None resource: str | None = None scope: str | None = None class OAuthAuthorizeResponse(SQLModel): redirect_url: str # ----------------------------------------------------------------------------- # Observability # # The engine's own history: what ran, how long it took, and what went wrong. # Rolled up per minute rather than kept per execution — a node firing every # second is 86 400 rows a day raw, and nobody reads a row of that. # ----------------------------------------------------------------------------- class MetricBucket(SQLModel, table=True): """One node's minute: how much ran, how long it took, how late it was.""" __tablename__ = "metric_minute" flow: str = Field(primary_key=True, max_length=255) node: str = Field(primary_key=True, max_length=255) bucket: datetime = Field( primary_key=True, index=True, sa_type=UTCDateTime, ) executions: int = 0 errors: int = 0 #: Messages emitted, which is not the same as executions: a node can run #: and publish nothing. messages: int = 0 duration_sum_ms: float = 0.0 duration_max_ms: float = 0.0 #: How long queued work waited before it ran. lag_sum_ms: float = 0.0 lag_max_ms: float = 0.0 #: Items the lag sums are over, so an average can be taken. items: int = 0 class EngineEvent(SQLModel, table=True): """Something worth keeping after the websocket has forgotten it.""" __tablename__ = "engine_event" id: int | None = Field(default=None, primary_key=True) ts: datetime = Field( index=True, sa_type=UTCDateTime, ) #: node_error, flow_quarantined, engine_degraded, …, or audit. type: str = Field(max_length=32, index=True) flow: str = "" node: str = "" #: The error and its traceback, or what an audit entry says was done. detail: str = "" #: Who did it, on audit rows. actor: str = "" #: The run this happened in: a batch run's id, or the journaled item a live #: cascade came from. Empty for everything that belongs to neither. run: str = Field(default="", index=True, max_length=64) class FlowRun(SQLModel, table=True): """One cascade, from the item that started it to the last node in it.""" __tablename__ = "flow_run" #: The queue entry id, or a `manual-` one for a run that never queued. id: str = Field(primary_key=True, max_length=64) flow: str = Field(index=True, max_length=255) #: What caused it: the work item's cause, or "manual". source: str = "" started_at: datetime = Field( index=True, sa_type=UTCDateTime, ) finished_at: datetime | None = Field( default=None, sa_type=UTCDateTime, ) #: running, ok, error or abandoned. status: str = "running" nodes: int = 0 errors: int = 0 duration_ms: float = 0.0 deliveries: int = 1 # ----------------------------------------------------------------------------- # Runs # # A cascade above is what an always-on flow does when a value arrives. A run is # the other shape: a batch flow taken from its inputs to its outputs once, with # parameters that identify it and a result worth keeping. An experiment is a # run, and so is a CI-style job — same entity, different caller. # # Kept apart from the observability tables on purpose: those are rolled up and # pruned on a retention window, and an experiment nobody wants deleted must not # share that fate. # ----------------------------------------------------------------------------- class Run(SQLModel, table=True): """One finite execution of a flow, with what it was asked and what it made.""" __tablename__ = "run" id: str = Field(primary_key=True, max_length=64) flow: str = Field(index=True, max_length=255) #: The flow document this ran, and the commit it was read at, so a result #: can be traced back to the code that produced it. flow_version: int = 1 commit: str = Field(default="", max_length=64) #: The commit of the *user's* repository, for a flow declared there with #: the decorators. The store's commit names a generated import shim; this #: one names the code it imported. A `-dirty` suffix is git's own way of #: saying the tree had changes that are in no commit at all. origin_commit: str = Field(default="", max_length=80) #: sha256 over that repository's python files, read when the run actually #: started. The commit cannot tell two runs of an uncommitted tree apart — #: both stamp `-dirty` — and the shim imports whatever is on disk, so this #: is the half that names the code. Empty when the engine cannot see the #: repository, or for a flow drawn on the canvas. code_digest: str = Field(default="", max_length=64) params: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict) #: sha256 over params and seed. Two runs of the same thing share it, which #: is what makes "have I already run this?" a lookup. params_digest: str = Field(default="", index=True, max_length=64) seed: int | None = None #: Runs submitted together — a sweep, an ensemble. group_id: str | None = Field(default=None, index=True, max_length=64) #: The run this one was made from, on a retry. parent_id: str | None = Field(default=None, max_length=64) #: A key the caller made up for one submission, so retrying a submit that #: may already have landed returns that run instead of starting a second. #: Unique where it is set; null for anything submitted without one. idempotency_key: str | None = Field( default=None, unique=True, index=True, max_length=64 ) #: Where it was asked for: api, cli, sdk, hook or sweep. cause: str = Field(default="api", max_length=32) #: Re-execute every node, whatever the stage cache holds for it. no_cache: bool = False #: queued, running, ok, error, cancelled or abandoned. status: str = Field(default="queued", index=True, max_length=16) #: Why it is where it is: what it waits for, or what went wrong. status_reason: str = Field(default="", max_length=1024) #: Worker labels its nodes need, so a run with nowhere to go can say so. labels: list[str] = Field(sa_column=Column(JSON), default_factory=list) #: The largest machine its nodes ask for, as cpus, gpus, ram_mb and device. #: Read while the run is queued, to tell waiting for a machine from having #: nowhere to run at all. needs: dict[str, Any] | None = Field(sa_column=Column(JSON), default=None) created_at: datetime = Field( index=True, sa_type=UTCDateTime, ) started_at: datetime | None = Field( default=None, sa_type=UTCDateTime, ) finished_at: datetime | None = Field( default=None, sa_type=UTCDateTime, ) duration_ms: float = 0.0 #: The flow's declared outputs once it finished. result: dict[str, Any] = Field(sa_column=Column(JSON), default_factory=dict) #: Which engine holds it, and when it last said so. A run whose lease has #: gone stale is one whose engine died mid-run. engine: str = Field(default="", max_length=64) lease_at: datetime | None = Field( default=None, sa_type=UTCDateTime, ) actor: str = Field(default="", max_length=255) class RunNode(SQLModel, table=True): """What one node did inside a run: the per-stage view of it.""" __tablename__ = "run_node" run_id: str = Field(primary_key=True, max_length=64) node: str = Field(primary_key=True, max_length=255) #: ok, error, skipped, cached or cancelled. status: str = Field(default="ok", max_length=16) attempt: int = 1 started_at: datetime | None = Field( default=None, sa_type=UTCDateTime, ) duration_ms: float = 0.0 #: Which worker ran it: "local", or a remote worker's name. worker: str = Field(default="local", max_length=64) error: str = "" logs: str = "" #: Everything this node's output depends on, hashed: its source, its #: settings and the values it read. What a later run looks itself up by. cache_key: str = Field(default="", index=True, max_length=64) #: The run this node's values were restored from, when it was not executed. #: Its series lives there too, which is what makes a cached run's curve #: readable without copying a few thousand rows per reuse. cached_from: str = Field(default="", max_length=64) #: What it returned, as canonical JSON, so a node with this key can be #: skipped and its outputs restored. None when it may not be reused — #: opted out, too large, or a value JSON cannot carry. Written with #: `cache_key` or not at all, so a keyed row is always restorable. outputs: str | None = None class RunMetric(SQLModel, table=True): """One number a run reported, at one step. Written by the run's own driver rather than folded off the event bus: the bus drops what it cannot keep up with, which is the right trade for a live canvas and the wrong one for a training curve. """ __tablename__ = "run_metric" run_id: str = Field(primary_key=True, max_length=64) name: str = Field(primary_key=True, max_length=128) #: -1 for a value with no step of its own — a final score. step: int = Field(primary_key=True) node: str = Field(default="", max_length=255) ts: float = 0.0 value: float = 0.0 class RunArtifact(SQLModel, table=True): """A file a run produced, addressed by the hash of its content.""" __tablename__ = "run_artifact" run_id: str = Field(primary_key=True, max_length=64) #: The message the bytes left the node on, which is what addresses them. name: str = Field(primary_key=True, max_length=255) #: What the node called the file, when it said. Kept because a reference #: rebuilt from this row is otherwise the same bytes under another name. filename: str | None = Field(default=None, max_length=255) node: str = Field(default="", max_length=255) digest: str = Field(default="", index=True, max_length=71) size: int = 0 media_type: str = Field(default="application/octet-stream", max_length=128) created_at: datetime = Field( default_factory=lambda: datetime.now(UTC), sa_type=UTCDateTime, ) # ----------------------------------------------------------------------------- # Flavors # ----------------------------------------------------------------------------- class FlavorBase(SQLModel): """A named amount of machine, so a node can ask for one by name. The point is not to save typing. Cores and megabytes are a property of the machines an installation actually has, and they change when those machines do — so a node saying "gpu-small" keeps meaning something after the cluster is replaced, where a node saying 8 and 16384 quietly stops. """ cpus: int = Field(default=1, ge=1) gpus: int = Field(default=0, ge=0) #: Megabytes. A flavor always says, which is what a node asking for one is #: buying — the raw form leaves memory unstated and unaccounted. ram: int = Field(default=2048, ge=1) description: str = Field(default="", max_length=255) class Flavor(FlavorBase, table=True): """A resource size somebody named, stored so nodes can refer to it.""" __tablename__ = "flavor" name: str = Field(primary_key=True, max_length=64) class FlavorCreate(FlavorBase): name: str = Field(max_length=64) class FlavorUpdate(SQLModel): """Everything but the name: renaming would orphan the nodes that ask.""" cpus: int | None = Field(default=None, ge=1) gpus: int | None = Field(default=None, ge=0) ram: int | None = Field(default=None, ge=1) description: str | None = Field(default=None, max_length=255) class FlavorPublic(FlavorBase): name: str class FlavorsPublic(SQLModel): data: list[FlavorPublic] count: int