File: auth.py

package info (click to toggle)
pydataverse 0.3.4-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 1,168 kB
  • sloc: python: 4,862; sh: 61; makefile: 13
file content (103 lines) | stat: -rw-r--r-- 3,321 bytes parent folder | download
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