coulomb/markitect-main
2
0
Fork 0
Code Issues 60 Pull Requests Actions Projects 12 Releases 1 Wiki Activity

feat(spaces): implement Phase 7 Composability

Browse Source 
Implements space composition and inheritance features:
- SpaceReference model for space-to-space references (includes, extends, links_to, composed_of)
- Variable inheritance through parent chain with local override
- Config inheritance with source tracking
- Access control models (SpacePermission, SpaceRole, AccessLevel)
- InheritanceResolver for walking parent chains
- AccessControlService for permission management
- ComposableSpaceService integrating all composability features
- Circular reference detection for EXTENDS references
- SQLite repositories for references and permissions
- 57 comprehensive unit tests

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2026-02-08 17:41:40 +01:00
parent 7de57a389d
commit 727ce4d3c5
6 changed files with 2710 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

825
markitect/spaces/composability/service.py Normal file
View File

@@ -0,0 +1,825 @@
"""
Service layer for space composability.
Provides inheritance resolution, reference management, and access control.
"""
from typing import List, Optional, Dict, Any, Set
from dataclasses import dataclass
from ..models import InformationSpace, SpaceVariable, SpaceConfig
from ..repositories.interfaces import ISpaceRepository, IVariableRepository
from .models import (
SpaceReference,
SpaceReferenceType,
SpacePermission,
AccessLevel,
SpaceRole,
SpaceAccess,
InheritedVariable,
InheritedConfig,
)
from .repository import ISpaceReferenceRepository, IAccessControlRepository
class CircularReferenceError(Exception):
"""Raised when a circular reference is detected."""
pass
class InheritanceResolver:
"""
Resolves inherited values through the space hierarchy.
Handles variable inheritance, config inheritance, and reference traversal.
"""
def __init__(
self,
space_repo: ISpaceRepository,
variable_repo: IVariableRepository,
reference_repo: Optional[ISpaceReferenceRepository] = None,
max_depth: int = 10,
):
"""
Initialize the resolver.
Args:
space_repo: Repository for space lookups
variable_repo: Repository for variable lookups
reference_repo: Optional repository for reference lookups
max_depth: Maximum inheritance depth to prevent infinite loops
"""
self._space_repo = space_repo
self._variable_repo = variable_repo
self._reference_repo = reference_repo
self._max_depth = max_depth
def get_parent_chain(
self, space_id: str, visited: Optional[Set[str]] = None
) -> List[InformationSpace]:
"""
Get the chain of parent spaces.
Args:
space_id: Starting space ID
visited: Set of already-visited space IDs (for cycle detection)
Returns:
List of parent spaces, from immediate parent to root
Raises:
CircularReferenceError: If a cycle is detected
"""
if visited is None:
visited = set()
if space_id in visited:
raise CircularReferenceError(
f"Circular reference detected: space {space_id} already in chain"
)
visited.add(space_id)
space = self._space_repo.get_by_id(space_id)
if not space or not space.parent_space_id:
return []
if len(visited) > self._max_depth:
raise CircularReferenceError(
f"Maximum inheritance depth ({self._max_depth}) exceeded"
)
parent = self._space_repo.get_by_id(space.parent_space_id)
if not parent:
return []
return [parent] + self.get_parent_chain(parent.id, visited)
def resolve_variable(
self, space_id: str, name: str
) -> Optional[InheritedVariable]:
"""
Resolve a variable, checking parent spaces if not found locally.
Args:
space_id: The space to start resolution from
name: Variable name to resolve
Returns:
InheritedVariable with value and source, or None if not found
"""
# Check local space first
local_var = self._variable_repo.get_variable(space_id, name)
if local_var:
return InheritedVariable(
name=name,
value=local_var.value,
source_space_id=space_id,
inheritance_depth=0,
scope=local_var.scope,
)
# Walk up parent chain
try:
parents = self.get_parent_chain(space_id)
except CircularReferenceError:
return None
for depth, parent in enumerate(parents, start=1):
parent_var = self._variable_repo.get_variable(parent.id, name)
if parent_var:
return InheritedVariable(
name=name,
value=parent_var.value,
source_space_id=parent.id,
inheritance_depth=depth,
scope=parent_var.scope,
)
return None
def resolve_all_variables(
self, space_id: str, scope: Optional[str] = None
) -> Dict[str, InheritedVariable]:
"""
Resolve all variables for a space, including inherited ones.
Variables defined locally override inherited ones with the same name.
Args:
space_id: The space ID
scope: Optional scope filter
Returns:
Dictionary of variable name to InheritedVariable
"""
result: Dict[str, InheritedVariable] = {}
# Start from the furthest ancestor and work down
try:
parents = self.get_parent_chain(space_id)
except CircularReferenceError:
parents = []
# Build list from root to current: [grandparent, parent, current]
# parents is [parent, grandparent] so reverse it
current_space = self._space_repo.get_by_id(space_id)
all_spaces = list(reversed(parents))
if current_space:
all_spaces.append(current_space)
# Process from root to current so local values override inherited ones
# Index 0 is root (highest depth), last index is current (depth 0)
total = len(all_spaces)
for i, space in enumerate(all_spaces):
# depth_from_current: current=0, parent=1, grandparent=2, etc.
depth_from_current = total - 1 - i
variables = self._variable_repo.list_variables(space.id, scope)
for var in variables:
result[var.name] = InheritedVariable(
name=var.name,
value=var.value,
source_space_id=space.id,
inheritance_depth=depth_from_current,
scope=var.scope,
)
return result
def resolve_config(self, space_id: str) -> InheritedConfig:
"""
Resolve the effective configuration for a space.
Config values are inherited from parents unless overridden locally.
Args:
space_id: The space ID
Returns:
InheritedConfig with effective values and sources
"""
space = self._space_repo.get_by_id(space_id)
if not space:
return InheritedConfig()
# Start with defaults
result = InheritedConfig(
default_variant="hierarchical",
enable_caching=True,
theme=None,
history_enabled=False,
variable_scope="space",
source_spaces={},
)
# Get parent chain
try:
parents = self.get_parent_chain(space_id)
except CircularReferenceError:
parents = []
# Process from root to current (reversed order)
all_spaces = list(reversed(parents)) + [space]
for s in all_spaces:
config = s.config
if config.default_variant:
result.default_variant = config.default_variant
result.source_spaces["default_variant"] = s.id
if config.enable_caching is not None:
result.enable_caching = config.enable_caching
result.source_spaces["enable_caching"] = s.id
if config.theme:
result.theme = config.theme
result.source_spaces["theme"] = s.id
if config.history_enabled is not None:
result.history_enabled = config.history_enabled
result.source_spaces["history_enabled"] = s.id
if config.variable_scope:
result.variable_scope = config.variable_scope
result.source_spaces["variable_scope"] = s.id
return result
class AccessControlService:
"""
Service for managing space access control.
Handles permission grants, checks, and inheritance.
"""
def __init__(
self,
permission_repo: IAccessControlRepository,
space_repo: ISpaceRepository,
inheritance_resolver: Optional[InheritanceResolver] = None,
):
"""
Initialize the service.
Args:
permission_repo: Repository for permission storage
space_repo: Repository for space lookups
inheritance_resolver: Optional resolver for inherited permissions
"""
self._permission_repo = permission_repo
self._space_repo = space_repo
self._inheritance_resolver = inheritance_resolver
def grant_permission(
self,
space_id: str,
principal_type: str,
principal_id: str,
access_level: AccessLevel,
role: Optional[SpaceRole] = None,
granted_by: Optional[str] = None,
) -> SpacePermission:
"""
Grant a permission to a principal.
Args:
space_id: The space ID
principal_type: Type of principal (user, group, role)
principal_id: Principal identifier
access_level: Access level to grant
role: Optional role assignment
granted_by: Who is granting the permission
Returns:
The created permission
"""
permission = SpacePermission(
space_id=space_id,
principal_type=principal_type,
principal_id=principal_id,
access_level=access_level,
role=role,
granted_by=granted_by,
)
return self._permission_repo.grant_permission(permission)
def revoke_permission(
self, space_id: str, principal_type: str, principal_id: str
) -> bool:
"""
Revoke a permission.
Args:
space_id: The space ID
principal_type: Type of principal
principal_id: Principal identifier
Returns:
True if revoked, False if not found
"""
return self._permission_repo.revoke_permission(
space_id, principal_type, principal_id
)
def check_access(
self,
space_id: str,
principal_type: str,
principal_id: str,
required_level: AccessLevel = AccessLevel.READ,
) -> bool:
"""
Check if a principal has the required access level.
Args:
space_id: The space ID
principal_type: Type of principal
principal_id: Principal identifier
required_level: Required access level
Returns:
True if access is granted
"""
access = self.get_effective_access(space_id, principal_type, principal_id)
if access.effective_level == AccessLevel.NONE:
return False
# Compare using level order, not string values
level_order = [AccessLevel.NONE, AccessLevel.READ, AccessLevel.WRITE, AccessLevel.ADMIN]
return level_order.index(access.effective_level) >= level_order.index(required_level)
def get_effective_access(
self, space_id: str, principal_type: str, principal_id: str
) -> SpaceAccess:
"""
Get the effective access for a principal on a space.
Considers direct permissions and inherited permissions from parent spaces.
Args:
space_id: The space ID
principal_type: Type of principal
principal_id: Principal identifier
Returns:
SpaceAccess with computed effective access
"""
result = SpaceAccess(
space_id=space_id,
principal_id=principal_id,
effective_level=AccessLevel.NONE,
roles=[],
inherited_from=[],
permissions=[],
)
# Check direct permission
direct = self._permission_repo.get_permission(
space_id, principal_type, principal_id
)
if direct and not direct.is_expired():
result.permissions.append(direct)
result.effective_level = direct.access_level
if direct.role:
result.roles.append(direct.role)
# Check inherited permissions if we have a resolver
if self._inheritance_resolver:
try:
parents = self._inheritance_resolver.get_parent_chain(space_id)
except CircularReferenceError:
parents = []
for parent in parents:
parent_perm = self._permission_repo.get_permission(
parent.id, principal_type, principal_id
)
if parent_perm and not parent_perm.is_expired():
result.permissions.append(parent_perm)
result.inherited_from.append(parent.id)
# Take the highest access level
level_order = [
AccessLevel.NONE,
AccessLevel.READ,
AccessLevel.WRITE,
AccessLevel.ADMIN,
]
if level_order.index(parent_perm.access_level) > level_order.index(
result.effective_level
):
result.effective_level = parent_perm.access_level
if parent_perm.role and parent_perm.role not in result.roles:
result.roles.append(parent_perm.role)
return result
def get_space_permissions(self, space_id: str) -> List[SpacePermission]:
"""
Get all permissions for a space.
Args:
space_id: The space ID
Returns:
List of permissions
"""
return self._permission_repo.get_permissions_for_space(space_id)
def get_principal_permissions(
self, principal_type: str, principal_id: str
) -> List[SpacePermission]:
"""
Get all permissions for a principal.
Args:
principal_type: Type of principal
principal_id: Principal identifier
Returns:
List of permissions
"""
return self._permission_repo.get_permissions_for_principal(
principal_type, principal_id
)
class ComposableSpaceService:
"""
Extended space service with composability features.
Provides space references, inheritance resolution, and access control
on top of the base SpaceService functionality.
"""
def __init__(
self,
space_repo: ISpaceRepository,
variable_repo: IVariableRepository,
reference_repo: ISpaceReferenceRepository,
permission_repo: IAccessControlRepository,
max_inheritance_depth: int = 10,
):
"""
Initialize the service.
Args:
space_repo: Repository for spaces
variable_repo: Repository for variables
reference_repo: Repository for space references
permission_repo: Repository for permissions
max_inheritance_depth: Maximum inheritance chain depth
"""
self._space_repo = space_repo
self._variable_repo = variable_repo
self._reference_repo = reference_repo
self._permission_repo = permission_repo
self._inheritance_resolver = InheritanceResolver(
space_repo=space_repo,
variable_repo=variable_repo,
reference_repo=reference_repo,
max_depth=max_inheritance_depth,
)
self._access_service = AccessControlService(
permission_repo=permission_repo,
space_repo=space_repo,
inheritance_resolver=self._inheritance_resolver,
)
# =========================================================================
# Space References
# =========================================================================
def add_reference(
self,
source_space_id: str,
target_space_id: str,
reference_type: SpaceReferenceType = SpaceReferenceType.LINKS_TO,
alias: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> SpaceReference:
"""
Add a reference from one space to another.
Args:
source_space_id: The space making the reference
target_space_id: The space being referenced
reference_type: Type of reference
alias: Optional alias for the target space
metadata: Optional metadata
Returns:
The created reference
Raises:
ValueError: If spaces don't exist or reference would create a cycle
CircularReferenceError: If reference would create a circular dependency
"""
# Validate spaces exist
if not self._space_repo.exists(source_space_id):
raise ValueError(f"Source space '{source_space_id}' not found")
if not self._space_repo.exists(target_space_id):
raise ValueError(f"Target space '{target_space_id}' not found")
# Check for self-reference
if source_space_id == target_space_id:
raise ValueError("Cannot reference self")
# Check for circular reference (for EXTENDS type)
if reference_type == SpaceReferenceType.EXTENDS:
self._check_circular_reference(source_space_id, target_space_id)
reference = SpaceReference(
source_space_id=source_space_id,
target_space_id=target_space_id,
reference_type=reference_type,
alias=alias,
metadata=metadata or {},
)
return self._reference_repo.add_reference(reference)
def _check_circular_reference(
self, source_id: str, target_id: str, visited: Optional[Set[str]] = None
) -> None:
"""Check if adding a reference would create a cycle."""
if visited is None:
visited = {source_id}
if target_id in visited:
raise CircularReferenceError(
f"Adding reference would create cycle: {target_id} already in chain"
)
visited.add(target_id)
# Check existing references from target
refs = self._reference_repo.get_references_from(
target_id, SpaceReferenceType.EXTENDS
)
for ref in refs:
self._check_circular_reference(source_id, ref.target_space_id, visited)
# Also check parent_space_id chain
target_space = self._space_repo.get_by_id(target_id)
if target_space and target_space.parent_space_id:
if target_space.parent_space_id in visited:
raise CircularReferenceError(
f"Adding reference would create cycle through parent: "
f"{target_space.parent_space_id}"
)
self._check_circular_reference(
source_id, target_space.parent_space_id, visited
)
def remove_reference(self, reference_id: str) -> bool:
"""
Remove a space reference.
Args:
reference_id: The reference ID
Returns:
True if removed, False if not found
"""
return self._reference_repo.remove_reference(reference_id)
def get_references_from(
self,
space_id: str,
reference_type: Optional[SpaceReferenceType] = None,
) -> List[SpaceReference]:
"""
Get all references from a space.
Args:
space_id: The source space ID
reference_type: Optional filter by type
Returns:
List of references
"""
return self._reference_repo.get_references_from(space_id, reference_type)
def get_references_to(
self,
space_id: str,
reference_type: Optional[SpaceReferenceType] = None,
) -> List[SpaceReference]:
"""
Get all references to a space.
Args:
space_id: The target space ID
reference_type: Optional filter by type
Returns:
List of references
"""
return self._reference_repo.get_references_to(space_id, reference_type)
def get_composed_spaces(self, space_id: str) -> List[InformationSpace]:
"""
Get all spaces that compose this space.
Includes parent, extended spaces, and included spaces.
Args:
space_id: The space ID
Returns:
List of composed spaces
"""
result: List[InformationSpace] = []
seen: Set[str] = {space_id}
space = self._space_repo.get_by_id(space_id)
if not space:
return []
# Add parent
if space.parent_space_id:
parent = self._space_repo.get_by_id(space.parent_space_id)
if parent and parent.id not in seen:
result.append(parent)
seen.add(parent.id)
# Add extended spaces
extends_refs = self._reference_repo.get_references_from(
space_id, SpaceReferenceType.EXTENDS
)
for ref in extends_refs:
if ref.target_space_id not in seen:
target = self._space_repo.get_by_id(ref.target_space_id)
if target:
result.append(target)
seen.add(ref.target_space_id)
# Add included spaces
includes_refs = self._reference_repo.get_references_from(
space_id, SpaceReferenceType.INCLUDES
)
for ref in includes_refs:
if ref.target_space_id not in seen:
target = self._space_repo.get_by_id(ref.target_space_id)
if target:
result.append(target)
seen.add(ref.target_space_id)
return result
# =========================================================================
# Inheritance Resolution
# =========================================================================
def resolve_variable(
self, space_id: str, name: str
) -> Optional[InheritedVariable]:
"""
Resolve a variable with inheritance.
Args:
space_id: The space ID
name: Variable name
Returns:
InheritedVariable or None if not found
"""
return self._inheritance_resolver.resolve_variable(space_id, name)
def resolve_all_variables(
self, space_id: str, scope: Optional[str] = None
) -> Dict[str, InheritedVariable]:
"""
Resolve all variables including inherited ones.
Args:
space_id: The space ID
scope: Optional scope filter
Returns:
Dictionary of variable name to InheritedVariable
"""
return self._inheritance_resolver.resolve_all_variables(space_id, scope)
def resolve_config(self, space_id: str) -> InheritedConfig:
"""
Resolve effective configuration with inheritance.
Args:
space_id: The space ID
Returns:
InheritedConfig with effective values
"""
return self._inheritance_resolver.resolve_config(space_id)
def get_parent_chain(self, space_id: str) -> List[InformationSpace]:
"""
Get the parent space chain.
Args:
space_id: The space ID
Returns:
List of parent spaces from immediate parent to root
"""
return self._inheritance_resolver.get_parent_chain(space_id)
# =========================================================================
# Access Control
# =========================================================================
def grant_access(
self,
space_id: str,
principal_type: str,
principal_id: str,
access_level: AccessLevel,
role: Optional[SpaceRole] = None,
granted_by: Optional[str] = None,
) -> SpacePermission:
"""
Grant access to a space.
Args:
space_id: The space ID
principal_type: Type of principal (user, group)
principal_id: Principal identifier
access_level: Access level to grant
role: Optional role
granted_by: Who is granting access
Returns:
The created permission
"""
return self._access_service.grant_permission(
space_id, principal_type, principal_id, access_level, role, granted_by
)
def revoke_access(
self, space_id: str, principal_type: str, principal_id: str
) -> bool:
"""
Revoke access to a space.
Args:
space_id: The space ID
principal_type: Type of principal
principal_id: Principal identifier
Returns:
True if revoked
"""
return self._access_service.revoke_permission(
space_id, principal_type, principal_id
)
def check_access(
self,
space_id: str,
principal_type: str,
principal_id: str,
required_level: AccessLevel = AccessLevel.READ,
) -> bool:
"""
Check if a principal has required access.
Args:
space_id: The space ID
principal_type: Type of principal
principal_id: Principal identifier
required_level: Required access level
Returns:
True if access is granted
"""
return self._access_service.check_access(
space_id, principal_type, principal_id, required_level
)
def get_effective_access(
self, space_id: str, principal_type: str, principal_id: str
) -> SpaceAccess:
"""
Get effective access including inheritance.
Args:
space_id: The space ID
principal_type: Type of principal
principal_id: Principal identifier
Returns:
SpaceAccess with computed access
"""
return self._access_service.get_effective_access(
space_id, principal_type, principal_id
)
def get_space_permissions(self, space_id: str) -> List[SpacePermission]:
"""Get all permissions for a space."""
return self._access_service.get_space_permissions(space_id)