xufuhan/desungongpai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
master
desungongpai/app/models.py
default 9cd9dd9d76 feat: 信标设备绑定 + 蓝牙模式管理 + 系统管理增强 + 数据导出 
- 新增 DeviceBeaconBinding 模型，信标-设备多对多绑定 CRUD
- 蓝牙打卡模式批量配置/恢复正常模式 API
- 反向同步: 查询设备 BTMACSET 配置更新数据库绑定 (独立 session 解决事务隔离)
- 设备列表快捷操作弹窗修复 (fire-and-forget IIFE 替代阻塞轮询)
- 保存按钮防抖: 围栏/信标绑定保存点击后 disabled + 转圈防重复提交
- 审计日志中间件 + 系统配置/备份/固件 API
- 设备分组管理 + 告警规则配置
- 5个数据导出 API (CSV UTF-8 BOM)
- 位置热力图 + 告警条件删除 + 位置清理

via [HAPI](https://hapi.run)

Co-Authored-By: HAPI <noreply@hapi.run>
2026-04-01 07:06:37 +00:00

533 lines
24 KiB
Python
Raw Permalink Blame History

from datetime import datetime
from sqlalchemy import (
BigInteger,
Boolean,
DateTime,
Float,
ForeignKey,
Index,
Integer,
String,
Text,
)
from sqlalchemy.dialects.sqlite import JSON
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.config import now_cst as _utcnow
from app.database import Base
class Device(Base):
"""Registered Bluetooth badge devices."""
__tablename__ = "devices"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
imei: Mapped[str] = mapped_column(String(20), unique=True, index=True, nullable=False)
device_type: Mapped[str] = mapped_column(String(10), nullable=False)
name: Mapped[str | None] = mapped_column(String(100), nullable=True)
status: Mapped[str] = mapped_column(String(20), default="offline", nullable=False)
battery_level: Mapped[int | None] = mapped_column(Integer, nullable=True)
gsm_signal: Mapped[int | None] = mapped_column(Integer, nullable=True)
last_heartbeat: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
last_login: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
iccid: Mapped[str | None] = mapped_column(String(30), nullable=True)
imsi: Mapped[str | None] = mapped_column(String(20), nullable=True)
timezone: Mapped[str] = mapped_column(String(30), default="+8", nullable=False)
language: Mapped[str] = mapped_column(String(10), default="cn", nullable=False)
group_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
updated_at: Mapped[datetime | None] = mapped_column(
DateTime, default=_utcnow, onupdate=_utcnow, nullable=True
)
# Relationships
locations: Mapped[list["LocationRecord"]] = relationship(
back_populates="device", cascade="all, delete-orphan", lazy="noload"
)
alarms: Mapped[list["AlarmRecord"]] = relationship(
back_populates="device", cascade="all, delete-orphan", lazy="noload"
)
heartbeats: Mapped[list["HeartbeatRecord"]] = relationship(
back_populates="device", cascade="all, delete-orphan", lazy="noload"
)
attendance_records: Mapped[list["AttendanceRecord"]] = relationship(
back_populates="device", cascade="all, delete-orphan", lazy="noload"
)
bluetooth_records: Mapped[list["BluetoothRecord"]] = relationship(
back_populates="device", cascade="all, delete-orphan", lazy="noload"
)
command_logs: Mapped[list["CommandLog"]] = relationship(
back_populates="device", cascade="all, delete-orphan", lazy="noload"
)
def __repr__(self) -> str:
return f"<Device(id={self.id}, imei={self.imei}, status={self.status})>"
class LocationRecord(Base):
"""GPS / LBS / WIFI location records."""
__tablename__ = "location_records"
__table_args__ = (
Index("ix_location_device_time", "device_id", "recorded_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
imei: Mapped[str | None] = mapped_column(String(20), nullable=True)
location_type: Mapped[str] = mapped_column(
String(10), nullable=False
) # gps, lbs, wifi, gps_4g, lbs_4g, wifi_4g
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
speed: Mapped[float | None] = mapped_column(Float, nullable=True)
course: Mapped[float | None] = mapped_column(Float, nullable=True)
gps_satellites: Mapped[int | None] = mapped_column(Integer, nullable=True)
gps_positioned: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
mcc: Mapped[int | None] = mapped_column(Integer, nullable=True)
mnc: Mapped[int | None] = mapped_column(Integer, nullable=True)
lac: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
cell_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
rssi: Mapped[int | None] = mapped_column(Integer, nullable=True)
neighbor_cells: Mapped[dict | None] = mapped_column(JSON, nullable=True)
wifi_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
report_mode: Mapped[int | None] = mapped_column(Integer, nullable=True)
is_realtime: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
mileage: Mapped[float | None] = mapped_column(Float, nullable=True)
address: Mapped[str | None] = mapped_column(Text, nullable=True)
raw_data: Mapped[str | None] = mapped_column(Text, nullable=True)
recorded_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
device: Mapped["Device"] = relationship(back_populates="locations")
def __repr__(self) -> str:
return (
f"<LocationRecord(id={self.id}, device_id={self.device_id}, "
f"type={self.location_type})>"
)
class AlarmRecord(Base):
"""Alarm events raised by devices."""
__tablename__ = "alarm_records"
__table_args__ = (
Index("ix_alarm_device_time", "device_id", "recorded_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
imei: Mapped[str | None] = mapped_column(String(20), nullable=True)
alarm_type: Mapped[str] = mapped_column(
String(30), nullable=False
) # sos, low_battery, power_on, power_off, enter_fence, exit_fence, ...
alarm_source: Mapped[str | None] = mapped_column(
String(20), nullable=True
) # single_fence, multi_fence, lbs, wifi
protocol_number: Mapped[int] = mapped_column(Integer, nullable=False)
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
speed: Mapped[float | None] = mapped_column(Float, nullable=True)
course: Mapped[float | None] = mapped_column(Float, nullable=True)
mcc: Mapped[int | None] = mapped_column(Integer, nullable=True)
mnc: Mapped[int | None] = mapped_column(Integer, nullable=True)
lac: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
cell_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
battery_level: Mapped[int | None] = mapped_column(Integer, nullable=True)
gsm_signal: Mapped[int | None] = mapped_column(Integer, nullable=True)
fence_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
wifi_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
address: Mapped[str | None] = mapped_column(Text, nullable=True)
acknowledged: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
recorded_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
device: Mapped["Device"] = relationship(back_populates="alarms")
def __repr__(self) -> str:
return (
f"<AlarmRecord(id={self.id}, device_id={self.device_id}, "
f"type={self.alarm_type})>"
)
class HeartbeatRecord(Base):
"""Heartbeat history from devices."""
__tablename__ = "heartbeat_records"
__table_args__ = (
Index("ix_heartbeat_device_time", "device_id", "created_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
imei: Mapped[str | None] = mapped_column(String(20), nullable=True)
protocol_number: Mapped[int] = mapped_column(Integer, nullable=False) # 0x13 or 0x36
terminal_info: Mapped[int] = mapped_column(Integer, nullable=False)
battery_level: Mapped[int] = mapped_column(Integer, nullable=False)
gsm_signal: Mapped[int] = mapped_column(Integer, nullable=False)
extension_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
device: Mapped["Device"] = relationship(back_populates="heartbeats")
def __repr__(self) -> str:
return f"<HeartbeatRecord(id={self.id}, device_id={self.device_id})>"
class AttendanceRecord(Base):
"""Attendance punch records from badges."""
__tablename__ = "attendance_records"
__table_args__ = (
Index("ix_attendance_device_time", "device_id", "recorded_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
imei: Mapped[str | None] = mapped_column(String(20), nullable=True)
attendance_type: Mapped[str] = mapped_column(
String(20), nullable=False
) # clock_in, clock_out
attendance_source: Mapped[str] = mapped_column(
String(20), nullable=False, default="device",
server_default="device",
) # device (0xB0/0xB1), bluetooth (0xB2), fence (auto)
protocol_number: Mapped[int] = mapped_column(Integer, nullable=False)
gps_positioned: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
speed: Mapped[float | None] = mapped_column(Float, nullable=True)
course: Mapped[float | None] = mapped_column(Float, nullable=True)
gps_satellites: Mapped[int | None] = mapped_column(Integer, nullable=True)
battery_level: Mapped[int | None] = mapped_column(Integer, nullable=True)
gsm_signal: Mapped[int | None] = mapped_column(Integer, nullable=True)
mcc: Mapped[int | None] = mapped_column(Integer, nullable=True)
mnc: Mapped[int | None] = mapped_column(Integer, nullable=True)
lac: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
cell_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
wifi_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
lbs_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
address: Mapped[str | None] = mapped_column(Text, nullable=True)
recorded_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
device: Mapped["Device"] = relationship(back_populates="attendance_records")
def __repr__(self) -> str:
return (
f"<AttendanceRecord(id={self.id}, device_id={self.device_id}, "
f"type={self.attendance_type})>"
)
class BluetoothRecord(Base):
"""Bluetooth punch card and location records."""
__tablename__ = "bluetooth_records"
__table_args__ = (
Index("ix_bluetooth_device_time", "device_id", "recorded_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
imei: Mapped[str | None] = mapped_column(String(20), nullable=True)
record_type: Mapped[str] = mapped_column(
String(20), nullable=False
) # punch, location
protocol_number: Mapped[int] = mapped_column(Integer, nullable=False)
beacon_mac: Mapped[str | None] = mapped_column(String(20), nullable=True)
beacon_uuid: Mapped[str | None] = mapped_column(String(36), nullable=True)
beacon_major: Mapped[int | None] = mapped_column(Integer, nullable=True)
beacon_minor: Mapped[int | None] = mapped_column(Integer, nullable=True)
rssi: Mapped[int | None] = mapped_column(Integer, nullable=True)
beacon_battery: Mapped[float | None] = mapped_column(Float, nullable=True)
beacon_battery_unit: Mapped[str | None] = mapped_column(String(10), nullable=True)
attendance_type: Mapped[str | None] = mapped_column(String(20), nullable=True)
bluetooth_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
recorded_at: Mapped[datetime] = mapped_column(DateTime, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
device: Mapped["Device"] = relationship(back_populates="bluetooth_records")
def __repr__(self) -> str:
return (
f"<BluetoothRecord(id={self.id}, device_id={self.device_id}, "
f"type={self.record_type})>"
)
class BeaconConfig(Base):
"""Registered Bluetooth beacon configuration for indoor positioning."""
__tablename__ = "beacon_configs"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
beacon_mac: Mapped[str] = mapped_column(String(20), unique=True, index=True, nullable=False)
beacon_uuid: Mapped[str | None] = mapped_column(String(36), nullable=True)
beacon_major: Mapped[int | None] = mapped_column(Integer, nullable=True)
beacon_minor: Mapped[int | None] = mapped_column(Integer, nullable=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
floor: Mapped[str | None] = mapped_column(String(20), nullable=True)
area: Mapped[str | None] = mapped_column(String(100), nullable=True)
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
address: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[str] = mapped_column(String(20), default="active", nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
updated_at: Mapped[datetime | None] = mapped_column(
DateTime, default=_utcnow, onupdate=_utcnow, nullable=True
)
def __repr__(self) -> str:
return f"<BeaconConfig(id={self.id}, mac={self.beacon_mac}, name={self.name})>"
class DeviceBeaconBinding(Base):
"""Many-to-many binding between devices and beacons."""
__tablename__ = "device_beacon_bindings"
__table_args__ = (
Index("ix_dbb_device_beacon", "device_id", "beacon_id", unique=True),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
beacon_id: Mapped[int] = mapped_column(
Integer, ForeignKey("beacon_configs.id", ondelete="CASCADE"), index=True, nullable=False
)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
def __repr__(self) -> str:
return f"<DeviceBeaconBinding(device_id={self.device_id}, beacon_id={self.beacon_id})>"
class FenceConfig(Base):
"""Geofence configuration for area monitoring."""
__tablename__ = "fence_configs"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
fence_type: Mapped[str] = mapped_column(String(20), nullable=False) # circle / polygon / rectangle
# Circle center (WGS-84) or polygon centroid for display
center_lat: Mapped[float | None] = mapped_column(Float, nullable=True)
center_lng: Mapped[float | None] = mapped_column(Float, nullable=True)
radius: Mapped[float | None] = mapped_column(Float, nullable=True) # meters, for circle
# Polygon/rectangle vertices as JSON: [[lng,lat], [lng,lat], ...] (WGS-84)
points: Mapped[str | None] = mapped_column(Text, nullable=True)
color: Mapped[str] = mapped_column(String(20), default="#3b82f6", nullable=False)
fill_color: Mapped[str | None] = mapped_column(String(20), nullable=True)
fill_opacity: Mapped[float] = mapped_column(Float, default=0.2, nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
is_active: Mapped[bool] = mapped_column(Integer, default=1, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
updated_at: Mapped[datetime | None] = mapped_column(
DateTime, default=_utcnow, onupdate=_utcnow, nullable=True
)
def __repr__(self) -> str:
return f"<FenceConfig(id={self.id}, name={self.name}, type={self.fence_type})>"
class DeviceFenceBinding(Base):
"""Many-to-many binding between devices and geofences."""
__tablename__ = "device_fence_bindings"
__table_args__ = (
Index("ix_dfb_device_fence", "device_id", "fence_id", unique=True),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
fence_id: Mapped[int] = mapped_column(
Integer, ForeignKey("fence_configs.id", ondelete="CASCADE"), index=True, nullable=False
)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
def __repr__(self) -> str:
return f"<DeviceFenceBinding(device_id={self.device_id}, fence_id={self.fence_id})>"
class DeviceFenceState(Base):
"""Runtime state tracking: is a device currently inside a fence?"""
__tablename__ = "device_fence_states"
__table_args__ = (
Index("ix_dfs_device_fence", "device_id", "fence_id", unique=True),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
fence_id: Mapped[int] = mapped_column(
Integer, ForeignKey("fence_configs.id", ondelete="CASCADE"), index=True, nullable=False
)
is_inside: Mapped[bool] = mapped_column(Integer, default=0, nullable=False)
last_transition_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
last_check_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
last_latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
last_longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
def __repr__(self) -> str:
return f"<DeviceFenceState(device_id={self.device_id}, fence_id={self.fence_id}, inside={self.is_inside})>"
class CommandLog(Base):
"""Log of commands sent to devices."""
__tablename__ = "command_logs"
__table_args__ = (
Index("ix_command_device_status", "device_id", "status"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
command_type: Mapped[str] = mapped_column(String(30), nullable=False)
command_content: Mapped[str] = mapped_column(Text, nullable=False)
server_flag: Mapped[str] = mapped_column(String(20), nullable=False)
response_content: Mapped[str | None] = mapped_column(Text, nullable=True)
status: Mapped[str] = mapped_column(
String(20), default="pending", nullable=False
) # pending, sent, success, failed
sent_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
response_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
device: Mapped["Device"] = relationship(back_populates="command_logs")
def __repr__(self) -> str:
return (
f"<CommandLog(id={self.id}, device_id={self.device_id}, "
f"type={self.command_type}, status={self.status})>"
)
class DeviceGroup(Base):
"""Device groups for organizing devices."""
__tablename__ = "device_groups"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(100), unique=True, nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
color: Mapped[str] = mapped_column(String(20), default="#3b82f6", nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
updated_at: Mapped[datetime | None] = mapped_column(
DateTime, default=_utcnow, onupdate=_utcnow, nullable=True
)
def __repr__(self) -> str:
return f"<DeviceGroup(id={self.id}, name={self.name})>"
class DeviceGroupMember(Base):
"""Many-to-many: devices belong to groups."""
__tablename__ = "device_group_members"
__table_args__ = (
Index("ix_dgm_device_group", "device_id", "group_id", unique=True),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
device_id: Mapped[int] = mapped_column(
Integer, ForeignKey("devices.id", ondelete="CASCADE"), index=True, nullable=False
)
group_id: Mapped[int] = mapped_column(
Integer, ForeignKey("device_groups.id", ondelete="CASCADE"), index=True, nullable=False
)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
def __repr__(self) -> str:
return f"<DeviceGroupMember(device_id={self.device_id}, group_id={self.group_id})>"
class AlertRule(Base):
"""Configurable alert rules for custom thresholds."""
__tablename__ = "alert_rules"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
rule_type: Mapped[str] = mapped_column(String(30), nullable=False)
# rule_type values: low_battery, no_heartbeat, fence_stay, speed_limit, offline_duration
conditions: Mapped[dict] = mapped_column(JSON, nullable=False)
# e.g. {"threshold": 20} for low_battery, {"minutes": 30} for no_heartbeat
is_active: Mapped[bool] = mapped_column(Integer, default=1, nullable=False)
device_ids: Mapped[str | None] = mapped_column(Text, nullable=True)
# comma-separated device IDs, null = all devices
group_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
# apply to a device group
description: Mapped[str | None] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
updated_at: Mapped[datetime | None] = mapped_column(
DateTime, default=_utcnow, onupdate=_utcnow, nullable=True
)
def __repr__(self) -> str:
return f"<AlertRule(id={self.id}, name={self.name}, type={self.rule_type})>"
class AuditLog(Base):
"""Audit trail for write operations (POST/PUT/DELETE)."""
__tablename__ = "audit_logs"
__table_args__ = (
Index("ix_audit_created", "created_at"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
method: Mapped[str] = mapped_column(String(10), nullable=False) # POST, PUT, DELETE
path: Mapped[str] = mapped_column(String(500), nullable=False)
status_code: Mapped[int] = mapped_column(Integer, nullable=False)
operator: Mapped[str | None] = mapped_column(String(100), nullable=True) # API key name or "master"
client_ip: Mapped[str | None] = mapped_column(String(50), nullable=True)
request_body: Mapped[dict | None] = mapped_column(JSON, nullable=True)
response_summary: Mapped[str | None] = mapped_column(String(500), nullable=True)
duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
def __repr__(self) -> str:
return f"<AuditLog(id={self.id}, {self.method} {self.path})>"
class ApiKey(Base):
"""API keys for external system authentication."""
__tablename__ = "api_keys"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
key_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
name: Mapped[str] = mapped_column(String(100), nullable=False)
permissions: Mapped[str] = mapped_column(
String(20), default="read", nullable=False
) # read, write, admin
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
last_used_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False)
def __repr__(self) -> str:
return f"<ApiKey(id={self.id}, name={self.name}, permissions={self.permissions})>"
Reference in New Issue View Git Blame Copy Permalink