Add typed IDM relationship contracts

This commit is contained in:
2026-08-02 14:44:41 +02:00
parent 5783d43547
commit c6ef644842
3 changed files with 236 additions and 2 deletions
+8 -1
View File
@@ -207,13 +207,20 @@ Other stable runtime capabilities currently include:
- `identity.directory` and `identity.search`
- `organizations.directory`
- `idm.directory`
- `idm.directory`, `idm.function_assignments`, `idm.relationships`, and
`idm.assignment_lifecycle`
- `calendar.outbox`, `calendar.scheduling`, `calendar.invitations`, and
`calendar.externalProfiles`
- `poll.scheduling`
- `notifications.dispatch`
- `workflow.definitionContributions` and `workflow.runtimeWorker`
The provider-neutral `idm.relationships` contract carries tenant-scoped typed
groups, effective-dated identity relationships, and explicit membership
decisions. It deliberately does not expose IDM persistence models or imply an
Access permission. Consumers can retain source revisions and inclusion or
exclusion provenance while remaining optional-module safe.
Modules contribute reusable process baselines through
`ModuleManifest.workflow_definitions`. Each contribution pins its origin module
and version, stable key, schema and content hash, native graph/BPMN content,
+149 -1
View File
@@ -1,7 +1,7 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal, Protocol, runtime_checkable
@@ -10,9 +10,12 @@ IDM_MODULE_ID = "idm"
CAPABILITY_IDM_DIRECTORY = "idm.directory"
CAPABILITY_IDM_FUNCTION_ASSIGNMENTS = "idm.function_assignments"
CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE = "idm.assignment_lifecycle"
CAPABILITY_IDM_RELATIONSHIPS = "idm.relationships"
IdmStatus = Literal["active", "inactive", "suspended"]
OrganizationFunctionAssignmentSource = Literal["direct", "delegated", "acting_for", "directory", "governance", "system"]
TypedGroupStatus = Literal["active", "inactive"]
IdentityRelationshipStatus = Literal["active", "revoked"]
@dataclass(frozen=True, slots=True)
@@ -44,6 +47,78 @@ class OrganizationFunctionIncumbencyRef:
return not self.assignments
@dataclass(frozen=True, slots=True)
class TypedGroupRef:
"""Provider-neutral IDM group fact scoped to one tenant."""
id: str
tenant_id: str
key: str
name: str
group_type: str
description: str | None = None
status: TypedGroupStatus = "active"
source_provider: str = "local"
source_resource_type: str | None = None
source_resource_id: str | None = None
source_revision: str | None = None
properties: Mapping[str, object] = field(default_factory=dict)
provenance: Mapping[str, object] = field(default_factory=dict)
revision: int = 1
@dataclass(frozen=True, slots=True)
class IdentityRelationshipRef:
"""An effective-dated relationship from an identity to a typed target."""
id: str
tenant_id: str
relationship_kind: str
subject_identity_id: str
target_group_id: str | None = None
related_identity_id: str | None = None
role: str | None = None
valid_from: datetime | None = None
valid_until: datetime | None = None
status: IdentityRelationshipStatus = "active"
revoked_at: datetime | None = None
revoked_by: str | None = None
revocation_reason: str | None = None
source_provider: str = "local"
source_resource_type: str | None = None
source_resource_id: str | None = None
source_revision: str | None = None
properties: Mapping[str, object] = field(default_factory=dict)
provenance: Mapping[str, object] = field(default_factory=dict)
revision: int = 1
@dataclass(frozen=True, slots=True)
class IdentityRelationshipDecisionRef:
relationship: IdentityRelationshipRef
included: bool
code: str
explanation: str
identity_status: IdmStatus | None = None
@dataclass(frozen=True, slots=True)
class TypedGroupMembershipResolutionRef:
group: TypedGroupRef
effective_at: datetime
decisions: tuple[IdentityRelationshipDecisionRef, ...] = ()
@property
def identity_ids(self) -> tuple[str, ...]:
return tuple(
dict.fromkeys(
item.relationship.subject_identity_id
for item in self.decisions
if item.included
)
)
@runtime_checkable
class IdmDirectory(Protocol):
def get_organization_function_assignment(self, assignment_id: str) -> OrganizationFunctionAssignmentRef | None:
@@ -109,6 +184,79 @@ class IdmFunctionAssignmentDirectory(Protocol):
...
@runtime_checkable
class IdmRelationshipDirectory(Protocol):
"""Tenant-safe forward/reverse lookup for typed IDM relationships."""
def get_typed_group(
self,
group_id: str,
*,
tenant_id: str | None = None,
) -> TypedGroupRef | None:
...
def list_typed_groups(
self,
*,
tenant_id: str,
query: str | None = None,
group_types: Sequence[str] = (),
include_inactive: bool = False,
limit: int = 100,
) -> Sequence[TypedGroupRef]:
...
def identity_relationships_for_identity(
self,
identity_id: str,
*,
tenant_id: str,
effective_at: datetime | None = None,
relationship_kinds: Sequence[str] = (),
) -> Sequence[IdentityRelationshipRef]:
...
def identity_relationships_for_identities(
self,
identity_ids: Sequence[str],
*,
tenant_id: str,
effective_at: datetime | None = None,
relationship_kinds: Sequence[str] = (),
) -> Mapping[str, Sequence[IdentityRelationshipRef]]:
...
def identity_relationships_for_group(
self,
group_id: str,
*,
tenant_id: str,
effective_at: datetime | None = None,
relationship_kinds: Sequence[str] = (),
) -> Sequence[IdentityRelationshipRef]:
...
def identity_relationships_for_groups(
self,
group_ids: Sequence[str],
*,
tenant_id: str,
effective_at: datetime | None = None,
relationship_kinds: Sequence[str] = (),
) -> Mapping[str, Sequence[IdentityRelationshipRef]]:
...
def resolve_typed_group_memberships(
self,
group_ids: Sequence[str],
*,
tenant_id: str,
effective_at: datetime | None = None,
relationship_kinds: Sequence[str] = ("member",),
) -> Mapping[str, TypedGroupMembershipResolutionRef]:
...
@runtime_checkable
class IdmAssignmentLifecycle(Protocol):
"""Worker boundary for time-driven function-assignment transitions."""
+79
View File
@@ -0,0 +1,79 @@
from __future__ import annotations
import unittest
from datetime import datetime, timezone
from govoplan_core.core.idm import (
IdentityRelationshipDecisionRef,
IdentityRelationshipRef,
IdmRelationshipDirectory,
TypedGroupMembershipResolutionRef,
TypedGroupRef,
)
class RelationshipDirectoryStub:
def get_typed_group(self, group_id, *, tenant_id=None):
return None
def list_typed_groups(self, **kwargs):
return ()
def identity_relationships_for_identity(self, identity_id, **kwargs):
return ()
def identity_relationships_for_identities(self, identity_ids, **kwargs):
return {identity_id: () for identity_id in identity_ids}
def identity_relationships_for_group(self, group_id, **kwargs):
return ()
def identity_relationships_for_groups(self, group_ids, **kwargs):
return {group_id: () for group_id in group_ids}
def resolve_typed_group_memberships(self, group_ids, **kwargs):
return {}
class IdmRelationshipContractTests(unittest.TestCase):
def test_runtime_protocol_and_resolution_identity_projection(self) -> None:
self.assertIsInstance(RelationshipDirectoryStub(), IdmRelationshipDirectory)
relationship = IdentityRelationshipRef(
id="relationship-1",
tenant_id="tenant-1",
relationship_kind="member",
subject_identity_id="identity-1",
target_group_id="group-1",
)
resolution = TypedGroupMembershipResolutionRef(
group=TypedGroupRef(
id="group-1",
tenant_id="tenant-1",
key="group",
name="Group",
group_type="business_status",
),
effective_at=datetime(2026, 8, 2, tzinfo=timezone.utc),
decisions=(
IdentityRelationshipDecisionRef(
relationship=relationship,
included=True,
code="relationship.effective",
explanation="The relationship is effective.",
identity_status="active",
),
IdentityRelationshipDecisionRef(
relationship=relationship,
included=False,
code="relationship.revoked",
explanation="The relationship was revoked.",
identity_status="active",
),
),
)
self.assertEqual(("identity-1",), resolution.identity_ids)
if __name__ == "__main__":
unittest.main()