Files
orion/models/database/admin.py
Samir Boulahtit eeafe6389f fix: resolve all remaining legacy import issues
- Update models/database/__init__.py to import from module locations
- Update models/schema/__init__.py to remove deleted modules
- Update models/__init__.py to import Inventory from module
- Remove duplicate AdminNotification from models/database/admin.py
- Fix monitoring module to import AdminNotification from messaging
- Update stats schema imports in admin/vendor API
- Update notification schema imports
- Add order_item_exception.py schema to orders module
- Fix app/api/v1/__init__.py to use storefront instead of shop
- Add cms_admin_pages import to main.py
- Fix password_reset_token imports
- Fix AdminNotification test imports

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-30 09:21:29 +01:00

199 lines
7.1 KiB
Python

# Admin-specific models
# models/database/admin.py
"""
Admin-specific database models.
This module provides models for:
- Admin audit logging (compliance and security tracking)
- Admin notifications (system alerts and warnings)
- Platform settings (global configuration)
- Platform alerts (system-wide issues)
- Application logs (critical events logging)
"""
from sqlalchemy import (
JSON,
Boolean,
Column,
DateTime,
ForeignKey,
Integer,
String,
Text,
)
from sqlalchemy.orm import relationship
from app.core.database import Base
from .base import TimestampMixin
class AdminAuditLog(Base, TimestampMixin):
"""
Track all admin actions for compliance and security.
Separate from regular audit logs - focuses on admin-specific operations
like vendor creation, user management, and system configuration changes.
"""
__tablename__ = "admin_audit_logs"
id = Column(Integer, primary_key=True, index=True)
admin_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
action = Column(
String(100), nullable=False, index=True
) # create_vendor, delete_vendor, etc.
target_type = Column(
String(50), nullable=False, index=True
) # vendor, user, import_job, setting
target_id = Column(String(100), nullable=False, index=True)
details = Column(JSON) # Additional context about the action
ip_address = Column(String(45)) # IPv4 or IPv6
user_agent = Column(Text)
request_id = Column(String(100)) # For correlating with application logs
# Relationships
admin_user = relationship("User", foreign_keys=[admin_user_id])
def __repr__(self):
return f"<AdminAuditLog(id={self.id}, action='{self.action}', target={self.target_type}:{self.target_id})>"
# AdminNotification has been moved to app/modules/messaging/models/admin_notification.py
# It's re-exported via models/database/__init__.py for backwards compatibility
class AdminSetting(Base, TimestampMixin):
"""
Platform-wide admin settings and configuration.
Stores global settings that affect the entire platform, different from
vendor-specific settings. Supports encryption for sensitive values.
Examples:
- max_vendors_allowed
- maintenance_mode
- default_vendor_trial_days
- smtp_settings
- stripe_api_keys (encrypted)
"""
__tablename__ = "admin_settings"
id = Column(Integer, primary_key=True, index=True)
key = Column(String(100), unique=True, nullable=False, index=True)
value = Column(Text, nullable=False)
value_type = Column(String(20), default="string") # string, integer, boolean, json
category = Column(
String(50), index=True
) # system, security, marketplace, notifications
description = Column(Text)
is_encrypted = Column(Boolean, default=False)
is_public = Column(Boolean, default=False) # Can be exposed to frontend?
last_modified_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
# Relationships
last_modified_by = relationship("User", foreign_keys=[last_modified_by_user_id])
def __repr__(self):
return f"<AdminSetting(key='{self.key}', category='{self.category}')>"
class PlatformAlert(Base, TimestampMixin):
"""
System-wide alerts that admins need to be aware of.
Tracks platform issues, performance problems, security incidents,
and other system-level concerns that require admin attention.
"""
__tablename__ = "platform_alerts"
id = Column(Integer, primary_key=True, index=True)
alert_type = Column(
String(50), nullable=False, index=True
) # security, performance, capacity, integration
severity = Column(
String(20), nullable=False, index=True
) # info, warning, error, critical
title = Column(String(200), nullable=False)
description = Column(Text)
affected_vendors = Column(JSON) # List of affected vendor IDs
affected_systems = Column(JSON) # List of affected system components
is_resolved = Column(Boolean, default=False, index=True)
resolved_at = Column(DateTime, nullable=True)
resolved_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
resolution_notes = Column(Text)
auto_generated = Column(Boolean, default=True) # System-generated vs manual
occurrence_count = Column(Integer, default=1) # Track repeated occurrences
first_occurred_at = Column(DateTime, nullable=False)
last_occurred_at = Column(DateTime, nullable=False)
# Relationships
resolved_by = relationship("User", foreign_keys=[resolved_by_user_id])
def __repr__(self):
return f"<PlatformAlert(id={self.id}, type='{self.alert_type}', severity='{self.severity}')>"
class AdminSession(Base, TimestampMixin):
"""
Track admin login sessions for security monitoring.
Helps identify suspicious login patterns, track concurrent sessions,
and enforce session policies for admin users.
"""
__tablename__ = "admin_sessions"
id = Column(Integer, primary_key=True, index=True)
admin_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
session_token = Column(String(255), unique=True, nullable=False, index=True)
ip_address = Column(String(45), nullable=False)
user_agent = Column(Text)
login_at = Column(DateTime, nullable=False, index=True)
last_activity_at = Column(DateTime, nullable=False)
logout_at = Column(DateTime, nullable=True)
is_active = Column(Boolean, default=True, index=True)
logout_reason = Column(String(50)) # manual, timeout, forced, suspicious
# Relationships
admin_user = relationship("User", foreign_keys=[admin_user_id])
def __repr__(self):
return f"<AdminSession(id={self.id}, admin_user_id={self.admin_user_id}, is_active={self.is_active})>"
class ApplicationLog(Base, TimestampMixin):
"""
Application-level logs stored in database for critical events.
Stores WARNING, ERROR, and CRITICAL level logs for easy searching,
filtering, and compliance. INFO and DEBUG logs are kept in files only.
"""
__tablename__ = "application_logs"
id = Column(Integer, primary_key=True, index=True)
timestamp = Column(DateTime, nullable=False, index=True)
level = Column(String(20), nullable=False, index=True) # WARNING, ERROR, CRITICAL
logger_name = Column(String(200), nullable=False, index=True)
module = Column(String(200))
function_name = Column(String(100))
line_number = Column(Integer)
message = Column(Text, nullable=False)
exception_type = Column(String(200))
exception_message = Column(Text)
stack_trace = Column(Text)
request_id = Column(String(100), index=True) # For correlating logs
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
vendor_id = Column(Integer, ForeignKey("vendors.id"), nullable=True, index=True)
context = Column(JSON) # Additional context data
# Relationships
user = relationship("User", foreign_keys=[user_id])
vendor = relationship("Vendor", foreign_keys=[vendor_id])
def __repr__(self):
return f"<ApplicationLog(id={self.id}, level='{self.level}', logger='{self.logger_name}')>"