Components / audit

audit

Append-only activity logging for audit trails

v0.1.0 stable commands queries events hooks

ActivityLogRepo is a platform port for recording user actions. Each ActivityRecord captures actor, owner, action, entity, and timestamp. Records are append-only and immutable. Query by entity, actor, or owner for activity feeds, audit reports, and history views.

1
Ports
0
Schemas
2
Hooks
1
Events
01

Ports

Required
  • Clock
    Provide timestamps for activity records
Optional
  • EventBus optional
    Publish ActivityRecorded events
Adapters Provided
  • InMemoryActivityLogRepo
    implements ActivityLogRepo
    In-memory activity log for testing
02

Schemas

Defines

No schemas defined

Uses

No external schemas used

03

Hooks

  • before_append
    Before an activity record is appended to the log
    outputs: Veto, Patch
  • after_append
    After an activity record is appended to the log
    outputs: SideEffectRequest
04

Events

  • ActivityRecorded v1
    After an activity record is persisted
    payload: {record_id, actor_id, owner_id, action, entity_ref}
05

Stories

Story
Activity Trail Demo

Log activities and query by entity, actor, and owner
06

Examples

Log an activity from a use case 
from psp.platform.audit import ActivityRecord, EntityRef

class CreateTaskUseCase:
    def __init__(self, repo: TaskRepository, log: ActivityLogRepo) -> None:
        self._repo = repo
        self._log = log

    async def execute(self, actor_id: UUID, input: CreateTaskInput) -> Task:
        task = Task.create(owner_id=actor_id, title=input.title)
        await self._repo.add(task)

        # Log the activity
        self._log.append(ActivityRecord.create(
            actor_id=actor_id,
            owner_id=task.owner_id,
            action="create_task",
            entity_ref=EntityRef("Task", task.id),
            metadata={"title": task.title},
        ))
        return task

Record actions after successful operations.

Query entity history 
# Get last 10 activities for a task
entity = EntityRef("Task", task_id)
history = activity_log.get_by_entity(entity, limit=10)

for record in history:
    print(f"{record.occurred_at}: {record.action} by {record.actor_id}")
    # 2024-01-15 10:30: create_task by user-123
    # 2024-01-15 11:00: update_task by user-123
    # 2024-01-15 14:22: complete_task by user-456

Get activity timeline for a specific entity.

Build an activity feed 
# Get activities for data owned by a user
activities = activity_log.get_by_owner(user_id, limit=50)

# Format for UI
feed = [
    {
        "action": r.action,
        "entity": f"{r.entity_ref.entity_type}#{r.entity_ref.entity_id}",
        "actor": str(r.actor_id),
        "when": r.occurred_at.isoformat(),
        "details": r.metadata,
    }
    for r in activities
]

Show recent activities for a user's data.

Audit user actions 
# Get all activities by an actor (for audit)
user_actions = activity_log.get_by_actor(actor_id)

# Group by action type
from collections import Counter
action_counts = Counter(r.action for r in user_actions)
# Counter({'create_task': 45, 'complete_task': 32, 'delete_task': 3})

Query all actions by a specific user.

Test with InMemoryActivityLogRepo 
from psp.platform.audit import InMemoryActivityLogRepo

def test_task_creation_logs_activity():
    log = InMemoryActivityLogRepo()
    use_case = CreateTaskUseCase(repo=repo, log=log)

    task = use_case.execute(actor_id, CreateTaskInput(title="Test"))

    # Verify activity was logged
    assert len(log.records) == 1
    record = log.records[0]
    assert record.action == "create_task"
    assert record.entity_ref.entity_type == "Task"
    assert record.actor_id == actor_id

    # Or query by action type
    creates = log.get_by_action("create_task")
    assert len(creates) == 1

Verify activities are logged correctly.

API Reference

This component mounts routes under /v1/audit. View OpenAPI specification