"""CMIS access-point profile primitives. These classes model how a CMIS adapter may expose engine capabilities without turning CMIS into a second domain model. """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any from .actors import ActorType, OperationContext from .assets import AssetRepresentation, KnowledgeAsset, RepresentationKind from .metadata import Sensitivity from .policy import PolicyDecision from .provenance import AssetVersion from .relationships import CoreRelationship, RelationshipTargetKind from .primitives import compact_dict class CMISBinding(str, Enum): BROWSER = "browser" ATOMPUB = "atompub" WEB_SERVICES = "web_services" class CMISCapability(str, Enum): REPOSITORY = "repository" TYPE_DEFINITIONS = "type_definitions" NAVIGATION = "navigation" OBJECT_READ = "object_read" OBJECT_WRITE = "object_write" CONTENT_STREAM_READ = "content_stream_read" CONTENT_STREAM_WRITE = "content_stream_write" VERSIONING = "versioning" DISCOVERY_QUERY = "discovery_query" RELATIONSHIPS = "relationships" ACL = "acl" POLICY = "policy" CHANGE_LOG = "change_log" RENDITIONS = "renditions" RETENTION_HOLD = "retention_hold" BULK_UPDATE = "bulk_update" class CMISAction(str, Enum): GET_REPOSITORY_INFO = "get_repository_info" GET_TYPE_DEFINITION = "get_type_definition" GET_CHILDREN = "get_children" GET_OBJECT = "get_object" GET_CONTENT_STREAM = "get_content_stream" GET_ACL = "get_acl" QUERY = "query" GET_RELATIONSHIPS = "get_relationships" GET_CHANGE_LOG = "get_change_log" CREATE_DOCUMENT = "create_document" UPDATE_PROPERTIES = "update_properties" DELETE_OBJECT = "delete_object" SET_CONTENT_STREAM = "set_content_stream" APPLY_ACL = "apply_acl" APPLY_POLICY = "apply_policy" BULK_UPDATE_PROPERTIES = "bulk_update_properties" class CMISBaseType(str, Enum): DOCUMENT = "cmis:document" FOLDER = "cmis:folder" RELATIONSHIP = "cmis:relationship" POLICY = "cmis:policy" ITEM = "cmis:item" SECONDARY = "cmis:secondary" ACTION_CAPABILITIES: dict[CMISAction, CMISCapability] = { CMISAction.GET_REPOSITORY_INFO: CMISCapability.REPOSITORY, CMISAction.GET_TYPE_DEFINITION: CMISCapability.TYPE_DEFINITIONS, CMISAction.GET_CHILDREN: CMISCapability.NAVIGATION, CMISAction.GET_OBJECT: CMISCapability.OBJECT_READ, CMISAction.GET_CONTENT_STREAM: CMISCapability.CONTENT_STREAM_READ, CMISAction.GET_ACL: CMISCapability.ACL, CMISAction.QUERY: CMISCapability.DISCOVERY_QUERY, CMISAction.GET_RELATIONSHIPS: CMISCapability.RELATIONSHIPS, CMISAction.GET_CHANGE_LOG: CMISCapability.CHANGE_LOG, CMISAction.CREATE_DOCUMENT: CMISCapability.OBJECT_WRITE, CMISAction.UPDATE_PROPERTIES: CMISCapability.OBJECT_WRITE, CMISAction.DELETE_OBJECT: CMISCapability.OBJECT_WRITE, CMISAction.SET_CONTENT_STREAM: CMISCapability.CONTENT_STREAM_WRITE, CMISAction.APPLY_ACL: CMISCapability.ACL, CMISAction.APPLY_POLICY: CMISCapability.POLICY, CMISAction.BULK_UPDATE_PROPERTIES: CMISCapability.BULK_UPDATE, } MUTATION_ACTIONS = { CMISAction.CREATE_DOCUMENT, CMISAction.UPDATE_PROPERTIES, CMISAction.DELETE_OBJECT, CMISAction.SET_CONTENT_STREAM, CMISAction.APPLY_ACL, CMISAction.APPLY_POLICY, CMISAction.BULK_UPDATE_PROPERTIES, } @dataclass(frozen=True) class CMISAccessProfile: name: str binding: CMISBinding | str = CMISBinding.BROWSER capabilities: tuple[CMISCapability | str, ...] = () allow_mutations: bool = False visible_sensitivities: tuple[Sensitivity | str, ...] = (Sensitivity.PUBLIC, Sensitivity.INTERNAL) denied_sensitivities: tuple[Sensitivity | str, ...] = (Sensitivity.CONFIDENTIAL, Sensitivity.RESTRICTED) visible_asset_types: tuple[str, ...] = () visible_topics: tuple[str, ...] = () visible_source_systems: tuple[str, ...] = () denied_metadata: dict[str, tuple[Any, ...]] = field(default_factory=dict) required_actor_types: tuple[ActorType | str, ...] = () metadata: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: object.__setattr__(self, "binding", CMISBinding(self.binding)) object.__setattr__( self, "capabilities", tuple(CMISCapability(capability) for capability in self.capabilities), ) object.__setattr__( self, "visible_sensitivities", tuple(Sensitivity(value) for value in self.visible_sensitivities), ) object.__setattr__( self, "denied_sensitivities", tuple(Sensitivity(value) for value in self.denied_sensitivities), ) object.__setattr__( self, "required_actor_types", tuple(ActorType(value) for value in self.required_actor_types), ) object.__setattr__( self, "denied_metadata", {key: tuple(values) for key, values in self.denied_metadata.items()}, ) @classmethod def readonly_browser(cls) -> "CMISAccessProfile": return cls( name="readonly-browser", capabilities=_read_capabilities(), allow_mutations=False, ) @classmethod def governed_authoring(cls) -> "CMISAccessProfile": return cls( name="governed-authoring", capabilities=_read_capabilities() + ( CMISCapability.OBJECT_WRITE, CMISCapability.CONTENT_STREAM_WRITE, ), allow_mutations=True, ) @classmethod def admin_export(cls) -> "CMISAccessProfile": return cls( name="admin-export", capabilities=_read_capabilities() + ( CMISCapability.RENDITIONS, CMISCapability.RETENTION_HOLD, ), allow_mutations=False, visible_sensitivities=( Sensitivity.PUBLIC, Sensitivity.INTERNAL, Sensitivity.CONFIDENTIAL, Sensitivity.RESTRICTED, ), denied_sensitivities=(), required_actor_types=(ActorType.SERVICE_ACCOUNT,), ) @classmethod def compat_tck(cls) -> "CMISAccessProfile": return cls( name="compat-tck", capabilities=_read_capabilities() + ( CMISCapability.OBJECT_WRITE, CMISCapability.CONTENT_STREAM_WRITE, ), allow_mutations=True, metadata={"compatibility": "selected-opencmis-tck-browser-subset"}, ) def has_capability(self, capability: CMISCapability | str) -> bool: return CMISCapability(capability) in self.capabilities def allows_action(self, action: CMISAction | str) -> bool: cmis_action = CMISAction(action) capability = ACTION_CAPABILITIES[cmis_action] if not self.has_capability(capability): return False if cmis_action in MUTATION_ACTIONS and not self.allow_mutations: return False return True def decide_action( self, action: CMISAction | str, context: OperationContext, *, resource: str = "cmis:repository", ) -> PolicyDecision: cmis_action = CMISAction(action) if not self.allows_actor(context): return PolicyDecision.deny( context.actor.id, cmis_action.value, resource, reason="actor_type_not_allowed_for_cmis_profile", context={"profile": self.name}, ) if self.allows_action(cmis_action): return PolicyDecision.allow( context.actor.id, cmis_action.value, resource, context={"profile": self.name}, ) capability = ACTION_CAPABILITIES[cmis_action] reason = "cmis_mutation_not_allowed" if cmis_action in MUTATION_ACTIONS else "cmis_capability_not_supported" return PolicyDecision.deny( context.actor.id, cmis_action.value, resource, reason=reason, context={"profile": self.name, "capability": capability.value}, ) def allows_actor(self, context: OperationContext) -> bool: return not self.required_actor_types or context.actor.actor_type in self.required_actor_types def exposes_asset(self, asset: KnowledgeAsset, context: OperationContext) -> bool: return self.decide_asset_visibility(asset, context).allowed def decide_asset_visibility( self, asset: KnowledgeAsset, context: OperationContext, ) -> PolicyDecision: resource = f"asset:{asset.id}" if not self.allows_actor(context): return PolicyDecision.deny( context.actor.id, "cmis.expose_asset", resource, reason="actor_type_not_allowed_for_cmis_profile", context={"profile": self.name}, ) classification = asset.classification if classification.sensitivity in self.denied_sensitivities: return PolicyDecision.deny( context.actor.id, "cmis.expose_asset", resource, reason="cmis_sensitivity_denied", context={"profile": self.name, "sensitivity": _enum_value(classification.sensitivity)}, ) if classification.sensitivity not in self.visible_sensitivities: return PolicyDecision.deny( context.actor.id, "cmis.expose_asset", resource, reason="cmis_sensitivity_not_visible", context={"profile": self.name, "sensitivity": _enum_value(classification.sensitivity)}, ) if self.visible_asset_types and classification.asset_type not in self.visible_asset_types: return PolicyDecision.deny( context.actor.id, "cmis.expose_asset", resource, reason="cmis_asset_type_not_visible", context={"profile": self.name, "asset_type": classification.asset_type}, ) if self.visible_topics and not set(classification.topics).intersection(self.visible_topics): return PolicyDecision.deny( context.actor.id, "cmis.expose_asset", resource, reason="cmis_topic_not_visible", context={"profile": self.name, "topics": list(classification.topics)}, ) source_system = asset.metadata.get("source_system") or classification.metadata.get("source_system") if self.visible_source_systems and source_system not in self.visible_source_systems: return PolicyDecision.deny( context.actor.id, "cmis.expose_asset", resource, reason="cmis_source_system_not_visible", context={"profile": self.name, "source_system": source_system}, ) for key, denied_values in self.denied_metadata.items(): value = asset.metadata.get(key, classification.metadata.get(key)) if value in denied_values: return PolicyDecision.deny( context.actor.id, "cmis.expose_asset", resource, reason="cmis_metadata_denied", context={"profile": self.name, "metadata_key": key}, ) return PolicyDecision.allow( context.actor.id, "cmis.expose_asset", resource, context={"profile": self.name}, ) def to_dict(self) -> dict[str, Any]: return compact_dict( { "name": self.name, "binding": self.binding.value, "capabilities": [capability.value for capability in self.capabilities], "allow_mutations": self.allow_mutations, "visible_sensitivities": [item.value for item in self.visible_sensitivities], "denied_sensitivities": [item.value for item in self.denied_sensitivities], "visible_asset_types": list(self.visible_asset_types), "visible_topics": list(self.visible_topics), "visible_source_systems": list(self.visible_source_systems), "denied_metadata": {key: list(values) for key, values in self.denied_metadata.items()}, "required_actor_types": [item.value for item in self.required_actor_types], "metadata": dict(self.metadata), } ) @classmethod def from_dict(cls, data: dict[str, Any]) -> "CMISAccessProfile": return cls( name=data["name"], binding=data.get("binding", CMISBinding.BROWSER.value), capabilities=tuple(data.get("capabilities", [])), allow_mutations=bool(data.get("allow_mutations", False)), visible_sensitivities=tuple(data.get("visible_sensitivities", [])), denied_sensitivities=tuple(data.get("denied_sensitivities", [])), visible_asset_types=tuple(data.get("visible_asset_types", [])), visible_topics=tuple(data.get("visible_topics", [])), visible_source_systems=tuple(data.get("visible_source_systems", [])), denied_metadata=dict(data.get("denied_metadata", {})), required_actor_types=tuple(data.get("required_actor_types", [])), metadata=dict(data.get("metadata", {})), ) @dataclass(frozen=True) class CMISAccessPoint: access_point_id: str repository_id: str profile: CMISAccessProfile base_path: str root_folder_id: str = "cmis-root" enabled: bool = True metadata: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: normalized = "/" + self.base_path.strip("/") object.__setattr__(self, "base_path", normalized) def decide_action( self, action: CMISAction | str, context: OperationContext, *, resource: str | None = None, ) -> PolicyDecision: if not self.enabled: return PolicyDecision.deny( context.actor.id, CMISAction(action).value, resource or f"cmis:{self.repository_id}", reason="cmis_access_point_disabled", context={"access_point_id": self.access_point_id, "profile": self.profile.name}, ) return self.profile.decide_action( action, context, resource=resource or f"cmis:{self.repository_id}", ) def exposes_asset(self, asset: KnowledgeAsset, context: OperationContext) -> bool: return self.enabled and self.profile.exposes_asset(asset, context) def to_dict(self) -> dict[str, Any]: return compact_dict( { "access_point_id": self.access_point_id, "repository_id": self.repository_id, "profile": self.profile.to_dict(), "base_path": self.base_path, "root_folder_id": self.root_folder_id, "enabled": self.enabled, "metadata": dict(self.metadata), } ) @classmethod def from_dict(cls, data: dict[str, Any]) -> "CMISAccessPoint": return cls( access_point_id=data["access_point_id"], repository_id=data["repository_id"], profile=CMISAccessProfile.from_dict(data["profile"]), base_path=data["base_path"], root_folder_id=data.get("root_folder_id", "cmis-root"), enabled=bool(data.get("enabled", True)), metadata=dict(data.get("metadata", {})), ) @dataclass(frozen=True) class CMISObjectProjection: object_id: str base_type_id: CMISBaseType type_id: str name: str properties: dict[str, Any] allowable_actions: tuple[CMISAction, ...] = () path: str | None = None content_stream: dict[str, Any] | None = None version: dict[str, Any] | None = None relationships: tuple[str, ...] = () acl: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: return compact_dict( { "object_id": self.object_id, "base_type_id": self.base_type_id.value, "type_id": self.type_id, "name": self.name, "path": self.path, "properties": dict(self.properties), "allowable_actions": [action.value for action in self.allowable_actions], "content_stream": dict(self.content_stream or {}), "version": dict(self.version or {}), "relationships": list(self.relationships), "acl": dict(self.acl or {}), } ) class CMISDomainMapper: """Project engine domain objects into CMIS-shaped envelopes.""" def __init__(self, access_point: CMISAccessPoint) -> None: self.access_point = access_point def repository_info(self) -> dict[str, Any]: profile = self.access_point.profile return { "repository_id": self.access_point.repository_id, "repository_name": self.access_point.metadata.get("repository_name", self.access_point.repository_id), "cmis_version_supported": "1.1", "root_folder_id": self.access_point.root_folder_id, "principal_anonymous": "anonymous", "principal_anyone": "anyone", "product_name": "kontextual-engine", "binding": profile.binding.value, "capabilities": self.capability_flags(), "profile": profile.name, } def capability_flags(self) -> dict[str, Any]: profile = self.access_point.profile return { "capability_content_stream_updatability": ( "anytime" if profile.has_capability(CMISCapability.CONTENT_STREAM_WRITE) else "none" ), "capability_changes": "objectidsonly" if profile.has_capability(CMISCapability.CHANGE_LOG) else "none", "capability_renditions": "read" if profile.has_capability(CMISCapability.RENDITIONS) else "none", "capability_get_descendants": profile.has_capability(CMISCapability.NAVIGATION), "capability_get_folder_tree": profile.has_capability(CMISCapability.NAVIGATION), "capability_multifiling": False, "capability_unfiling": False, "capability_version_specific_filing": False, "capability_pwc_searchable": False, "capability_pwc_updatable": False, "capability_all_versions_searchable": profile.has_capability(CMISCapability.VERSIONING), "capability_query": "metadataonly" if profile.has_capability(CMISCapability.DISCOVERY_QUERY) else "none", "capability_join": "none", "capability_acl": "discover" if profile.has_capability(CMISCapability.ACL) else "none", } def type_definitions(self) -> list[dict[str, Any]]: can_write = self.access_point.profile.allow_mutations return [ _type_definition(CMISBaseType.DOCUMENT, "kontextual:document", "Kontextual Document", can_write), _type_definition(CMISBaseType.FOLDER, "kontextual:folder", "Kontextual Folder", False), _type_definition( CMISBaseType.RELATIONSHIP, "kontextual:relationship", "Kontextual Relationship", can_write, ), _type_definition(CMISBaseType.POLICY, "kontextual:policy", "Kontextual Policy", False), _type_definition(CMISBaseType.ITEM, "kontextual:item", "Kontextual Item", False), _type_definition(CMISBaseType.SECONDARY, "kontextual:secondary", "Kontextual Secondary", False), ] def map_asset( self, asset: KnowledgeAsset, context: OperationContext, *, representations: list[AssetRepresentation] | tuple[AssetRepresentation, ...] = (), versions: list[AssetVersion] | tuple[AssetVersion, ...] = (), relationship_ids: list[str] | tuple[str, ...] = (), metadata_records: list[Any] | tuple[Any, ...] = (), ) -> CMISObjectProjection | None: if not self.access_point.exposes_asset(asset, context): return None current_version = _current_version(asset, versions) content_stream = self.map_content_stream(asset, representations) return CMISObjectProjection( object_id=self.asset_object_id(asset.id), base_type_id=CMISBaseType.DOCUMENT, type_id=f"kontextual:{asset.classification.asset_type}", name=asset.title, path=self.asset_path(asset), properties=self.asset_properties(asset, metadata_records=metadata_records), allowable_actions=self.allowable_actions(context, has_content_stream=content_stream is not None), content_stream=content_stream, version=self.version_properties(asset, current_version, versions), relationships=tuple(relationship_ids), acl=self.acl_for_asset(asset, context), ) def map_relationship( self, relationship: CoreRelationship, context: OperationContext, ) -> CMISObjectProjection | None: decision = self.access_point.decide_action( CMISAction.GET_RELATIONSHIPS, context, resource=f"relationship:{relationship.relationship_id}", ) if not decision.allowed: return None source_id = self.asset_object_id(relationship.source_id) target_id = ( self.asset_object_id(relationship.target_id) if relationship.target_kind == RelationshipTargetKind.ASSET else f"cmis:entity:{relationship.target_id}" ) return CMISObjectProjection( object_id=f"cmis:relationship:{relationship.relationship_id}", base_type_id=CMISBaseType.RELATIONSHIP, type_id="kontextual:relationship", name=relationship.predicate, properties={ "cmis:objectId": f"cmis:relationship:{relationship.relationship_id}", "cmis:name": relationship.predicate, "cmis:baseTypeId": CMISBaseType.RELATIONSHIP.value, "cmis:objectTypeId": "kontextual:relationship", "cmis:sourceId": source_id, "cmis:targetId": target_id, "kontextual:predicate": relationship.predicate, "kontextual:confidence": relationship.confidence, "kontextual:targetKind": relationship.target_kind.value, }, allowable_actions=(CMISAction.GET_OBJECT, CMISAction.GET_RELATIONSHIPS), ) def acl_for_asset(self, asset: KnowledgeAsset, context: OperationContext) -> dict[str, Any] | None: visibility = self.access_point.profile.decide_asset_visibility(asset, context) if not visibility.allowed: return None permissions = ["cmis:read"] if self.access_point.profile.allow_mutations: permissions.extend(["cmis:write", "cmis:delete"]) entries = [ { "principal_id": context.actor.id, "permissions": permissions, "direct": True, } ] if asset.classification.sensitivity == Sensitivity.PUBLIC: entries.append( { "principal_id": "anyone", "permissions": ["cmis:read"], "direct": False, } ) return { "object_id": self.asset_object_id(asset.id), "is_exact": True, "aces": entries, "derived_from": "kontextual-profile-policy", "profile": self.access_point.profile.name, } def asset_object_id(self, asset_id: str) -> str: return f"cmis:asset:{asset_id}" def asset_path(self, asset: KnowledgeAsset) -> str: explicit = asset.metadata.get("cmis_path") if explicit: return _normalize_path(str(explicit)) if asset.source_refs: source_ref = asset.source_refs[0] source_root = _safe_path_segment(source_ref.source_system) if source_ref.path: return _normalize_path(f"/sources/{source_root}/{source_ref.path}") if source_ref.external_id: return _normalize_path(f"/sources/{source_root}/{source_ref.external_id}") topics = asset.classification.topics if topics: return _normalize_path(f"/topics/{topics[0]}/{asset.id}") return _normalize_path(f"/assets/{asset.classification.asset_type}/{asset.id}") def asset_properties( self, asset: KnowledgeAsset, *, metadata_records: list[Any] | tuple[Any, ...] = (), ) -> dict[str, Any]: classification = asset.classification properties = { "cmis:objectId": self.asset_object_id(asset.id), "cmis:name": asset.title, "cmis:baseTypeId": CMISBaseType.DOCUMENT.value, "cmis:objectTypeId": f"kontextual:{classification.asset_type}", "cmis:createdBy": asset.metadata.get("created_by"), "cmis:lastModifiedBy": asset.metadata.get("updated_by"), "cmis:creationDate": asset.created_at, "cmis:lastModificationDate": asset.updated_at, "cmis:changeToken": asset.current_version_id, "kontextual:assetId": asset.id, "kontextual:assetType": classification.asset_type, "kontextual:sensitivity": _enum_value(classification.sensitivity), "kontextual:lifecycle": _enum_value(asset.lifecycle), "kontextual:owner": classification.owner, "kontextual:topics": list(classification.topics), "kontextual:reviewState": classification.review_state, } for record in metadata_records: key = getattr(record, "key", None) if key: properties[f"kontextual:metadata:{key}"] = getattr(record, "value", None) return compact_dict(properties) def map_content_stream( self, asset: KnowledgeAsset, representations: list[AssetRepresentation] | tuple[AssetRepresentation, ...], ) -> dict[str, Any] | None: representation = _preferred_representation(representations) if representation is None: return None return compact_dict( { "stream_id": representation.representation_id, "file_name": asset.metadata.get("file_name", asset.title), "mime_type": representation.media_type, "length": representation.size_bytes, "digest": representation.digest, "storage_ref": representation.storage_ref, "kind": representation.kind.value, } ) def version_properties( self, asset: KnowledgeAsset, current_version: AssetVersion | None, versions: list[AssetVersion] | tuple[AssetVersion, ...], ) -> dict[str, Any]: if current_version is None: return { "cmis:isLatestVersion": True, "cmis:isMajorVersion": True, "cmis:isLatestMajorVersion": True, "cmis:versionSeriesId": f"cmis:version-series:{asset.id}", "cmis:versionLabel": "1", } latest_sequence = max((version.sequence for version in versions), default=current_version.sequence) return { "cmis:isLatestVersion": current_version.sequence == latest_sequence, "cmis:isMajorVersion": True, "cmis:isLatestMajorVersion": current_version.sequence == latest_sequence, "cmis:versionSeriesId": f"cmis:version-series:{asset.id}", "cmis:versionLabel": str(current_version.sequence), "kontextual:versionId": current_version.version_id, "kontextual:versionChangeType": current_version.change_type.value, } def allowable_actions( self, context: OperationContext, *, has_content_stream: bool, ) -> tuple[CMISAction, ...]: candidates = [ CMISAction.GET_OBJECT, CMISAction.GET_CONTENT_STREAM, CMISAction.GET_ACL, CMISAction.GET_RELATIONSHIPS, CMISAction.UPDATE_PROPERTIES, CMISAction.DELETE_OBJECT, CMISAction.SET_CONTENT_STREAM, ] actions: list[CMISAction] = [] for action in candidates: if action == CMISAction.GET_CONTENT_STREAM and not has_content_stream: continue if self.access_point.decide_action(action, context).allowed: actions.append(action) return tuple(actions) def _read_capabilities() -> tuple[CMISCapability, ...]: return ( CMISCapability.REPOSITORY, CMISCapability.TYPE_DEFINITIONS, CMISCapability.NAVIGATION, CMISCapability.OBJECT_READ, CMISCapability.CONTENT_STREAM_READ, CMISCapability.VERSIONING, CMISCapability.DISCOVERY_QUERY, CMISCapability.RELATIONSHIPS, CMISCapability.ACL, CMISCapability.CHANGE_LOG, ) def _enum_value(value: Any) -> Any: return getattr(value, "value", value) def _type_definition( base_type_id: CMISBaseType, type_id: str, display_name: str, can_write: bool, ) -> dict[str, Any]: return { "id": type_id, "local_name": type_id.split(":", 1)[-1], "display_name": display_name, "base_type_id": base_type_id.value, "queryable": True, "controllable_acl": base_type_id in {CMISBaseType.DOCUMENT, CMISBaseType.FOLDER}, "controllable_policy": False, "creatable": can_write and base_type_id == CMISBaseType.DOCUMENT, "fileable": base_type_id == CMISBaseType.DOCUMENT, "fulltext_indexed": False, "included_in_supertype_query": True, "versionable": base_type_id == CMISBaseType.DOCUMENT, "property_definitions": _property_definitions(base_type_id), } def _property_definitions(base_type_id: CMISBaseType) -> dict[str, dict[str, Any]]: definitions = { "cmis:objectId": {"property_type": "id", "cardinality": "single", "required": True}, "cmis:name": {"property_type": "string", "cardinality": "single", "required": True}, "cmis:baseTypeId": {"property_type": "id", "cardinality": "single", "required": True}, "cmis:objectTypeId": {"property_type": "id", "cardinality": "single", "required": True}, "kontextual:sensitivity": {"property_type": "string", "cardinality": "single", "required": False}, "kontextual:lifecycle": {"property_type": "string", "cardinality": "single", "required": False}, } if base_type_id == CMISBaseType.RELATIONSHIP: definitions["cmis:sourceId"] = {"property_type": "id", "cardinality": "single", "required": True} definitions["cmis:targetId"] = {"property_type": "id", "cardinality": "single", "required": True} return definitions def _current_version( asset: KnowledgeAsset, versions: list[AssetVersion] | tuple[AssetVersion, ...], ) -> AssetVersion | None: if asset.current_version_id: for version in versions: if version.version_id == asset.current_version_id: return version if versions: return sorted(versions, key=lambda version: version.sequence)[-1] return None def _preferred_representation( representations: list[AssetRepresentation] | tuple[AssetRepresentation, ...], ) -> AssetRepresentation | None: if not representations: return None priority = { RepresentationKind.SOURCE: 0, RepresentationKind.NORMALIZED: 1, RepresentationKind.DERIVED: 2, } return sorted(representations, key=lambda item: priority.get(item.kind, 99))[0] def _normalize_path(path: str) -> str: parts = [_safe_path_segment(part) for part in path.replace("\\", "/").split("/") if part] return "/" + "/".join(parts) def _safe_path_segment(value: str) -> str: return str(value).strip().strip("/") or "_"