authentik.endpoints.controller 

 1from django.db import models
 2from structlog.stdlib import BoundLogger, get_logger
 3
 4from authentik.endpoints.models import Connector
 5from authentik.flows.stage import StageView
 6from authentik.lib.sentry import SentryIgnoredException
 7
 8MERGED_VENDOR = "goauthentik.io/@merged"
 9
10
11class Capabilities(models.TextChoices):
12    # Automatically enrolled through user action
13    ENROLL_AUTOMATIC_USER = "enroll_automatic_user"
14    # Automatically enrolled through connector integration
15    ENROLL_AUTOMATIC_API = "enroll_automatic_api"
16    # Manually enrolled with user interaction (user scanning a QR code for example)
17    ENROLL_MANUAL_USER = "enroll_manual_user"
18    # Supported for use with Endpoints stage
19    STAGE_ENDPOINTS = "stage_endpoints"
20
21
22class ConnectorSyncException(SentryIgnoredException):
23    """Base exceptions for errors during sync"""
24
25
26class BaseController[T: "Connector"]:
27
28    connector: T
29    logger: BoundLogger
30
31    def __init__(self, connector: T) -> None:
32        self.connector = connector
33        self.logger = get_logger().bind(connector=connector.name)
34
35    @staticmethod
36    def vendor_identifier() -> str:
37        raise NotImplementedError
38
39    def capabilities(self) -> list[Capabilities]:
40        return []
41
42    def stage_view_enrollment(self) -> StageView | None:
43        return None
44
45    def stage_view_authentication(self) -> StageView | None:
46        return None
47
48    def sync_endpoints(self):
49        raise NotImplementedError
MERGED_VENDOR = 'goauthentik.io/@merged'
class Capabilities(django.db.models.enums.TextChoices): 
12class Capabilities(models.TextChoices):
13    # Automatically enrolled through user action
14    ENROLL_AUTOMATIC_USER = "enroll_automatic_user"
15    # Automatically enrolled through connector integration
16    ENROLL_AUTOMATIC_API = "enroll_automatic_api"
17    # Manually enrolled with user interaction (user scanning a QR code for example)
18    ENROLL_MANUAL_USER = "enroll_manual_user"
19    # Supported for use with Endpoints stage
20    STAGE_ENDPOINTS = "stage_endpoints"

Class for creating enumerated string choices.

ENROLL_AUTOMATIC_USER = Capabilities.ENROLL_AUTOMATIC_USER
ENROLL_AUTOMATIC_API = Capabilities.ENROLL_AUTOMATIC_API
ENROLL_MANUAL_USER = Capabilities.ENROLL_MANUAL_USER
STAGE_ENDPOINTS = Capabilities.STAGE_ENDPOINTS
class ConnectorSyncException(authentik.lib.sentry.SentryIgnoredException): 
23class ConnectorSyncException(SentryIgnoredException):
24    """Base exceptions for errors during sync"""

Base exceptions for errors during sync

class BaseController(typing.Generic[T]): 
27class BaseController[T: "Connector"]:
28
29    connector: T
30    logger: BoundLogger
31
32    def __init__(self, connector: T) -> None:
33        self.connector = connector
34        self.logger = get_logger().bind(connector=connector.name)
35
36    @staticmethod
37    def vendor_identifier() -> str:
38        raise NotImplementedError
39
40    def capabilities(self) -> list[Capabilities]:
41        return []
42
43    def stage_view_enrollment(self) -> StageView | None:
44        return None
45
46    def stage_view_authentication(self) -> StageView | None:
47        return None
48
49    def sync_endpoints(self):
50        raise NotImplementedError

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
BaseController(connector: T) 
32    def __init__(self, connector: T) -> None:
33        self.connector = connector
34        self.logger = get_logger().bind(connector=connector.name)
connector: T
logger: structlog.stdlib.BoundLogger
@staticmethod
def vendor_identifier() -> str: 
36    @staticmethod
37    def vendor_identifier() -> str:
38        raise NotImplementedError
def capabilities(self) -> list[Capabilities]: 
40    def capabilities(self) -> list[Capabilities]:
41        return []
def stage_view_enrollment(self) -> authentik.flows.stage.StageView | None: 
43    def stage_view_enrollment(self) -> StageView | None:
44        return None
def stage_view_authentication(self) -> authentik.flows.stage.StageView | None: 
46    def stage_view_authentication(self) -> StageView | None:
47        return None
def sync_endpoints(self): 
49    def sync_endpoints(self):
50        raise NotImplementedError