139 lines
4.2 KiB
Python
139 lines
4.2 KiB
Python
"""
|
|
Audit Log Events
|
|
"""
|
|
|
|
from sqlalchemy import event, inspect
|
|
from sqlalchemy.orm import Session
|
|
from .services.service import AuditService
|
|
from .utils.serialization import serialize_for_json
|
|
from core.context import get_user_context
|
|
|
|
|
|
def register_audit_listeners(models_to_audit):
|
|
"""
|
|
Register SQLAlchemy listeners for given models
|
|
"""
|
|
for model in models_to_audit:
|
|
event.listen(model, "after_insert", after_insert_listener)
|
|
event.listen(model, "after_update", after_update_listener)
|
|
event.listen(model, "after_delete", after_delete_listener)
|
|
|
|
|
|
def _get_current_username():
|
|
try:
|
|
context = get_user_context()
|
|
if context:
|
|
# Token usually has 'preferred_username' or 'name' or 'sub'
|
|
return (
|
|
context.get("preferred_username")
|
|
or context.get("email")
|
|
or context.get("sub")
|
|
or "System"
|
|
)
|
|
except:
|
|
pass
|
|
return "System"
|
|
|
|
|
|
def after_insert_listener(mapper, connection, target):
|
|
"""
|
|
Listener for INSERT operations
|
|
"""
|
|
table_name = target.__tablename__
|
|
record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
|
|
username = _get_current_username()
|
|
company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
|
|
tenant_id = getattr(target, "tenant_id", None)
|
|
|
|
# Create a session bound to the connection
|
|
session = Session(bind=connection)
|
|
try:
|
|
AuditService.log_crud_operation(
|
|
db=session,
|
|
table_name=table_name,
|
|
operation_type="CREATE",
|
|
record_data=record_data,
|
|
username=username,
|
|
record_id=str(getattr(target, "id", "")),
|
|
company_id=company_id,
|
|
tenant_id=tenant_id,
|
|
)
|
|
except Exception as e:
|
|
print(f"Error logging insert: {e}")
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def after_update_listener(mapper, connection, target):
|
|
"""
|
|
Listener for UPDATE operations
|
|
"""
|
|
table_name = target.__tablename__
|
|
|
|
state = inspect(target)
|
|
changes = {}
|
|
old_values = {}
|
|
new_values = {}
|
|
|
|
for attr in state.attrs:
|
|
hist = attr.history
|
|
if hist.has_changes():
|
|
changes[attr.key] = hist.added[0] if hist.added else None
|
|
old_values[attr.key] = hist.deleted[0] if hist.deleted else None
|
|
new_values[attr.key] = hist.added[0] if hist.added else None
|
|
|
|
if not changes:
|
|
return
|
|
|
|
record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
|
|
username = _get_current_username()
|
|
company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
|
|
tenant_id = getattr(target, "tenant_id", None)
|
|
|
|
session = Session(bind=connection)
|
|
try:
|
|
AuditService.log_crud_operation(
|
|
db=session,
|
|
table_name=table_name,
|
|
operation_type="UPDATE",
|
|
record_data=record_data,
|
|
username=username,
|
|
record_id=str(getattr(target, "id", "")),
|
|
old_values=serialize_for_json(old_values),
|
|
new_values=serialize_for_json(new_values),
|
|
company_id=company_id,
|
|
tenant_id=tenant_id,
|
|
)
|
|
except Exception as e:
|
|
print(f"Error logging update: {e}")
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def after_delete_listener(mapper, connection, target):
|
|
"""
|
|
Listener for DELETE operations
|
|
"""
|
|
table_name = target.__tablename__
|
|
record_data = {c.name: getattr(target, c.name) for c in mapper.columns}
|
|
username = _get_current_username()
|
|
company_id = getattr(target, "company_id", None) or getattr(target, "id", None)
|
|
tenant_id = getattr(target, "tenant_id", None)
|
|
|
|
session = Session(bind=connection)
|
|
try:
|
|
AuditService.log_crud_operation(
|
|
db=session,
|
|
table_name=table_name,
|
|
operation_type="DELETE",
|
|
record_data=record_data,
|
|
username=username,
|
|
record_id=str(getattr(target, "id", "")),
|
|
company_id=company_id,
|
|
tenant_id=tenant_id,
|
|
)
|
|
except Exception as e:
|
|
print(f"Error logging delete: {e}")
|
|
finally:
|
|
session.close()
|