Source code for src.module.authorization.infrastructure.condition_engines

"""Pluggable ABAC condition engines.

The active engine is selected via MULTITENANT['ABAC_CONDITION_ENGINE'].
"""

from __future__ import annotations

from src.share_kernel.domain.interfaces import ConditionEngine

_engine_instance: ConditionEngine | None = None




def get_condition_engine() -> ConditionEngine:
    """Return the configured condition engine singleton."""
    global _engine_instance
    if _engine_instance is not None:
        return _engine_instance

    from src.share_kernel.settings import get_setting

    engine_type = str(get_setting("ABAC_CONDITION_ENGINE"))

    if engine_type == "json":
        from src.module.authorization.infrastructure.condition_engines.json_engine import (
            JsonConditionEngine,
        )

        _engine_instance = JsonConditionEngine()
    elif engine_type == "cedar":
        from src.module.authorization.infrastructure.condition_engines.cedar_engine import (
            CedarConditionEngine,
        )

        _engine_instance = CedarConditionEngine()
    elif engine_type == "casbin":
        from src.module.authorization.infrastructure.condition_engines.casbin_engine import (
            CasbinConditionEngine,
        )

        _engine_instance = CasbinConditionEngine()
    else:
        from src.share_kernel.domain.exceptions import ConfigurationError

        raise ConfigurationError(f"Unknown condition engine: {engine_type}")

    return _engine_instance



def reset_engine() -> None:
    """Reset the cached engine instance. For testing only."""
    global _engine_instance
    _engine_instance = None