Skip to content

core.container.authorization

src.core.container.authorization

Authorization dependency factories.

Casbin RBAC enforcement for role and permission checks. Enforcer is initialized at application startup.

Reference

See docs/architecture/authorization-architecture.md for complete RBAC patterns and permission specifications.

Classes

Functions

init_enforcer async 

init_enforcer() -> AsyncEnforcer

Initialize Casbin AsyncEnforcer at application startup.

Creates enforcer with: - Model config from infrastructure/authorization/model.conf - PostgreSQL adapter for persistent policy storage

MUST be called during FastAPI lifespan startup. Enforcer is app-scoped singleton (stored in _enforcer module variable).

Returns:

Type Description
AsyncEnforcer

Initialized AsyncEnforcer instance.

Raises:

Type Description
RuntimeError

If enforcer is already initialized.
Reference
  • docs/architecture/authorization-architecture.md
Source code in src/core/container/authorization.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def init_enforcer() -> "AsyncEnforcer":
    """Initialize Casbin AsyncEnforcer at application startup.

    Creates enforcer with:
    - Model config from infrastructure/authorization/model.conf
    - PostgreSQL adapter for persistent policy storage

    MUST be called during FastAPI lifespan startup.
    Enforcer is app-scoped singleton (stored in _enforcer module variable).

    Returns:
        Initialized AsyncEnforcer instance.

    Raises:
        RuntimeError: If enforcer is already initialized.

    Reference:
        - docs/architecture/authorization-architecture.md
    """
    global _enforcer

    if _enforcer is not None:
        raise RuntimeError("Enforcer already initialized")

    import os

    import casbin
    from casbin_async_sqlalchemy_adapter import Adapter as CasbinSQLAdapter

    from src.core.container.infrastructure import get_logger

    # Model config path (relative to src/core/container/)
    model_path = os.path.join(
        os.path.dirname(__file__),
        "..",
        "..",
        "infrastructure",
        "authorization",
        "model.conf",
    )

    # Create PostgreSQL adapter for policy storage
    adapter = CasbinSQLAdapter(settings.database_url)

    # Create async enforcer
    _enforcer = casbin.AsyncEnforcer(model_path, adapter)

    # Load policies from database
    await _enforcer.load_policy()

    get_logger().info(
        "casbin_enforcer_initialized",
        model_path=model_path,
    )

    return _enforcer

get_enforcer 

get_enforcer() -> AsyncEnforcer

Get Casbin AsyncEnforcer singleton.

MUST be called after init_enforcer() during startup.

Returns:

Type Description
AsyncEnforcer

The initialized enforcer.

Raises:

Type Description
RuntimeError

If called before init_enforcer().
Source code in src/core/container/authorization.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def get_enforcer() -> "AsyncEnforcer":
    """Get Casbin AsyncEnforcer singleton.

    MUST be called after init_enforcer() during startup.

    Returns:
        The initialized enforcer.

    Raises:
        RuntimeError: If called before init_enforcer().
    """
    if _enforcer is None:
        raise RuntimeError(
            "Enforcer not initialized. Call init_enforcer() during startup."
        )
    return _enforcer

get_authorization async 

get_authorization(
    audit: AuditProtocol = Depends(get_audit),
) -> AuthorizationProtocol

Get authorization adapter (request-scoped).

Creates CasbinAdapter with: - App-scoped enforcer (pre-initialized at startup) - Request-scoped audit (for per-request audit logging) - App-scoped cache, event_bus, logger

Parameters:

Name Type Description Default
audit AuditProtocol

Request-scoped audit adapter for logging authorization checks.

 Depends(get_audit)

Returns:

Type Description
AuthorizationProtocol

CasbinAdapter implementing AuthorizationProtocol.
Usage
Presentation Layer (FastAPI endpoint)

from fastapi import Depends from src.domain.protocols import AuthorizationProtocol

@router.get("/accounts") async def list_accounts( auth: AuthorizationProtocol = Depends(get_authorization), user: User = Depends(get_current_user), ): if not await auth.check_permission(user.id, "accounts", "read"): raise HTTPException(403, "Permission denied") ...

Reference
  • docs/architecture/authorization-architecture.md
Source code in src/core/container/authorization.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
async def get_authorization(
    audit: "AuditProtocol" = Depends(get_audit),
) -> "AuthorizationProtocol":
    """Get authorization adapter (request-scoped).

    Creates CasbinAdapter with:
    - App-scoped enforcer (pre-initialized at startup)
    - Request-scoped audit (for per-request audit logging)
    - App-scoped cache, event_bus, logger

    Args:
        audit: Request-scoped audit adapter for logging authorization checks.

    Returns:
        CasbinAdapter implementing AuthorizationProtocol.

    Usage:
        # Presentation Layer (FastAPI endpoint)
        from fastapi import Depends
        from src.domain.protocols import AuthorizationProtocol

        @router.get("/accounts")
        async def list_accounts(
            auth: AuthorizationProtocol = Depends(get_authorization),
            user: User = Depends(get_current_user),
        ):
            if not await auth.check_permission(user.id, "accounts", "read"):
                raise HTTPException(403, "Permission denied")
            ...

    Reference:
        - docs/architecture/authorization-architecture.md
    """
    from src.infrastructure.authorization.casbin_adapter import CasbinAdapter

    from src.core.container.events import get_event_bus
    from src.core.container.infrastructure import get_cache, get_logger

    return CasbinAdapter(
        enforcer=get_enforcer(),
        cache=get_cache(),
        audit=audit,
        event_bus=get_event_bus(),
        logger=get_logger(),
    )