feat(13-02): add device_interfaces table migration and ORM model

- Migration 032 creates device_interfaces with RLS, MAC index, unique(device_id, name)
- DeviceInterface SQLAlchemy model with all columns and device relationship

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jason Staack
2026-03-19 06:01:22 -05:00
parent 4b5bb949e9
commit 7147b15e13
2 changed files with 129 additions and 0 deletions

View File

@@ -0,0 +1,80 @@
"""Create device_interfaces table for MAC-to-device resolution.
Revision ID: 032
Revises: 031
Create Date: 2026-03-19
Stores interface metadata (name, MAC, type, running state) per device.
Used by link discovery to resolve MAC addresses to specific devices.
"""
import sqlalchemy as sa
from alembic import op
revision = "032"
down_revision = "031"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"device_interfaces",
sa.Column(
"id",
sa.dialects.postgresql.UUID(as_uuid=True),
primary_key=True,
server_default=sa.text("gen_random_uuid()"),
),
sa.Column(
"device_id",
sa.dialects.postgresql.UUID(as_uuid=True),
sa.ForeignKey("devices.id", ondelete="CASCADE"),
nullable=False,
index=True,
),
sa.Column(
"tenant_id",
sa.dialects.postgresql.UUID(as_uuid=True),
sa.ForeignKey("tenants.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("name", sa.String(255), nullable=False),
sa.Column("mac_address", sa.String(17), nullable=False),
sa.Column("type", sa.String(50), nullable=False),
sa.Column("running", sa.Boolean, nullable=False, server_default=sa.text("false")),
sa.Column(
"updated_at",
sa.DateTime(timezone=True),
server_default=sa.func.now(),
nullable=False,
),
sa.UniqueConstraint("device_id", "name", name="uq_device_interfaces_device_name"),
)
op.create_index("idx_device_interfaces_mac", "device_interfaces", ["mac_address"])
op.create_index("idx_device_interfaces_tenant", "device_interfaces", ["tenant_id"])
# Enable RLS with tenant isolation policy
conn = op.get_bind()
conn.execute(sa.text("ALTER TABLE device_interfaces ENABLE ROW LEVEL SECURITY"))
conn.execute(sa.text("ALTER TABLE device_interfaces FORCE ROW LEVEL SECURITY"))
conn.execute(
sa.text("""
CREATE POLICY tenant_isolation ON device_interfaces
USING (
tenant_id::text = current_setting('app.current_tenant', true)
OR current_setting('app.current_tenant', true) = 'super_admin'
)
WITH CHECK (
tenant_id::text = current_setting('app.current_tenant', true)
OR current_setting('app.current_tenant', true) = 'super_admin'
)
""")
)
def downgrade() -> None:
conn = op.get_bind()
conn.execute(sa.text("DROP POLICY IF EXISTS tenant_isolation ON device_interfaces"))
op.drop_table("device_interfaces")

View File

@@ -0,0 +1,49 @@
"""DeviceInterface model -- interface metadata for MAC-to-device resolution."""
import uuid
from datetime import datetime
from sqlalchemy import Boolean, DateTime, ForeignKey, String, UniqueConstraint, func
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.models.base import Base
class DeviceInterface(Base):
__tablename__ = "device_interfaces"
__table_args__ = (
UniqueConstraint("device_id", "name", name="uq_device_interfaces_device_name"),
)
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
server_default=func.gen_random_uuid(),
)
device_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("devices.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
tenant_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("tenants.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
mac_address: Mapped[str] = mapped_column(String(17), nullable=False)
type: Mapped[str] = mapped_column(String(50), nullable=False)
running: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), nullable=False
)
# Relationships
device: Mapped["Device"] = relationship("Device") # type: ignore[name-defined]
def __repr__(self) -> str:
return f"<DeviceInterface id={self.id} name={self.name!r} mac={self.mac_address!r}>"