Domain Models¶

The verity-common library (libs/verity-common/verity_common/models.py) defines all shared Pydantic domain models and enumerations used across the Verity platform. These models map directly to the PostgreSQL schema and are used for API serialisation, inter-service messaging, and ORM hydration.

Model Hierarchy¶

classDiagram
    class VerityModel {
        +ConfigDict from_attributes=True
        +ConfigDict populate_by_name=True
    }

    class Principal {
        +UUID id
        +str external_id
        +str source
        +PrincipalType type
        +str display_name
        +str email
        +str department
        +str job_title
        +UUID manager_id
        +UUID peer_group_id
        +bool is_active
        +datetime hired_at
        +datetime terminated_at
        +datetime last_seen_at
        +dict metadata
        +datetime created_at
        +datetime updated_at
    }

    class Asset {
        +UUID id
        +str platform
        +str platform_id
        +str asset_type
        +str fqn
        +str display_name
        +Sensitivity sensitivity
        +str sensitivity_source
        +bool contains_pii
        +int pii_column_count
        +UUID data_owner_id
        +UUID steward_id
        +bool is_active
        +dict metadata
        +datetime created_at
        +datetime updated_at
    }

    class AccessGrant {
        +UUID id
        +UUID principal_id
        +UUID asset_id
        +str platform
        +Privilege privilege
        +GrantMechanism grant_mechanism
        +str granted_via
        +datetime granted_at
        +UUID granted_by_id
        +bool is_active
        +datetime revoked_at
        +UUID revoked_by_id
        +datetime snapshot_at
        +dict metadata
    }

    class AccessScore {
        +datetime computed_at
        +UUID principal_id
        +UUID asset_id
        +int score
        +dict component_json
        +ScoreTrigger trigger
    }

    class AccessEvent {
        +datetime event_at
        +UUID event_id
        +UUID principal_id
        +UUID asset_id
        +str platform
        +str operation
        +str tool_context
        +str query_fingerprint
        +int rows_touched
        +int duration_ms
        +str source_ip
        +str raw_event_id
    }

    class ReviewPacket {
        +UUID id
        +UUID principal_id
        +UUID asset_id
        +list~UUID~ access_grant_ids
        +float decay_score
        +RiskLevel risk_level
        +dict evidence_json
        +Recommendation recommendation
        +UUID assigned_to_id
        +str workflow_id
        +ReviewStatus status
        +datetime due_at
        +datetime created_at
    }

    class ReviewDecision {
        +UUID id
        +UUID packet_id
        +Decision decision
        +UUID decided_by_id
        +datetime decided_at
        +str justification
        +datetime next_review_at
        +bool is_escalation
        +UUID escalated_to_id
    }

    VerityModel <|-- Principal
    VerityModel <|-- Asset
    VerityModel <|-- AccessGrant
    VerityModel <|-- AccessScore
    VerityModel <|-- AccessEvent
    VerityModel <|-- ReviewPacket
    VerityModel <|-- ReviewDecision

Base Class: `VerityModel`¶

All domain models inherit from VerityModel, which configures Pydantic for seamless SQLAlchemy ORM integration:

from pydantic import BaseModel, ConfigDict

class VerityModel(BaseModel):
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)

Setting	Purpose
`from_attributes=True`	Allows construction from SQLAlchemy ORM objects via attribute access
`populate_by_name=True`	Allows populating fields by their Python name even when an alias is set

ORM Compatibility: `metadata` Field¶

Several models use AliasChoices to handle the SQLAlchemy reserved name metadata:

from pydantic import AliasChoices, Field

metadata: dict = Field(
    default_factory=dict,
    validation_alias=AliasChoices("metadata_", "metadata")
)

SQLAlchemy reserves metadata as a class attribute on declarative models. The ORM column is named metadata_, and AliasChoices allows Pydantic to accept either metadata_ (from the ORM) or metadata (from JSON/API payloads).

Enumerations¶

PrincipalType¶

Classifies the type of identity principal.

class PrincipalType(str, Enum):
    USER = "user"
    SERVICE_PRINCIPAL = "service_principal"
    GROUP = "group"
    APPLICATION = "application"

Sensitivity¶

Data sensitivity classification for assets.

class Sensitivity(str, Enum):
    PII = "PII"
    FINANCIAL = "FINANCIAL"
    CONFIDENTIAL = "CONFIDENTIAL"
    INTERNAL = "INTERNAL"
    PUBLIC = "PUBLIC"
    UNKNOWN = "UNKNOWN"

Privilege¶

Access privilege levels.

class Privilege(str, Enum):
    READ = "READ"
    WRITE = "WRITE"
    EXECUTE = "EXECUTE"
    ADMIN = "ADMIN"

GrantMechanism¶

How the access grant was provisioned.

class GrantMechanism(str, Enum):
    DIRECT = "direct"
    GROUP = "group"
    ROLE = "role"
    POLICY = "policy"

ScoreTrigger¶

What triggered a decay score computation.

class ScoreTrigger(str, Enum):
    SCHEDULED = "scheduled"
    EVENT = "event"
    MANUAL = "manual"

Operation¶

Database operation types captured in access events.

class Operation(str, Enum):
    SELECT = "SELECT"
    INSERT = "INSERT"
    UPDATE = "UPDATE"
    DELETE = "DELETE"
    EXECUTE = "EXECUTE"

RiskLevel¶

Risk classification for review packets.

class RiskLevel(str, Enum):
    CRITICAL = "CRITICAL"
    HIGH = "HIGH"
    MEDIUM = "MEDIUM"
    LOW = "LOW"

Recommendation¶

System-generated recommendation for a review.

class Recommendation(str, Enum):
    REVOKE = "REVOKE"
    REVIEW = "REVIEW"
    CONFIRM = "CONFIRM"

ReviewStatus¶

Lifecycle status of a review packet.

class ReviewStatus(str, Enum):
    PENDING = "PENDING"
    DECIDED = "DECIDED"
    EXPIRED = "EXPIRED"
    ESCALATED = "ESCALATED"

Decision¶

The human decision on a review packet.

class Decision(str, Enum):
    CONFIRM = "CONFIRM"
    REVOKE = "REVOKE"
    ESCALATE = "ESCALATE"

Domain Models¶

Principal¶

Represents a user, service principal, group, or application identity.

class Principal(VerityModel):
    id: UUID
    external_id: str              # ID from the source platform (e.g. Azure AD object ID)
    source: str                   # Source platform (e.g. "azure_ad", "hr")
    type: PrincipalType
    display_name: str
    email: Optional[str] = None
    department: Optional[str] = None
    job_title: Optional[str] = None
    manager_id: Optional[UUID] = None
    peer_group_id: Optional[UUID] = None
    is_active: bool = True
    hired_at: Optional[datetime] = None
    terminated_at: Optional[datetime] = None
    last_seen_at: Optional[datetime] = None
    metadata: dict = Field(default_factory=dict,
                           validation_alias=AliasChoices("metadata_", "metadata"))
    created_at: datetime
    updated_at: datetime

Asset¶

Represents a data asset — a table, database, report, or any resource being accessed.

class Asset(VerityModel):
    id: UUID
    platform: str                 # Source platform (e.g. "fabric", "synapse")
    platform_id: str              # Platform-specific asset identifier
    asset_type: str               # e.g. "table", "database", "report"
    fqn: str                      # Fully qualified name
    display_name: str
    sensitivity: Sensitivity = Sensitivity.UNKNOWN
    sensitivity_source: Optional[str] = None
    contains_pii: bool = False
    pii_column_count: int = 0
    data_owner_id: Optional[UUID] = None
    steward_id: Optional[UUID] = None
    is_active: bool = True
    first_seen_at: Optional[datetime] = None
    last_seen_at: Optional[datetime] = None
    metadata: dict = Field(default_factory=dict,
                           validation_alias=AliasChoices("metadata_", "metadata"))
    created_at: datetime
    updated_at: datetime

AccessGrant¶

Represents an active or revoked access grant between a principal and an asset.

class AccessGrant(VerityModel):
    id: UUID
    principal_id: UUID
    asset_id: UUID
    platform: str
    privilege: Privilege
    grant_mechanism: GrantMechanism
    granted_via: Optional[str] = None     # e.g. "Data Engineers" group name
    granted_at: datetime
    granted_by_id: Optional[UUID] = None
    is_active: bool = True
    revoked_at: Optional[datetime] = None
    revoked_by_id: Optional[UUID] = None
    snapshot_at: Optional[datetime] = None
    metadata: dict = Field(default_factory=dict,
                           validation_alias=AliasChoices("metadata_", "metadata"))

AccessScore¶

A computed decay score (0–100) for a principal-asset pair. Lower scores indicate more decayed (less justified) access.

class AccessScore(VerityModel):
    computed_at: datetime
    principal_id: UUID
    asset_id: UUID
    score: int = Field(ge=0, le=100)      # 0 = fully decayed, 100 = actively used
    component_json: dict = Field(default_factory=dict)  # Breakdown of score components
    trigger: ScoreTrigger

AccessEvent¶

A normalised access event — a record of a principal using their access to an asset.

class AccessEvent(VerityModel):
    event_at: datetime
    event_id: UUID
    principal_id: UUID
    asset_id: UUID
    platform: str
    operation: str
    tool_context: Optional[str] = None       # e.g. "Power BI", "dbt"
    query_fingerprint: Optional[str] = None  # Hashed query pattern
    rows_touched: Optional[int] = None
    duration_ms: Optional[int] = None
    source_ip: Optional[str] = Field(None)
    raw_event_id: Optional[str] = None

IPv4Address Coercion

The source_ip field includes a field_validator that coerces PostgreSQL INET types (returned as IPv4Address objects) to plain strings:

@field_validator("source_ip", mode="before")
@classmethod
def _coerce_source_ip(cls, v: object) -> Optional[str]:
    if v is None:
        return None
    return str(v)

ReviewPacket¶

A bundled review request generated when a decay score crosses a risk threshold.

class ReviewPacket(VerityModel):
    id: UUID
    principal_id: UUID
    asset_id: UUID
    access_grant_ids: list[UUID] = Field(default_factory=list)
    decay_score: float
    risk_level: RiskLevel
    evidence_json: dict = Field(default_factory=dict)
    recommendation: Recommendation
    assigned_to_id: Optional[UUID] = None
    workflow_id: Optional[str] = None        # Temporal workflow ID
    status: ReviewStatus = ReviewStatus.PENDING
    due_at: Optional[datetime] = None
    created_at: datetime

ReviewDecision¶

A human decision on a review packet — confirm, revoke, or escalate.

class ReviewDecision(VerityModel):
    id: UUID
    packet_id: UUID
    decision: Decision
    decided_by_id: UUID
    decided_at: datetime
    justification: Optional[str] = None
    next_review_at: Optional[datetime] = None
    is_escalation: bool = False
    escalated_to_id: Optional[UUID] = None

Usage Examples¶

Constructing from a dict (API payload)¶

from verity_common.models import Principal, PrincipalType

principal = Principal(
    id="550e8400-e29b-41d4-a716-446655440000",
    external_id="user@company.com",
    source="azure_ad",
    type=PrincipalType.USER,
    display_name="Jane Smith",
    email="jane.smith@company.com",
    department="Engineering",
    created_at="2024-01-15T10:00:00Z",
    updated_at="2024-06-01T12:00:00Z",
)

Constructing from an ORM object¶

# SQLAlchemy ORM object with metadata_ column
orm_obj = await session.get(PrincipalORM, some_uuid)
principal = Principal.model_validate(orm_obj)
# Works because from_attributes=True and AliasChoices handles metadata_ → metadata

Serialising to JSON¶

print(principal.model_dump_json(indent=2))

Domain Models¶

Model Hierarchy¶

Base Class: VerityModel¶

ORM Compatibility: metadata Field¶

Enumerations¶

PrincipalType¶

Sensitivity¶

Privilege¶

GrantMechanism¶

ScoreTrigger¶

Operation¶

RiskLevel¶

Recommendation¶

ReviewStatus¶

Decision¶

Domain Models¶

Principal¶

Asset¶

AccessGrant¶

AccessScore¶

AccessEvent¶

ReviewPacket¶

ReviewDecision¶

Usage Examples¶

Constructing from a dict (API payload)¶

Constructing from an ORM object¶

Serialising to JSON¶

Base Class: `VerityModel`¶

ORM Compatibility: `metadata` Field¶