aboutsummaryrefslogtreecommitdiff
path: root/gn_auth/auth/authentication/oauth2/endpoints/utilities.py
blob: 08b2a3b55f6066e9d3de9c96e5c38ae8dc5f1e8e (plain)
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
"""endpoint utilities"""
from typing import Any, Optional

from flask import current_app
from pymonad.maybe import Nothing

from gn_auth.auth.db import sqlite3 as db
from gn_auth.auth.authentication.oauth2.models.oauth2token import (
    OAuth2Token, token_by_access_token, token_by_refresh_token)

def query_token(# pylint: disable=[unused-argument]
        endpoint_object: Any, token_str: str, token_type_hint) -> Optional[
            OAuth2Token]:
    """Retrieve the token from the database."""
    def __identity__(val):
        """Identity function."""
        return val
    token = Nothing
    with db.connection(current_app.config["AUTH_DB"]) as conn:
        match token_type_hint:
            case "access_token":
                token = token_by_access_token(
                    conn, token_str
                )
            case "refresh_token":
                token = token_by_refresh_token(
                    conn, token_str
                )
            case _:
                token = Nothing

        return token.maybe(
            token_by_access_token(conn, token_str).maybe(
                token_by_refresh_token(conn, token_str).maybe(
                    None, __identity__),
                __identity__),
            __identity__)

    return None