import enum import uuid from datetime import datetime from sqlalchemy import ( BigInteger, Boolean, DateTime, Enum, ForeignKey, ForeignKeyConstraint, Index, Integer, String, Text, UniqueConstraint, func, text, ) from sqlalchemy.dialects.postgresql import CIDR, INET, JSONB, UUID from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class PrefixStatus(str, enum.Enum): active = "active" reserved = "reserved" deprecated = "deprecated" class AddressStatus(str, enum.Enum): assigned = "assigned" reserved = "reserved" deprecated = "deprecated" class Role(str, enum.Enum): admin = "admin" viewer = "viewer" class Organization(Base): __tablename__ = "organizations" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(255), unique=True) short_name: Mapped[str] = mapped_column(String(100), default="") inn: Mapped[str] = mapped_column(String(12), unique=True) address: Mapped[str] = mapped_column(String(500), default="") contact_person: Mapped[str] = mapped_column(String(255), default="") phone: Mapped[str] = mapped_column(String(50), default="") email: Mapped[str] = mapped_column(String(255), default="") note: Mapped[str] = mapped_column(Text, default="") class Vrf(Base): __tablename__ = "vrfs" # (id, organization_id) — цель составного FK префиксов: VRF префикса всегда из его организации __table_args__ = (UniqueConstraint("id", "organization_id", name="uq_vrfs_id_organization_id"),) id: Mapped[int] = mapped_column(primary_key=True) organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id")) name: Mapped[str] = mapped_column(String(100)) route_target: Mapped[str] = mapped_column(String(50), default="") note: Mapped[str] = mapped_column(Text, default="") Index("uq_vrfs_org_lower_name", Vrf.organization_id, func.lower(Vrf.name), unique=True) # имя VRF уникально в организации без учёта регистра class Prefix(Base): __tablename__ = "prefixes" __table_args__ = ( UniqueConstraint("vrf_id", "prefix"), UniqueConstraint("id", "vrf_id", name="uq_prefixes_id_vrf_id"), # цель составного FK адресов ForeignKeyConstraint(["vrf_id", "organization_id"], ["vrfs.id", "vrfs.organization_id"], name="fk_prefixes_vrf_org"), ) id: Mapped[int] = mapped_column(primary_key=True) organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id"), index=True) vrf_id: Mapped[int] = mapped_column(ForeignKey("vrfs.id"), index=True) prefix: Mapped[str] = mapped_column(CIDR) description: Mapped[str] = mapped_column(String(500), default="") status: Mapped[PrefixStatus] = mapped_column(Enum(PrefixStatus, name="prefix_status"), default=PrefixStatus.active) parent_id: Mapped[int | None] = mapped_column(ForeignKey("prefixes.id", ondelete="SET NULL")) is_pool: Mapped[bool] = mapped_column(Boolean, default=False) note: Mapped[str] = mapped_column(Text, default="") vrf: Mapped[Vrf] = relationship(primaryjoin="Prefix.vrf_id == Vrf.id", foreign_keys=[vrf_id]) class Isp(Base): __tablename__ = "isps" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(255)) organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id"), index=True) contract_number: Mapped[str] = mapped_column(String(100), default="") hotline: Mapped[str] = mapped_column(String(50), default="") note: Mapped[str] = mapped_column(Text, default="") networks: Mapped[list["IspNetwork"]] = relationship(cascade="all, delete-orphan", order_by="IspNetwork.id") class IspNetwork(Base): __tablename__ = "isp_networks" id: Mapped[int] = mapped_column(primary_key=True) isp_id: Mapped[int] = mapped_column(ForeignKey("isps.id", ondelete="CASCADE"), index=True) cidr: Mapped[str] = mapped_column(CIDR) class DeviceType(Base): __tablename__ = "device_types" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(100), unique=True) is_default: Mapped[bool] = mapped_column(Boolean, default=False) class Device(Base): __tablename__ = "devices" __table_args__ = (UniqueConstraint("organization_id", "name"),) id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(255)) device_type_id: Mapped[int] = mapped_column(ForeignKey("device_types.id")) organization_id: Mapped[int] = mapped_column(ForeignKey("organizations.id"), index=True) mac: Mapped[str] = mapped_column(String(17), default="") note: Mapped[str] = mapped_column(Text, default="") class Address(Base): __tablename__ = "addresses" # vrf_id дублирует VRF префикса (составной FK, обновляется каскадом при переносе): так БД гарантирует уникальность IP в VRF __table_args__ = ( UniqueConstraint("prefix_id", "address"), UniqueConstraint("vrf_id", "address", name="uq_addresses_vrf_address"), ForeignKeyConstraint(["prefix_id", "vrf_id"], ["prefixes.id", "prefixes.vrf_id"], name="fk_addresses_prefix_vrf", ondelete="CASCADE", onupdate="CASCADE"), ) id: Mapped[int] = mapped_column(primary_key=True) prefix_id: Mapped[int] = mapped_column(index=True) vrf_id: Mapped[int] = mapped_column() address: Mapped[str] = mapped_column(INET) status: Mapped[AddressStatus] = mapped_column(Enum(AddressStatus, name="address_status"), default=AddressStatus.assigned) dns_name: Mapped[str] = mapped_column(String(255), default="") description: Mapped[str] = mapped_column(String(500), default="") device_id: Mapped[int | None] = mapped_column(ForeignKey("devices.id", ondelete="SET NULL"), index=True) note: Mapped[str] = mapped_column(Text, default="") updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) username: Mapped[str] = mapped_column(String(100), unique=True) password_hash: Mapped[str] = mapped_column(String(255)) role: Mapped[Role] = mapped_column(Enum(Role, name="user_role"), default=Role.viewer) is_active: Mapped[bool] = mapped_column(Boolean, default=True) password_changed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) # токены с iat раньше — недействительны Index("uq_users_lower_username", func.lower(User.username), unique=True) # логин уникален без учёта регистра class AuditLog(Base): __tablename__ = "audit_log" id: Mapped[int] = mapped_column(BigInteger, primary_key=True) ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) username: Mapped[str] = mapped_column(String(100)) entity_type: Mapped[str] = mapped_column(String(50), index=True) entity_id: Mapped[int | None] = mapped_column(Integer) entity_label: Mapped[str] = mapped_column(String(255)) action: Mapped[str] = mapped_column(String(20)) diff: Mapped[dict | None] = mapped_column(JSONB) uid: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), server_default=text("gen_random_uuid()"), unique=True) message: Mapped[str] = mapped_column(Text, default="", server_default="") client_ip: Mapped[str | None] = mapped_column(INET, index=True) # IP клиента запроса; NULL для системных событий meta: Mapped[dict | None] = mapped_column(JSONB) # user_agent, method, path, request_id __table_args__ = (Index("ix_audit_log_entity_type_action", "entity_type", "action"),) class AppSetting(Base): __tablename__ = "app_settings" key: Mapped[str] = mapped_column(String(50), primary_key=True) value: Mapped[dict] = mapped_column(JSONB) class LoginAttempt(Base): """Неудачные попытки входа (для ограничения перебора паролей); записи старше суток удаляет ротация.""" __tablename__ = "login_attempts" id: Mapped[int] = mapped_column(primary_key=True) ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) client_ip: Mapped[str | None] = mapped_column(INET) username: Mapped[str] = mapped_column(String(100)) # в нижнем регистре __table_args__ = (Index("ix_login_attempts_ip_ts", "client_ip", "ts"), Index("ix_login_attempts_user_ts", "username", "ts")) class ClearAttempt(Base): """Неудачные попытки подтверждения пароля при очистке журнала (для блокировки).""" __tablename__ = "clear_attempts" id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True) ts: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())