750 lines
26 KiB
Python
750 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.dsar import (
|
|
DsarErasureActionRef,
|
|
DsarExecutionResultRef,
|
|
DsarRecordRef,
|
|
DsarSubjectRef,
|
|
dsar_capability_name,
|
|
)
|
|
from govoplan_idm.backend.db.models import (
|
|
IdmFunctionAssignmentChange,
|
|
IdmFunctionAssignmentChangeEvent,
|
|
IdmIdentityRelationship,
|
|
IdmOrganizationFunctionAssignment,
|
|
IdmTypedGroup,
|
|
)
|
|
|
|
|
|
IDM_DSAR_CAPABILITY = dsar_capability_name("idm")
|
|
_MAX_RECORDS = 5_000
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _SubjectSelectors:
|
|
account_id: str | None
|
|
identity_id: str | None
|
|
references: Mapping[str, str]
|
|
|
|
@property
|
|
def has_canonical_selector(self) -> bool:
|
|
return bool(self.account_id or self.identity_id)
|
|
|
|
|
|
class IdmDsarProvider:
|
|
provider_id = "idm"
|
|
module_id = "idm"
|
|
|
|
def search_subject(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
) -> Sequence[DsarRecordRef]:
|
|
db = _session(session)
|
|
selectors = _subject_selectors(subject)
|
|
if selectors is None:
|
|
return ()
|
|
|
|
assignments = _matching_assignments(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
selectors=selectors,
|
|
)
|
|
relationships = _matching_relationships(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
selectors=selectors,
|
|
)
|
|
changes = _matching_changes(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
selectors=selectors,
|
|
)
|
|
if _direct_reference_conflicts(
|
|
selectors,
|
|
assignments=assignments,
|
|
relationships=relationships,
|
|
changes=changes,
|
|
):
|
|
return ()
|
|
|
|
records: list[DsarRecordRef] = []
|
|
seen: set[tuple[str, str]] = set()
|
|
|
|
def append(record: DsarRecordRef) -> None:
|
|
key = (record.resource_type, record.resource_id)
|
|
if key in seen:
|
|
return
|
|
if len(records) >= _MAX_RECORDS:
|
|
raise ValueError(
|
|
"IDM DSAR result limit exceeded; narrow the subject selectors."
|
|
)
|
|
seen.add(key)
|
|
records.append(record)
|
|
|
|
for assignment in assignments:
|
|
match_fields = _assignment_match_fields(assignment, selectors)
|
|
append(
|
|
_record(
|
|
"idm_function_assignment",
|
|
assignment.id,
|
|
"institutional_function_fact",
|
|
"IDM organization-function assignment",
|
|
{
|
|
"match_fields": match_fields,
|
|
"identity_id": (
|
|
assignment.identity_id
|
|
if "identity_id" in match_fields
|
|
else None
|
|
),
|
|
"account_id": (
|
|
assignment.account_id
|
|
if "account_id" in match_fields
|
|
else None
|
|
),
|
|
"function_id": assignment.function_id,
|
|
"organization_unit_id": assignment.organization_unit_id,
|
|
"applies_to_subunits": assignment.applies_to_subunits,
|
|
"source": assignment.source,
|
|
"has_delegated_source": bool(
|
|
assignment.delegated_from_assignment_id
|
|
),
|
|
"acting_for_account_id": (
|
|
assignment.acting_for_account_id
|
|
if "acting_for_account_id" in match_fields
|
|
else None
|
|
),
|
|
"valid_from": _iso(assignment.valid_from),
|
|
"valid_until": _iso(assignment.valid_until),
|
|
"expired_event_at": _iso(assignment.expired_event_at),
|
|
"is_active": assignment.is_active,
|
|
},
|
|
observed_at=assignment.updated_at,
|
|
)
|
|
)
|
|
|
|
group_ids = {
|
|
row.target_group_id
|
|
for row in relationships
|
|
if row.target_group_id is not None
|
|
}
|
|
groups = {
|
|
row.id: row
|
|
for row in _rows_by_ids(
|
|
db,
|
|
IdmTypedGroup,
|
|
tenant_id=tenant_id,
|
|
ids=group_ids,
|
|
)
|
|
}
|
|
for group in groups.values():
|
|
append(
|
|
_record(
|
|
"idm_typed_group_context",
|
|
group.id,
|
|
"typed_relationship_context",
|
|
"IDM typed-group relationship context",
|
|
{
|
|
"key": group.key,
|
|
"name": _bounded_text(group.name, 255),
|
|
"group_type": group.group_type,
|
|
"status": group.status,
|
|
"source_provider": group.source_provider,
|
|
"revision": group.revision,
|
|
},
|
|
observed_at=group.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"The minimized typed-group definition is retained as context "
|
|
"for the subject's effective-dated relationship evidence."
|
|
),
|
|
)
|
|
)
|
|
for relationship in relationships:
|
|
match_fields = _relationship_match_fields(relationship, selectors)
|
|
group = groups.get(relationship.target_group_id or "")
|
|
append(
|
|
_record(
|
|
"idm_identity_relationship",
|
|
relationship.id,
|
|
"typed_identity_relationship",
|
|
"IDM typed identity relationship",
|
|
{
|
|
"match_fields": match_fields,
|
|
"relationship_kind": relationship.relationship_kind,
|
|
"subject_identity_id": (
|
|
relationship.subject_identity_id
|
|
if "subject_identity_id" in match_fields
|
|
else None
|
|
),
|
|
"target_group_id": relationship.target_group_id,
|
|
"target_group_key": group.key if group else None,
|
|
"target_group_name": _bounded_text(
|
|
group.name if group else None,
|
|
255,
|
|
),
|
|
"target_group_type": group.group_type if group else None,
|
|
"related_identity_id": (
|
|
relationship.related_identity_id
|
|
if "related_identity_id" in match_fields
|
|
else None
|
|
),
|
|
"role": _bounded_text(relationship.role, 120),
|
|
"valid_from": _iso(relationship.valid_from),
|
|
"valid_until": _iso(relationship.valid_until),
|
|
"status": relationship.status,
|
|
"revoked_at": _iso(relationship.revoked_at),
|
|
"revoked_by": (
|
|
relationship.revoked_by
|
|
if relationship.revoked_by == selectors.account_id
|
|
else None
|
|
),
|
|
"revocation_reason": _bounded_text(
|
|
relationship.revocation_reason,
|
|
2_000,
|
|
),
|
|
"expired_event_at": _iso(relationship.expired_event_at),
|
|
"source_provider": relationship.source_provider,
|
|
"revision": relationship.revision,
|
|
},
|
|
observed_at=relationship.updated_at,
|
|
)
|
|
)
|
|
|
|
change_ids: set[str] = set()
|
|
for change in changes:
|
|
change_ids.add(change.id)
|
|
match_fields = _change_match_fields(change, selectors)
|
|
append(
|
|
_record(
|
|
"idm_function_assignment_change",
|
|
change.id,
|
|
"function_assignment_governance_evidence",
|
|
"IDM governed function-assignment change",
|
|
{
|
|
"match_fields": match_fields,
|
|
"kind": change.kind,
|
|
"state": change.state,
|
|
"profile": change.profile,
|
|
"function_id": change.function_id,
|
|
"organization_unit_id": change.organization_unit_id,
|
|
"candidate_identity_id": (
|
|
change.candidate_identity_id
|
|
if "candidate_identity_id" in match_fields
|
|
else None
|
|
),
|
|
"candidate_account_id": (
|
|
change.candidate_account_id
|
|
if "candidate_account_id" in match_fields
|
|
else None
|
|
),
|
|
"initiator_account_id": (
|
|
change.initiator_account_id
|
|
if "initiator_account_id" in match_fields
|
|
else None
|
|
),
|
|
"initiator_identity_id": (
|
|
change.initiator_identity_id
|
|
if "initiator_identity_id" in match_fields
|
|
else None
|
|
),
|
|
"has_represented_assignment": bool(
|
|
change.represented_assignment_id
|
|
),
|
|
"requested_valid_from": _iso(change.requested_valid_from),
|
|
"requested_valid_until": _iso(change.requested_valid_until),
|
|
"applies_to_subunits": change.applies_to_subunits,
|
|
"assignment_source": change.assignment_source,
|
|
"resulting_assignment_id": change.resulting_assignment_id,
|
|
"expires_at": _iso(change.expires_at),
|
|
"resource_revision": change.resource_revision,
|
|
},
|
|
observed_at=change.updated_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Governed assignment requests and grants retain their state, "
|
|
"subject linkage, and outcome as institutional decision evidence."
|
|
),
|
|
)
|
|
)
|
|
|
|
for event in _matching_change_events(
|
|
db,
|
|
tenant_id=tenant_id,
|
|
selectors=selectors,
|
|
change_ids=change_ids,
|
|
):
|
|
match_fields = []
|
|
if event.change_id in change_ids:
|
|
match_fields.append("change_id")
|
|
if event.actor_account_id == selectors.account_id:
|
|
match_fields.append("actor_account_id")
|
|
if event.actor_identity_id == selectors.identity_id:
|
|
match_fields.append("actor_identity_id")
|
|
append(
|
|
_record(
|
|
"idm_function_assignment_change_event",
|
|
event.id,
|
|
"function_assignment_governance_evidence",
|
|
"IDM function-assignment lifecycle event",
|
|
{
|
|
"match_fields": match_fields,
|
|
"change_id": event.change_id,
|
|
"sequence": event.sequence,
|
|
"action": event.action,
|
|
"from_state": event.from_state,
|
|
"to_state": event.to_state,
|
|
"actor_account_id": (
|
|
event.actor_account_id
|
|
if "actor_account_id" in match_fields
|
|
else None
|
|
),
|
|
"actor_identity_id": (
|
|
event.actor_identity_id
|
|
if "actor_identity_id" in match_fields
|
|
else None
|
|
),
|
|
"actor_assignment_id": (
|
|
event.actor_assignment_id
|
|
if (
|
|
"actor_account_id" in match_fields
|
|
or "actor_identity_id" in match_fields
|
|
)
|
|
else None
|
|
),
|
|
"created_at": _iso(event.created_at),
|
|
},
|
|
observed_at=event.created_at,
|
|
immutable=True,
|
|
retention_reason=(
|
|
"Assignment lifecycle events are immutable decision and "
|
|
"accountability evidence."
|
|
),
|
|
)
|
|
)
|
|
|
|
return tuple(records)
|
|
|
|
def plan_erasure(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
records: Sequence[DsarRecordRef],
|
|
) -> Sequence[DsarErasureActionRef]:
|
|
del tenant_id
|
|
_session(session)
|
|
if _subject_selectors(subject) is None:
|
|
raise ValueError("IDM DSAR subject selectors conflict.")
|
|
actions: list[DsarErasureActionRef] = []
|
|
for record in records:
|
|
_validate_record(record)
|
|
if record.immutable_evidence:
|
|
kind = "retain"
|
|
rationale = record.retention_reason or (
|
|
"IDM governance evidence must be retained."
|
|
)
|
|
title = f"Retain {record.title}"
|
|
else:
|
|
kind = "manual_review"
|
|
rationale = (
|
|
"Function assignments and typed relationships are effective-dated "
|
|
"institutional facts. An authorized IDM operator must correct, "
|
|
"revoke, deactivate, or expire them through the governed lifecycle "
|
|
"after reviewing organizational and third-party consequences."
|
|
)
|
|
title = f"Review {record.title}"
|
|
actions.append(
|
|
DsarErasureActionRef(
|
|
action_id=f"idm:{kind}:{record.resource_type}:{record.resource_id}",
|
|
provider_id=self.provider_id,
|
|
module_id=self.module_id,
|
|
kind=kind,
|
|
resource_type=record.resource_type,
|
|
resource_id=record.resource_id,
|
|
title=title,
|
|
rationale=rationale,
|
|
executable=False,
|
|
)
|
|
)
|
|
return tuple(actions)
|
|
|
|
def execute_erasure(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
actions: Sequence[DsarErasureActionRef],
|
|
request_id: str,
|
|
) -> Sequence[DsarExecutionResultRef]:
|
|
del tenant_id
|
|
_session(session)
|
|
if _subject_selectors(subject) is None:
|
|
raise ValueError("IDM DSAR subject selectors conflict.")
|
|
results: list[DsarExecutionResultRef] = []
|
|
for action in actions:
|
|
_validate_action(action)
|
|
if action.executable:
|
|
raise ValueError(
|
|
"IDM DSAR does not publish executable erasure actions."
|
|
)
|
|
results.append(
|
|
DsarExecutionResultRef(
|
|
action_id=action.action_id,
|
|
status="blocked",
|
|
summary=(
|
|
"Use the governed IDM assignment or relationship lifecycle "
|
|
"after organizational, evidence, and third-party review."
|
|
),
|
|
evidence={"request_id": request_id},
|
|
)
|
|
)
|
|
return tuple(results)
|
|
|
|
|
|
def _matching_assignments(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
selectors: _SubjectSelectors,
|
|
) -> list[IdmOrganizationFunctionAssignment]:
|
|
conditions = []
|
|
if selectors.identity_id:
|
|
conditions.append(
|
|
IdmOrganizationFunctionAssignment.identity_id == selectors.identity_id
|
|
)
|
|
if selectors.account_id:
|
|
conditions.extend(
|
|
(
|
|
IdmOrganizationFunctionAssignment.account_id == selectors.account_id,
|
|
IdmOrganizationFunctionAssignment.acting_for_account_id
|
|
== selectors.account_id,
|
|
)
|
|
)
|
|
if reference := selectors.references.get("assignment"):
|
|
conditions.append(IdmOrganizationFunctionAssignment.id == reference)
|
|
return _query_conditions(
|
|
session,
|
|
IdmOrganizationFunctionAssignment,
|
|
tenant_id=tenant_id,
|
|
conditions=conditions,
|
|
)
|
|
|
|
|
|
def _matching_relationships(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
selectors: _SubjectSelectors,
|
|
) -> list[IdmIdentityRelationship]:
|
|
conditions = []
|
|
if selectors.identity_id:
|
|
conditions.extend(
|
|
(
|
|
IdmIdentityRelationship.subject_identity_id == selectors.identity_id,
|
|
IdmIdentityRelationship.related_identity_id == selectors.identity_id,
|
|
)
|
|
)
|
|
if reference := selectors.references.get("relationship"):
|
|
conditions.append(IdmIdentityRelationship.id == reference)
|
|
return _query_conditions(
|
|
session,
|
|
IdmIdentityRelationship,
|
|
tenant_id=tenant_id,
|
|
conditions=conditions,
|
|
)
|
|
|
|
|
|
def _matching_changes(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
selectors: _SubjectSelectors,
|
|
) -> list[IdmFunctionAssignmentChange]:
|
|
conditions = []
|
|
if selectors.identity_id:
|
|
conditions.extend(
|
|
(
|
|
IdmFunctionAssignmentChange.candidate_identity_id
|
|
== selectors.identity_id,
|
|
IdmFunctionAssignmentChange.initiator_identity_id
|
|
== selectors.identity_id,
|
|
)
|
|
)
|
|
if selectors.account_id:
|
|
conditions.extend(
|
|
(
|
|
IdmFunctionAssignmentChange.candidate_account_id
|
|
== selectors.account_id,
|
|
IdmFunctionAssignmentChange.initiator_account_id
|
|
== selectors.account_id,
|
|
)
|
|
)
|
|
if reference := selectors.references.get("assignment_change"):
|
|
conditions.append(IdmFunctionAssignmentChange.id == reference)
|
|
return _query_conditions(
|
|
session,
|
|
IdmFunctionAssignmentChange,
|
|
tenant_id=tenant_id,
|
|
conditions=conditions,
|
|
)
|
|
|
|
|
|
def _matching_change_events(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
selectors: _SubjectSelectors,
|
|
change_ids: set[str],
|
|
) -> list[IdmFunctionAssignmentChangeEvent]:
|
|
conditions = []
|
|
if change_ids:
|
|
conditions.append(IdmFunctionAssignmentChangeEvent.change_id.in_(change_ids))
|
|
if selectors.account_id:
|
|
conditions.append(
|
|
IdmFunctionAssignmentChangeEvent.actor_account_id == selectors.account_id
|
|
)
|
|
if selectors.identity_id:
|
|
conditions.append(
|
|
IdmFunctionAssignmentChangeEvent.actor_identity_id == selectors.identity_id
|
|
)
|
|
return _query_conditions(
|
|
session,
|
|
IdmFunctionAssignmentChangeEvent,
|
|
tenant_id=tenant_id,
|
|
conditions=conditions,
|
|
)
|
|
|
|
|
|
def _direct_reference_conflicts(
|
|
selectors: _SubjectSelectors,
|
|
*,
|
|
assignments: Sequence[IdmOrganizationFunctionAssignment],
|
|
relationships: Sequence[IdmIdentityRelationship],
|
|
changes: Sequence[IdmFunctionAssignmentChange],
|
|
) -> bool:
|
|
if not selectors.has_canonical_selector:
|
|
return False
|
|
checks = (
|
|
(
|
|
"assignment",
|
|
assignments,
|
|
lambda row: _assignment_match_fields(row, selectors),
|
|
),
|
|
(
|
|
"relationship",
|
|
relationships,
|
|
lambda row: _relationship_match_fields(row, selectors),
|
|
),
|
|
(
|
|
"assignment_change",
|
|
changes,
|
|
lambda row: _change_match_fields(row, selectors),
|
|
),
|
|
)
|
|
for kind, rows, match in checks:
|
|
reference = selectors.references.get(kind)
|
|
if not reference:
|
|
continue
|
|
row = next((item for item in rows if item.id == reference), None)
|
|
if row is None or not [field for field in match(row) if field != "reference"]:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _assignment_match_fields(
|
|
row: IdmOrganizationFunctionAssignment,
|
|
selectors: _SubjectSelectors,
|
|
) -> list[str]:
|
|
fields = []
|
|
if row.identity_id == selectors.identity_id:
|
|
fields.append("identity_id")
|
|
if selectors.account_id and row.account_id == selectors.account_id:
|
|
fields.append("account_id")
|
|
if selectors.account_id and row.acting_for_account_id == selectors.account_id:
|
|
fields.append("acting_for_account_id")
|
|
if row.id == selectors.references.get("assignment"):
|
|
fields.append("reference")
|
|
return fields
|
|
|
|
|
|
def _relationship_match_fields(
|
|
row: IdmIdentityRelationship,
|
|
selectors: _SubjectSelectors,
|
|
) -> list[str]:
|
|
fields = []
|
|
if row.subject_identity_id == selectors.identity_id:
|
|
fields.append("subject_identity_id")
|
|
if selectors.identity_id and row.related_identity_id == selectors.identity_id:
|
|
fields.append("related_identity_id")
|
|
if row.revoked_by == selectors.account_id:
|
|
fields.append("revoked_by")
|
|
if row.id == selectors.references.get("relationship"):
|
|
fields.append("reference")
|
|
return fields
|
|
|
|
|
|
def _change_match_fields(
|
|
row: IdmFunctionAssignmentChange,
|
|
selectors: _SubjectSelectors,
|
|
) -> list[str]:
|
|
fields = []
|
|
for field, expected in (
|
|
("candidate_identity_id", selectors.identity_id),
|
|
("candidate_account_id", selectors.account_id),
|
|
("initiator_identity_id", selectors.identity_id),
|
|
("initiator_account_id", selectors.account_id),
|
|
):
|
|
if expected and getattr(row, field) == expected:
|
|
fields.append(field)
|
|
if row.id == selectors.references.get("assignment_change"):
|
|
fields.append("reference")
|
|
return fields
|
|
|
|
|
|
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
|
|
groups = {
|
|
"account_id": (
|
|
subject.account_id,
|
|
subject.external_references.get("idm.account"),
|
|
subject.external_references.get("access.account"),
|
|
),
|
|
"identity_id": (
|
|
subject.identity_id,
|
|
subject.external_references.get("idm.identity"),
|
|
subject.external_references.get("identity.id"),
|
|
),
|
|
}
|
|
normalized: dict[str, str | None] = {}
|
|
for key, values in groups.items():
|
|
distinct = {value for item in values if (value := _normalized_id(item))}
|
|
if len(distinct) > 1:
|
|
return None
|
|
normalized[key] = next(iter(distinct), None)
|
|
|
|
aliases = {
|
|
"idm.assignment": "assignment",
|
|
"idm.relationship": "relationship",
|
|
"idm.assignment_change": "assignment_change",
|
|
}
|
|
references = {
|
|
target: value
|
|
for source, target in aliases.items()
|
|
if (value := _normalized_id(subject.external_references.get(source)))
|
|
}
|
|
return _SubjectSelectors(references=references, **normalized)
|
|
|
|
|
|
def _rows_by_ids(
|
|
session: Session,
|
|
model: type,
|
|
*,
|
|
tenant_id: str,
|
|
ids: set[str],
|
|
) -> list[object]:
|
|
if not ids:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(model)
|
|
.filter(model.tenant_id == tenant_id, model.id.in_(ids))
|
|
.order_by(model.id)
|
|
)
|
|
|
|
|
|
def _query_conditions(
|
|
session: Session,
|
|
model: type,
|
|
*,
|
|
tenant_id: str,
|
|
conditions: Sequence[object],
|
|
) -> list[object]:
|
|
if not conditions:
|
|
return []
|
|
return _bounded_rows(
|
|
session.query(model)
|
|
.filter(model.tenant_id == tenant_id, or_(*conditions))
|
|
.order_by(model.id)
|
|
)
|
|
|
|
|
|
def _validate_record(record: DsarRecordRef) -> None:
|
|
if record.provider_id != "idm" or record.module_id != "idm":
|
|
raise ValueError("IDM DSAR received a foreign provider record.")
|
|
|
|
|
|
def _validate_action(action: DsarErasureActionRef) -> None:
|
|
if action.provider_id != "idm" or action.module_id != "idm":
|
|
raise ValueError("IDM DSAR received a foreign provider action.")
|
|
|
|
|
|
def _record(
|
|
resource_type: str,
|
|
resource_id: str,
|
|
category: str,
|
|
title: str,
|
|
data: Mapping[str, object],
|
|
*,
|
|
observed_at: datetime | None,
|
|
immutable: bool = False,
|
|
retention_reason: str | None = None,
|
|
) -> DsarRecordRef:
|
|
return DsarRecordRef(
|
|
provider_id="idm",
|
|
module_id="idm",
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
category=category,
|
|
title=title,
|
|
data=data,
|
|
observed_at=observed_at,
|
|
immutable_evidence=immutable,
|
|
retention_reason=retention_reason,
|
|
source_path="/idm",
|
|
)
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not isinstance(value, Session):
|
|
raise TypeError("IDM DSAR provider requires a SQLAlchemy session.")
|
|
return value
|
|
|
|
|
|
def _bounded_rows(query: object) -> list[object]:
|
|
rows = query.limit(_MAX_RECORDS + 1).all() # type: ignore[attr-defined]
|
|
if len(rows) > _MAX_RECORDS:
|
|
raise ValueError("IDM DSAR match limit exceeded; narrow the subject selectors.")
|
|
return rows
|
|
|
|
|
|
def _bounded_text(value: str | None, limit: int) -> str | None:
|
|
return value[:limit] if value else None
|
|
|
|
|
|
def _normalized_id(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
value = str(value).strip()
|
|
return value or None
|
|
|
|
|
|
def _iso(value: datetime | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=timezone.utc)
|
|
return value.isoformat()
|
|
|
|
|
|
__all__ = ["IDM_DSAR_CAPABILITY", "IdmDsarProvider"]
|