2025-12-10 12:02:17 +08:00

69 lines
2.6 KiB
Python

from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Optional
from sqlalchemy import DateTime, Enum as SAEnum, ForeignKey, Index, JSON, String, text
from sqlalchemy.dialects.mysql import BIGINT
from sqlalchemy.orm import Mapped, mapped_column, relationship
from backend.db.base import Base
class AuditAction(str, Enum):
LOGIN = "LOGIN"
LOGOUT = "LOGOUT"
INSTANCE_CREATE = "INSTANCE_CREATE"
INSTANCE_START = "INSTANCE_START"
INSTANCE_STOP = "INSTANCE_STOP"
INSTANCE_REBOOT = "INSTANCE_REBOOT"
INSTANCE_TERMINATE = "INSTANCE_TERMINATE"
INSTANCE_SYNC = "INSTANCE_SYNC"
CREDENTIAL_CREATE = "CREDENTIAL_CREATE"
CREDENTIAL_UPDATE = "CREDENTIAL_UPDATE"
CREDENTIAL_DELETE = "CREDENTIAL_DELETE"
CUSTOMER_CREATE = "CUSTOMER_CREATE"
CUSTOMER_UPDATE = "CUSTOMER_UPDATE"
CUSTOMER_DELETE = "CUSTOMER_DELETE"
USER_CREATE = "USER_CREATE"
USER_UPDATE = "USER_UPDATE"
USER_DELETE = "USER_DELETE"
OTHER = "OTHER"
class AuditResourceType(str, Enum):
USER = "USER"
CUSTOMER = "CUSTOMER"
AWS_CREDENTIAL = "AWS_CREDENTIAL"
INSTANCE = "INSTANCE"
JOB = "JOB"
OTHER = "OTHER"
class AuditLog(Base):
__tablename__ = "audit_logs"
__table_args__ = (
Index("idx_audit_customer", "customer_id"),
Index("idx_audit_action", "action"),
Index("idx_audit_created_at", "created_at"),
Index("idx_audit_resource", "resource_type", "resource_id"),
)
id: Mapped[int] = mapped_column(BIGINT(unsigned=True), primary_key=True, autoincrement=True)
user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.id", ondelete="SET NULL", onupdate="CASCADE"))
customer_id: Mapped[Optional[int]] = mapped_column(
ForeignKey("customers.id", ondelete="SET NULL", onupdate="CASCADE")
)
action: Mapped[AuditAction] = mapped_column(SAEnum(AuditAction), nullable=False)
resource_type: Mapped[AuditResourceType] = mapped_column(SAEnum(AuditResourceType), nullable=False)
resource_id: Mapped[Optional[int]] = mapped_column(BIGINT(unsigned=True))
description: Mapped[Optional[str]] = mapped_column(String(512))
payload: Mapped[Optional[dict]] = mapped_column(JSON)
ip_address: Mapped[Optional[str]] = mapped_column(String(45))
user_agent: Mapped[Optional[str]] = mapped_column(String(255))
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP"), nullable=False)
user: Mapped[Optional["User"]] = relationship("User")
customer: Mapped[Optional["Customer"]] = relationship("Customer")