95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from app.models.journal import JournalEntry, EntityType
|
|
from datetime import datetime
|
|
|
|
|
|
class CRUDJournal:
|
|
"""CRUD operations for the generic journal table.
|
|
|
|
This journal is immutable - entries can only be created (by backend) and queried.
|
|
There are no API endpoints for creating journal entries; the backend logs changes directly.
|
|
"""
|
|
|
|
def log_change(
|
|
self,
|
|
db: Session,
|
|
entity_type: EntityType,
|
|
entity_id: int,
|
|
entry: str,
|
|
user: str,
|
|
ip: Optional[str] = None
|
|
) -> JournalEntry:
|
|
"""Log a change to an entity. Internal backend use only."""
|
|
journal_entry = JournalEntry(
|
|
entity_type=entity_type.value,
|
|
entity_id=entity_id,
|
|
entry=entry,
|
|
user=user,
|
|
ip=ip,
|
|
entry_dt=datetime.utcnow()
|
|
)
|
|
db.add(journal_entry)
|
|
db.commit()
|
|
db.refresh(journal_entry)
|
|
return journal_entry
|
|
|
|
def get_entity_journal(
|
|
self,
|
|
db: Session,
|
|
entity_type: EntityType,
|
|
entity_id: int,
|
|
limit: int = 100
|
|
) -> List[JournalEntry]:
|
|
"""Get all journal entries for a specific entity. Read-only API endpoint."""
|
|
return db.query(JournalEntry).filter(
|
|
JournalEntry.entity_type == entity_type.value,
|
|
JournalEntry.entity_id == entity_id
|
|
).order_by(JournalEntry.entry_dt.desc()).limit(limit).all()
|
|
|
|
def get_user_journal(
|
|
self,
|
|
db: Session,
|
|
user: str,
|
|
limit: int = 100
|
|
) -> List[JournalEntry]:
|
|
"""Get all journal entries created by a specific user."""
|
|
return db.query(JournalEntry).filter(
|
|
JournalEntry.user == user
|
|
).order_by(JournalEntry.entry_dt.desc()).limit(limit).all()
|
|
|
|
# Convenience methods for backward compatibility with PPR journal
|
|
def log_ppr_change(
|
|
self,
|
|
db: Session,
|
|
ppr_id: int,
|
|
entry: str,
|
|
user: str,
|
|
ip: Optional[str] = None
|
|
) -> JournalEntry:
|
|
"""Log a change to a PPR (convenience method)."""
|
|
return self.log_change(
|
|
db=db,
|
|
entity_type=EntityType.PPR,
|
|
entity_id=ppr_id,
|
|
entry=entry,
|
|
user=user,
|
|
ip=ip
|
|
)
|
|
|
|
def get_ppr_journal(
|
|
self,
|
|
db: Session,
|
|
ppr_id: int,
|
|
limit: int = 100
|
|
) -> List[JournalEntry]:
|
|
"""Get all journal entries for a PPR (convenience method)."""
|
|
return self.get_entity_journal(
|
|
db=db,
|
|
entity_type=EntityType.PPR,
|
|
entity_id=ppr_id,
|
|
limit=limit
|
|
)
|
|
|
|
|
|
journal = CRUDJournal() |