Skip to content

auth module

Authenticates with D2S.

Source code in d2spy/auth.py 
 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
104
105
106
107
108
109
110
111
112
113
114
class Auth:
    """Authenticates with D2S."""

    def __init__(self, base_url: str) -> None:
        """Constructor for Auth class.

        Args:
            base_url (str): Base URL for D2S instance.

        Raises:
            ValueError: Raised if unable to communicate with host.
        """
        self.base_url: str = base_url

        if is_valid_base_url(self.base_url) is False:
            raise ValueError("unable to connect to provided host")

        self.session: D2SpySession = D2SpySession()

    def login(
        self,
        email: Optional[str] = None,
        password: Optional[str] = None,
        skip_env: bool = False,
    ) -> Optional[D2SpySession]:
        """Login to D2S platform with email and password. Alternatively, use
        environment variables `D2S_EMAIL` and `D2S_PASSWORD` to set email and password.
        If the password is not passed as an argument and `D2S_PASSWORD` is not set,
        `getpass` will be used to prompt user for password.

        Args:
            email Optional[str]: Email address used to sign in to D2S.
            password Optional[str]: Password used to sign in to D2S.
            skip_env bool: Skip checking environment variables for password.

        Returns:
            Optional[D2SpySession]: Session with user access cookie.
        """
        # Check for email environment variable if not provided as argument
        if not email:
            email = os.environ.get("D2S_EMAIL")
            if not email:
                raise ValueError(
                    "Must provide 'email' to login method as argument or set email "
                    "as environment variable 'D2S_EMAIL'"
                )
        # Check for password environment variable if not provided as argument
        if not password:
            password = os.environ.get("D2S_PASSWORD")
            # Request password from user if not set as environment variable
            if not password or (password and skip_env):
                password = getpass.getpass(prompt="Enter your D2S password:")
        # Credentials that will be sent to D2S auth API
        credentials = {"username": email, "password": password}
        # URL for D2S access-token endpoint
        url = f"{self.base_url}/api/v1/auth/access-token"
        # Post credentials to access-token endpoint
        response = requests.post(url, data=credentials)
        # JWT access token returned for successful request
        if response.status_code == 200 and "access_token" in response.cookies:
            # Add JWT access token to session cookies
            self.session.cookies.set("access_token", response.cookies["access_token"])
            # Fetch user object associated with access token
            user = self.get_current_user()
            # Return dictionary of user attributes and values
            if user:
                # Check if user has api key and set it to session header if so
                if hasattr(user, "api_access_token") and user.api_access_token:
                    self.session.d2s_data = {"API_KEY": user.api_access_token}
                return self.session
            else:
                return None
        elif response.status_code == 401:
            self.login(email, password=None, skip_env=True)
            return None
        else:
            # Print response if request fails
            pretty_print_response(response)
            return None

    def logout(self) -> None:
        """Logout of D2S platform."""
        # Delete access-token cookie from session and end session
        self.session.cookies.clear(domain="", path="/", name="access_token")
        self.session.close()
        print("session ended")

    def get_current_user(self) -> Optional[User]:
        """Get user object for logged in user.

        Returns:
            Optional[User]: User object or None.
        """
        # D2S endpoint for fetching user object for signed in user
        url = f"{self.base_url}/api/v1/users/current"
        # Request user object from D2S instance
        response = self.session.get(url)
        # Return user object if request successful
        if response.status_code == 200:
            return User.from_dict(response.json())
        else:
            # Print response if request fails
            pretty_print_response(response)
            return None

__init__(base_url)

Constructor for Auth class.

Parameters:

Name Type Description Default
base_url str

Base URL for D2S instance.

 required

Raises:

Type Description
ValueError

Raised if unable to communicate with host.
Source code in d2spy/auth.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def __init__(self, base_url: str) -> None:
    """Constructor for Auth class.

    Args:
        base_url (str): Base URL for D2S instance.

    Raises:
        ValueError: Raised if unable to communicate with host.
    """
    self.base_url: str = base_url

    if is_valid_base_url(self.base_url) is False:
        raise ValueError("unable to connect to provided host")

    self.session: D2SpySession = D2SpySession()

get_current_user()

Get user object for logged in user.

Returns:

Type Description
Optional[User]

Optional[User]: User object or None.
Source code in d2spy/auth.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def get_current_user(self) -> Optional[User]:
    """Get user object for logged in user.

    Returns:
        Optional[User]: User object or None.
    """
    # D2S endpoint for fetching user object for signed in user
    url = f"{self.base_url}/api/v1/users/current"
    # Request user object from D2S instance
    response = self.session.get(url)
    # Return user object if request successful
    if response.status_code == 200:
        return User.from_dict(response.json())
    else:
        # Print response if request fails
        pretty_print_response(response)
        return None

login(email=None, password=None, skip_env=False)

Login to D2S platform with email and password. Alternatively, use environment variables D2S_EMAIL and D2S_PASSWORD to set email and password. If the password is not passed as an argument and D2S_PASSWORD is not set, getpass will be used to prompt user for password.

Parameters:

Name Type Description Default
email Optional[str]

Email address used to sign in to D2S.

 None
password Optional[str]

Password used to sign in to D2S.

 None
skip_env bool

Skip checking environment variables for password.

 False

Returns:

Type Description
Optional[D2SpySession]

Optional[D2SpySession]: Session with user access cookie.
Source code in d2spy/auth.py 
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
def login(
    self,
    email: Optional[str] = None,
    password: Optional[str] = None,
    skip_env: bool = False,
) -> Optional[D2SpySession]:
    """Login to D2S platform with email and password. Alternatively, use
    environment variables `D2S_EMAIL` and `D2S_PASSWORD` to set email and password.
    If the password is not passed as an argument and `D2S_PASSWORD` is not set,
    `getpass` will be used to prompt user for password.

    Args:
        email Optional[str]: Email address used to sign in to D2S.
        password Optional[str]: Password used to sign in to D2S.
        skip_env bool: Skip checking environment variables for password.

    Returns:
        Optional[D2SpySession]: Session with user access cookie.
    """
    # Check for email environment variable if not provided as argument
    if not email:
        email = os.environ.get("D2S_EMAIL")
        if not email:
            raise ValueError(
                "Must provide 'email' to login method as argument or set email "
                "as environment variable 'D2S_EMAIL'"
            )
    # Check for password environment variable if not provided as argument
    if not password:
        password = os.environ.get("D2S_PASSWORD")
        # Request password from user if not set as environment variable
        if not password or (password and skip_env):
            password = getpass.getpass(prompt="Enter your D2S password:")
    # Credentials that will be sent to D2S auth API
    credentials = {"username": email, "password": password}
    # URL for D2S access-token endpoint
    url = f"{self.base_url}/api/v1/auth/access-token"
    # Post credentials to access-token endpoint
    response = requests.post(url, data=credentials)
    # JWT access token returned for successful request
    if response.status_code == 200 and "access_token" in response.cookies:
        # Add JWT access token to session cookies
        self.session.cookies.set("access_token", response.cookies["access_token"])
        # Fetch user object associated with access token
        user = self.get_current_user()
        # Return dictionary of user attributes and values
        if user:
            # Check if user has api key and set it to session header if so
            if hasattr(user, "api_access_token") and user.api_access_token:
                self.session.d2s_data = {"API_KEY": user.api_access_token}
            return self.session
        else:
            return None
    elif response.status_code == 401:
        self.login(email, password=None, skip_env=True)
        return None
    else:
        # Print response if request fails
        pretty_print_response(response)
        return None

logout()

Logout of D2S platform.

Source code in d2spy/auth.py 
91
92
93
94
95
96
def logout(self) -> None:
    """Logout of D2S platform."""
    # Delete access-token cookie from session and end session
    self.session.cookies.clear(domain="", path="/", name="access_token")
    self.session.close()
    print("session ended")