from datetime import UTC, datetime from sqlalchemy import ( BigInteger, Boolean, DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint, text, ) from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column class Base(DeclarativeBase): pass def now_utc() -> datetime: return datetime.now(UTC) class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True) username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False) password_digest: Mapped[str] = mapped_column(String(255), nullable=False) admin: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) class Overlay(Base): __tablename__ = "overlays" __table_args__ = ( Index( "uq_overlay_name_system", "name", unique=True, sqlite_where=text("user_id IS NULL"), ), Index( "uq_overlay_name_per_user", "name", "user_id", unique=True, sqlite_where=text("user_id IS NOT NULL"), ), Index("ix_overlays_type_user_id", "type", "user_id"), {"sqlite_autoincrement": True}, ) id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(128), nullable=False) path: Mapped[str] = mapped_column(String(512), nullable=False) type: Mapped[str] = mapped_column(String(16), nullable=False, default="workshop") user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), nullable=True) script: Mapped[str] = mapped_column(Text, default="", nullable=False) last_build_status: Mapped[str] = mapped_column(String(16), default="", nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) class WorkshopItem(Base): __tablename__ = "workshop_items" id: Mapped[int] = mapped_column(Integer, primary_key=True) steam_id: Mapped[str] = mapped_column(String(20), unique=True, nullable=False) title: Mapped[str] = mapped_column(String(255), default="", nullable=False) filename: Mapped[str] = mapped_column(String(255), default="", nullable=False) file_url: Mapped[str] = mapped_column(Text, default="", nullable=False) file_size: Mapped[int] = mapped_column(BigInteger, default=0, nullable=False) time_updated: Mapped[int] = mapped_column(Integer, default=0, nullable=False) preview_url: Mapped[str] = mapped_column(Text, default="", nullable=False) last_downloaded_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) last_error: Mapped[str] = mapped_column(Text, default="", nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) class OverlayWorkshopItem(Base): __tablename__ = "overlay_workshop_items" __table_args__ = ( UniqueConstraint("overlay_id", "workshop_item_id", name="uq_overlay_workshop_item"), Index("ix_owi_workshop_item", "workshop_item_id"), ) id: Mapped[int] = mapped_column(Integer, primary_key=True) overlay_id: Mapped[int] = mapped_column( ForeignKey("overlays.id", ondelete="CASCADE"), nullable=False ) workshop_item_id: Mapped[int] = mapped_column( ForeignKey("workshop_items.id", ondelete="RESTRICT"), nullable=False ) class Blueprint(Base): __tablename__ = "blueprints" id: Mapped[int] = mapped_column(Integer, primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False) name: Mapped[str] = mapped_column(String(128), nullable=False) arguments: Mapped[str] = mapped_column(Text, default="[]", nullable=False) config: Mapped[str] = mapped_column(Text, default="[]", nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) class BlueprintOverlay(Base): __tablename__ = "blueprint_overlays" id: Mapped[int] = mapped_column(Integer, primary_key=True) blueprint_id: Mapped[int] = mapped_column(ForeignKey("blueprints.id"), nullable=False) overlay_id: Mapped[int] = mapped_column(ForeignKey("overlays.id"), nullable=False) position: Mapped[int] = mapped_column(Integer, nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) class Server(Base): __tablename__ = "servers" id: Mapped[int] = mapped_column(Integer, primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False) blueprint_id: Mapped[int] = mapped_column(ForeignKey("blueprints.id"), nullable=False) name: Mapped[str] = mapped_column(String(128), unique=True, nullable=False) port: Mapped[int] = mapped_column(Integer, unique=True, nullable=False) desired_state: Mapped[str] = mapped_column(String(16), default="stopped", nullable=False) actual_state: Mapped[str] = mapped_column(String(16), default="unknown", nullable=False) actual_state_updated_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) last_error: Mapped[str] = mapped_column(Text, default="", nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) class Job(Base): __tablename__ = "jobs" id: Mapped[int] = mapped_column(Integer, primary_key=True) user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id"), nullable=True) server_id: Mapped[int | None] = mapped_column(ForeignKey("servers.id"), nullable=True) overlay_id: Mapped[int | None] = mapped_column(ForeignKey("overlays.id"), nullable=True) operation: Mapped[str] = mapped_column(String(32), nullable=False) state: Mapped[str] = mapped_column(String(16), default="queued", nullable=False) exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True) started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) updated_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False) class JobLog(Base): __tablename__ = "job_logs" id: Mapped[int] = mapped_column(Integer, primary_key=True) job_id: Mapped[int] = mapped_column(ForeignKey("jobs.id"), nullable=False) seq: Mapped[int] = mapped_column(Integer, nullable=False) stream: Mapped[str] = mapped_column(String(8), nullable=False) line: Mapped[str] = mapped_column(Text, nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, default=now_utc, nullable=False)