ac2a09e2bd
- Move Base to app/models/base.py so alembic env.py can import it without triggering engine creation (which connects to hardcoded DB) - Update all 13 models to import Base from app.models.base - Pin golangci-lint to latest (supports Go 1.25) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
"""Firmware version tracking and upgrade job ORM models."""
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import (
|
|
BigInteger,
|
|
Boolean,
|
|
DateTime,
|
|
Text,
|
|
UniqueConstraint,
|
|
func,
|
|
)
|
|
from sqlalchemy.dialects.postgresql import UUID
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
from sqlalchemy import ForeignKey
|
|
|
|
from app.models.base import Base
|
|
|
|
|
|
class FirmwareVersion(Base):
|
|
"""Cached firmware version from MikroTik download server or poller discovery.
|
|
|
|
Not tenant-scoped — firmware versions are global data shared across all tenants.
|
|
"""
|
|
|
|
__tablename__ = "firmware_versions"
|
|
__table_args__ = (UniqueConstraint("architecture", "channel", "version"),)
|
|
|
|
id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
primary_key=True,
|
|
default=uuid.uuid4,
|
|
server_default=func.gen_random_uuid(),
|
|
)
|
|
architecture: Mapped[str] = mapped_column(Text, nullable=False)
|
|
channel: Mapped[str] = mapped_column(Text, nullable=False) # "stable", "long-term", "testing"
|
|
version: Mapped[str] = mapped_column(Text, nullable=False)
|
|
npk_url: Mapped[str] = mapped_column(Text, nullable=False)
|
|
npk_local_path: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
npk_size_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
|
|
checked_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
server_default=func.now(),
|
|
nullable=False,
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<FirmwareVersion arch={self.architecture} ch={self.channel} ver={self.version}>"
|
|
|
|
|
|
class FirmwareUpgradeJob(Base):
|
|
"""Tracks a firmware upgrade operation for a single device.
|
|
|
|
Multiple jobs can share a rollout_group_id for mass upgrades.
|
|
"""
|
|
|
|
__tablename__ = "firmware_upgrade_jobs"
|
|
|
|
id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
primary_key=True,
|
|
default=uuid.uuid4,
|
|
server_default=func.gen_random_uuid(),
|
|
)
|
|
tenant_id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("tenants.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
)
|
|
device_id: Mapped[uuid.UUID] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
ForeignKey("devices.id", ondelete="CASCADE"),
|
|
nullable=False,
|
|
)
|
|
rollout_group_id: Mapped[uuid.UUID | None] = mapped_column(
|
|
UUID(as_uuid=True),
|
|
nullable=True,
|
|
)
|
|
target_version: Mapped[str] = mapped_column(Text, nullable=False)
|
|
architecture: Mapped[str] = mapped_column(Text, nullable=False)
|
|
channel: Mapped[str] = mapped_column(Text, nullable=False)
|
|
status: Mapped[str] = mapped_column(
|
|
Text, nullable=False, default="pending", server_default="pending"
|
|
)
|
|
pre_upgrade_backup_sha: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
scheduled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
|
|
error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
|
|
confirmed_major_upgrade: Mapped[bool] = mapped_column(
|
|
Boolean, nullable=False, default=False, server_default="false"
|
|
)
|
|
created_at: Mapped[datetime] = mapped_column(
|
|
DateTime(timezone=True),
|
|
server_default=func.now(),
|
|
nullable=False,
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return (
|
|
f"<FirmwareUpgradeJob id={self.id} status={self.status} target={self.target_version}>"
|
|
)
|