Michael Hanke
c332abbd1b
Some checks failed
Codespell / Check for spelling errors (push) Successful in 42s
Codespell / Check for spelling errors (pull_request) Successful in 43s
Ruff / Code linting (push) Failing after 52s
Ruff / Code linting (pull_request) Failing after 54s
Test execution / Test-all (push) Successful in 1m27s
Type annotation (PR) / check (pull_request) Failing after 1m30s
43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
"""Use configuration information to fetch token permissions, ids, and incoming_label"""
|
|
|
|
from dump_things_service.abstract_config import (
|
|
Configuration,
|
|
get_token_config_for_representation_and_collection,
|
|
get_token_permissions,
|
|
)
|
|
from dump_things_service.auth import (
|
|
AuthenticationInfo,
|
|
AuthenticationSource,
|
|
InvalidTokenError,
|
|
)
|
|
|
|
|
|
class ConfigAuthenticationSource(AuthenticationSource):
|
|
def __init__(
|
|
self,
|
|
abstract_configuration: Configuration,
|
|
collection_name: str,
|
|
):
|
|
self.abstract_configuration = abstract_configuration
|
|
self.collection_name = collection_name
|
|
|
|
def authenticate(
|
|
self,
|
|
token_representation: str,
|
|
) -> AuthenticationInfo:
|
|
result = get_token_config_for_representation_and_collection(
|
|
self.abstract_configuration,
|
|
self.collection_name,
|
|
token_representation,
|
|
)
|
|
|
|
if not result:
|
|
msg = f'Token not valid for collection `{self.collection_name}`'
|
|
raise InvalidTokenError(msg)
|
|
|
|
_, token_config, token_collection_config = result
|
|
return AuthenticationInfo(
|
|
token_permission=get_token_permissions(token_collection_config.mode),
|
|
user_id=token_config.user_id,
|
|
incoming_label=token_collection_config.incoming_label,
|
|
)
|