authentik.tenants.flags 

 1from collections.abc import Callable
 2from copy import copy
 3from functools import wraps
 4from typing import TYPE_CHECKING, Any, Literal
 5
 6from django.db import DatabaseError, InternalError, ProgrammingError
 7
 8from authentik.lib.utils.reflection import all_subclasses
 9
10if TYPE_CHECKING:
11    from authentik.tenants.models import Tenant
12
13
14class Flag[T]:
15    default: T | None = None
16    visibility: Literal["none"] | Literal["public"] | Literal["authenticated"] = "none"
17    description: str | None = None
18
19    def __init_subclass__(cls, key: str, **kwargs):
20        cls.__key = key
21
22    @property
23    def key(self) -> str:
24        return self.__key
25
26    @classmethod
27    def get(cls) -> T | None:
28        from authentik.tenants.utils import get_current_tenant
29
30        flags = {}
31        try:
32            flags: dict[str, Any] = get_current_tenant(["flags"]).flags
33        except DatabaseError, ProgrammingError, InternalError:
34            pass
35        value = flags.get(cls.__key, None)
36        if value is None:
37            return cls().get_default()
38        return value
39
40    def get_default(self) -> T | None:
41        return self.default
42
43    @staticmethod
44    def available(
45        visibility: Literal["none"] | Literal["public"] | Literal["authenticated"] | None = None,
46    ):
47        flags = all_subclasses(Flag)
48        if visibility:
49            for flag in flags:
50                if flag.visibility == visibility:
51                    yield flag
52        else:
53            yield from flags
54
55
56def patch_flag[T](flag: Flag[T], value: T):
57    """Decorator for tests to set a flag to a value"""
58
59    def wrapper_outer(func: Callable):
60        """Set a flag for a test"""
61        from authentik.tenants.utils import get_current_tenant
62
63        def cleanup(tenant: Tenant, flags: dict[str, Any]):
64            tenant.flags = flags
65            tenant.save()
66
67        @wraps(func)
68        def wrapper(*args, **kwargs):
69            tenant = get_current_tenant()
70            old_flags = copy(tenant.flags)
71            tenant.flags[flag().key] = value
72            tenant.save()
73            try:
74                res = func(*args, **kwargs)
75                cleanup(tenant, old_flags)
76                return res
77            finally:
78                cleanup(tenant, old_flags)
79
80        return wrapper
81
82    return wrapper_outer
class Flag(typing.Generic[T]): 
15class Flag[T]:
16    default: T | None = None
17    visibility: Literal["none"] | Literal["public"] | Literal["authenticated"] = "none"
18    description: str | None = None
19
20    def __init_subclass__(cls, key: str, **kwargs):
21        cls.__key = key
22
23    @property
24    def key(self) -> str:
25        return self.__key
26
27    @classmethod
28    def get(cls) -> T | None:
29        from authentik.tenants.utils import get_current_tenant
30
31        flags = {}
32        try:
33            flags: dict[str, Any] = get_current_tenant(["flags"]).flags
34        except DatabaseError, ProgrammingError, InternalError:
35            pass
36        value = flags.get(cls.__key, None)
37        if value is None:
38            return cls().get_default()
39        return value
40
41    def get_default(self) -> T | None:
42        return self.default
43
44    @staticmethod
45    def available(
46        visibility: Literal["none"] | Literal["public"] | Literal["authenticated"] | None = None,
47    ):
48        flags = all_subclasses(Flag)
49        if visibility:
50            for flag in flags:
51                if flag.visibility == visibility:
52                    yield flag
53        else:
54            yield from flags

Abstract base class for generic types.

On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::

class Mapping[KT, VT]:
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

On older versions of Python, however, generic classes have to explicitly inherit from Generic.

After a class has been declared to be generic, it can then be used as follows::

def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default
default: T | None = None
visibility: Literal['none'] | Literal['public'] | Literal['authenticated'] = 'none'
description: str | None = None
key: str 
23    @property
24    def key(self) -> str:
25        return self.__key
@classmethod
def get(cls) -> T | None: 
27    @classmethod
28    def get(cls) -> T | None:
29        from authentik.tenants.utils import get_current_tenant
30
31        flags = {}
32        try:
33            flags: dict[str, Any] = get_current_tenant(["flags"]).flags
34        except DatabaseError, ProgrammingError, InternalError:
35            pass
36        value = flags.get(cls.__key, None)
37        if value is None:
38            return cls().get_default()
39        return value
def get_default(self) -> T | None: 
41    def get_default(self) -> T | None:
42        return self.default
@staticmethod
def available( visibility: Literal['none'] | Literal['public'] | Literal['authenticated'] | None = None): 
44    @staticmethod
45    def available(
46        visibility: Literal["none"] | Literal["public"] | Literal["authenticated"] | None = None,
47    ):
48        flags = all_subclasses(Flag)
49        if visibility:
50            for flag in flags:
51                if flag.visibility == visibility:
52                    yield flag
53        else:
54            yield from flags
def patch_flag(flag: Flag[T], value: T): 
57def patch_flag[T](flag: Flag[T], value: T):
58    """Decorator for tests to set a flag to a value"""
59
60    def wrapper_outer(func: Callable):
61        """Set a flag for a test"""
62        from authentik.tenants.utils import get_current_tenant
63
64        def cleanup(tenant: Tenant, flags: dict[str, Any]):
65            tenant.flags = flags
66            tenant.save()
67
68        @wraps(func)
69        def wrapper(*args, **kwargs):
70            tenant = get_current_tenant()
71            old_flags = copy(tenant.flags)
72            tenant.flags[flag().key] = value
73            tenant.save()
74            try:
75                res = func(*args, **kwargs)
76                cleanup(tenant, old_flags)
77                return res
78            finally:
79                cleanup(tenant, old_flags)
80
81        return wrapper
82
83    return wrapper_outer

Decorator for tests to set a flag to a value