orxtra v0.13.0 /services.src.orxtra.services._identity
Updated
On this page

Principal CRUD with the service-layer policy storage omits: kind validation, refusal to mint or delete the system singleton, and self-subscription creation for consumer/app-registered kinds.

#services.src.orxtra.services._identity

#services.src.orxtra.services._identity

Principal CRUD service functions.

Thin wrappers over PrincipalStorage that add the service-layer policy the storage deliberately omits: kind validation (via KindRegistry) and the refusal to delete the singleton system principal. Storage accepts any string kind and deletes any row; the enforcement point lives here.

#create_principal

python
async def create_principal(dispatch_backend: DispatchBackend, principal_storage: PrincipalStorage, kind_registry: KindRegistry, *, kind: str, external_ref: UUID, display_name: str | None=None, notification_event_types: list[str] | None=None) -> Principal

Validate the kind, then idempotently mint the principal.

The service layer is the kind-enforcement point: kind_registry.validate hard-errors on an unregistered kind before any row is written (storage itself accepts any string). Minting is idempotent on (kind, external_ref) -- a retry after a partial failure, or a second call with the same reference, returns the existing row rather than creating a duplicate.

kind == "system" is rejected outright: the system principal is a seeded singleton, not something the API mints. Allowing it would let a caller create a SECOND system-kind row under an arbitrary external_ref -- a row the delete path refuses to remove, leaving it permanently stuck.

For consumer and app-registered kinds, notification_event_types is required and triggers automatic self-subscription creation: the minted principal subscribes to events matching the given types with its own id as both the filter's principal_id and the NotifyAction target. For infrastructure kinds (run, source, system), self-subscriptions are rejected.

#_create_self_subscription

python
async def _create_self_subscription(backend: DispatchBackend, principal: Principal, event_types: list[str]) -> None

Create a self-subscription for a newly minted principal.

The subscription filters events by the given event types AND the principal's own id, delivering a NotifyAction back to the same principal. This is the mechanism by which consumers (and app-registered kinds) automatically receive notifications for events they care about.

#get_principal

python
async def get_principal(principal_storage: PrincipalStorage, *, principal_id: UUID) -> Principal | None

Fetch a principal by id, or None if it does not exist.

#list_principals

python
async def list_principals(principal_storage: PrincipalStorage, *, kind: str | None=None) -> list[Principal]

List principals, optionally filtered by kind.

#delete_principal

python
async def delete_principal(principal_storage: PrincipalStorage, *, principal_id: UUID) -> None

Delete a principal, refusing to delete the system principal.

The singleton system principal anchors framework-owned attribution and must never be removed, so it is fetched first and a kind == "system" match is a hard error. Any other principal is delegated to storage, where a PrincipalInUseError propagates if the row is still referenced.

#sweep_orphaned_run_principals

python
async def sweep_orphaned_run_principals(principal_storage: PrincipalStorage) -> int

Sweep kind=run principals with no matching runs row.

Delegates to storage with a conservative age guard to avoid racing with a concurrent start_run that has minted the principal but not yet created the runs row.

Search