1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103
|
"""This module contains authentication handlers compatible with :class:`httpx.Auth`"""
from typing import Generator
from httpx import Auth, Request, Response
from pyDataverse.exceptions import ApiAuthorizationError
class ApiTokenAuth(Auth):
"""An authentication handler to add an API token as the X-Dataverse-key
header.
For more information on how to retrieve an API token and how it is used,
please refer to https://guides.dataverse.org/en/latest/api/auth.html.
"""
def __init__(self, api_token: str):
"""Initializes the auth handler with an API token.
Parameters
----------
api_token : str
The API token retrieved from your Dataverse instance user profile.
Examples
--------
>>> import os
>>> from pyDataverse.api import DataAccessApi
>>> base_url = 'https://demo.dataverse.org'
>>> api_token_auth = ApiTokenAuth(os.getenv('API_TOKEN'))
>>> api = DataAccessApi(base_url, api_token_auth)
"""
if not isinstance(api_token, str):
raise ApiAuthorizationError("API token passed is not a string.")
self.api_token = api_token
def auth_flow(self, request: Request) -> Generator[Request, Response, None]:
"""Adds the X-Dataverse-key header with the API token and yields the
original :class:`httpx.Request`.
Parameters
----------
request : httpx.Request
The request object which requires authentication headers
Yields
------
httpx.Request
The original request with modified headers
"""
request.headers["X-Dataverse-key"] = self.api_token
yield request
class BearerTokenAuth(Auth):
"""An authentication handler to add a Bearer token as defined in `RFC 6750
<https://datatracker.ietf.org/doc/html/rfc6750>`_ to the request.
A bearer token could be obtained from an OIDC provider, for example,
Keycloak.
"""
def __init__(self, bearer_token: str):
"""Initializes the auth handler with a bearer token.
Parameters
----------
bearer_token : str
The bearer token retrieved from your OIDC provider.
Examples
--------
>>> import os
>>> from pyDataverse.api import DataAccessApi
>>> base_url = 'https://demo.dataverse.org'
>>> bearer_token_auth = OAuthBearerTokenAuth(os.getenv('OAUTH_TOKEN'))
>>> api = DataAccessApi(base_url, bearer_token_auth)
"""
if not isinstance(bearer_token, str):
raise ApiAuthorizationError("API token passed is not a string.")
self.bearer_token = bearer_token
def auth_flow(self, request: Request) -> Generator[Request, Response, None]:
"""Adds the X-Dataverse-key header with the API token and yields the
original :class:`httpx.Request`.
Parameters
----------
request : httpx.Request
The request object which requires authentication headers
Yields
------
httpx.Request
The original request with modified headers
"""
request.headers["Authorization"] = f"Bearer {self.bearer_token}"
yield request
|