Function init_auth_manager

sdk/python/feast/permissions/server/utils.py:87–131

Initialize the global authorization manager. Must be invoked at Feast server initialization time to create the `AuthManager` instance. Args: server_type: The server type. auth_type: The authorization manager type. Raises: ValueError: If any input argument h

(
    server_type: ServerType, auth_type: AuthManagerType, auth_config: AuthConfig
)

def init_auth_manager(
    server_type: ServerType, auth_type: AuthManagerType, auth_config: AuthConfig
):
    """
    Initialize the global authorization manager.
    Must be invoked at Feast server initialization time to create the `AuthManager` instance.

    Args:
        server_type: The server type.
        auth_type: The authorization manager type.

    Raises:
        ValueError: If any input argument has an unmanaged value.
    """
    if auth_type == AuthManagerType.NONE:
        set_auth_manager(AllowAll())
    else:
        token_extractor: TokenExtractor
        token_parser: TokenParser

        if server_type == ServerType.REST:
            token_extractor = RestTokenExtractor()
        elif server_type == ServerType.ARROW:
            token_extractor = ArrowFlightTokenExtractor()
        elif server_type == ServerType.GRPC:
            token_extractor = GrpcTokenExtractor()
        else:
            raise ValueError(f"Unmanaged server type {server_type}")

        if auth_type == AuthManagerType.KUBERNETES:
            from feast.permissions.auth.kubernetes_token_parser import (
                KubernetesTokenParser,
            )

            token_parser = KubernetesTokenParser()
        elif auth_type == AuthManagerType.OIDC:
            assert isinstance(auth_config, OidcAuthConfig)
            token_parser = OidcTokenParser(auth_config=auth_config)
        else:
            raise ValueError(f"Unmanaged authorization manager type {auth_type}")

        auth_manager = AuthManager(
            token_extractor=token_extractor, token_parser=token_parser
        )
        set_auth_manager(auth_manager)