Shared Models

shared/
├── pyproject.toml                    # Library metadata and dependencies
└── src/
    └── aegis_shared/
        ├── __init__.py
        ├── models/
        │   ├── __init__.py           # Re-exports all public models
        │   ├── common.py             # AegisBase, AegisEntity, TenantEntity, enums
        │   ├── oilgas.py             # Oil & gas domain entities
        │   ├── agent.py              # Agent config, state, messages
        │   ├── compliance.py         # Compliance status and matrix models
        │   ├── checklist.py          # Filing checklist workflow models
        │   ├── rules.py              # Rule versioning and lifecycle models
        │   ├── entity_resolution.py  # ResolvedEntity + MatchedOn (R34 entity resolution)
        │   ├── events.py             # Operational event system models
        │   └── event_types.py        # Custom event type builder models
        └── db/
            ├── postgres.py           # PostgresPool (asyncpg wrapper)
            ├── redis.py              # RedisManager (async Redis wrapper)
            └── kafka.py              # Kafka producer manager

class AegisBase(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,       # Allow creating from ORM objects / dicts
        populate_by_name=True,      # Allow field aliases
        str_strip_whitespace=True,  # Auto-strip string whitespace
    )

class AegisEntity(AegisBase):
    id: UUID = Field(default_factory=uuid4)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

class TenantEntity(AegisEntity):
    tenant_id: str

from aegis_shared.db.postgres import PostgresPool
 
pg = PostgresPool(dsn="postgresql://aegis:aegis_local@localhost:5432/aegis")
await pg.connect()
 
# Convenience methods
row = await pg.fetchrow("SELECT * FROM agents WHERE id = $1", agent_id)
rows = await pg.fetch("SELECT * FROM compliance_status WHERE tenant_id = $1", tenant_id)
await pg.execute("INSERT INTO audit_logs (id, event_type) VALUES ($1, $2)", log_id, event_type)
 
# Direct connection access
async with pg.acquire() as conn:
    rows = await conn.fetch("SELECT * FROM agents")
 
await pg.disconnect()

from aegis_shared.db.redis import RedisManager
 
rm = RedisManager(url="redis://localhost:6379")
await rm.connect()
 
# Key-value operations
await rm.set("key", "value", ex=3600)  # with TTL
val = await rm.get("key")
await rm.delete("key")
 
# Injection ledger (Redis Hash at skill:ledger:{conversation_id})
await rm.ledger_mark("conv-1", "skill:spacing-calc", "1")
has = await rm.ledger_has("conv-1", "skill:spacing-calc")  # True
items = await rm.ledger_get_all("conv-1")  # {"skill:spacing-calc": "1"}
 
await rm.disconnect()

# For a request/response DTO
class MyRequest(AegisBase):
    entity_id: str
    action: str
 
# For a persistent entity without tenant scope
class MyEntity(AegisEntity):
    name: str
    status: str = "active"
 
# For a tenant-scoped persistent entity (most domain objects)
class MyDomainEntity(TenantEntity):
    name: str
    entity_type: str
    data: dict[str, Any] = Field(default_factory=dict)

from enum import StrEnum
from datetime import date, datetime
from typing import Any
from uuid import UUID
 
from pydantic import Field as PydanticField
 
from aegis_shared.models.common import AegisBase, TenantEntity
 
 
class MyStatus(StrEnum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    ARCHIVED = "archived"
 
 
class MyEntity(TenantEntity):
    # Required fields
    name: str
    entity_type: str
 
    # Optional fields with None default
    description: str | None = None
    parent_id: str | None = None
 
    # Enum fields with default
    status: MyStatus = MyStatus.ACTIVE
 
    # Date fields
    effective_date: date | None = None
    expires_at: datetime | None = None
 
    # Collections -- always use Field(default_factory=...)
    tags: list[str] = PydanticField(default_factory=list)
    metadata: dict[str, Any] = PydanticField(default_factory=dict)
 
    # Numeric fields
    score: float | None = None
    count: int = 0

from aegis_shared.models.my_module import MyEntity, MyStatus
 
__all__ = [
    # ... existing exports ...
    "MyEntity",
    "MyStatus",
]

from aegis_shared.models import MyEntity, MyStatus

cd services/orchestration-engine
poetry install