Source code for src.share_kernel.domain.interfaces

"""Abstract interfaces shared across bounded contexts.

These ABCs define the contracts that concrete implementations must fulfill.
The domain layer depends only on these interfaces, never on concrete implementations.
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any




class TenantIsolationBackend(ABC):
    """Interface for tenant isolation strategies (schema, RLS, shared_fk)."""



    @abstractmethod
    def activate_tenant(self, tenant: Any, connection: Any) -> None:
        """Configure the DB connection for the active tenant."""
        ...




    @abstractmethod
    def deactivate_tenant(self, connection: Any) -> None:
        """Restore the DB connection to its default state."""
        ...




    @abstractmethod
    def create_tenant_storage(self, tenant: Any) -> None:
        """Provision storage for a new tenant (schema, RLS policies, etc.)."""
        ...




    @abstractmethod
    def destroy_tenant_storage(self, tenant: Any) -> None:
        """Destroy a tenant's storage."""
        ...




    @abstractmethod
    def migrate_tenant(self, tenant: Any) -> None:
        """Apply migrations for a specific tenant."""
        ...






class ConditionEngine(ABC):
    """Interface for pluggable ABAC condition evaluation engines."""



    @abstractmethod
    def evaluate(self, conditions: dict[str, Any], attributes: dict[str, Any]) -> bool:
        """Evaluate conditions against the provided attributes.

        Returns True if the conditions match the attributes, False otherwise.
        """
        ...






class AttributeProvider(ABC):
    """Interface for custom attribute providers used by the PIP."""



    @abstractmethod
    def get_attributes(self, context: dict[str, Any]) -> dict[str, Any]:
        """Return a dict of additional attributes for ABAC evaluation."""
        ...






class NotificationChannel(ABC):
    """Interface for approval notification channels."""



    @abstractmethod
    def send_notification(
        self,
        recipients: list[str],
        subject: str,
        message: str,
        context: dict[str, Any] | None = None,
    ) -> None:
        """Send a notification to the given recipients."""
        ...