aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth/authentication/oauth2/endpoints/utilities.py
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2022-12-19 16:02:19 +0300
committerFrederick Muriuki Muriithi2022-12-22 09:05:53 +0300
commitb0641272491eb51d321b1b8a7d062e395e70800f (patch)
treec9b2065ea60399579c4c4d84c648b61ed67402ba /gn3/auth/authentication/oauth2/endpoints/utilities.py
parente9031e28594fcd21371adb2b9b26e17a1df95599 (diff)
downloadgenenetwork3-b0641272491eb51d321b1b8a7d062e395e70800f.tar.gz
auth: implement OAuth2 flow.oauth2_auth_flow
Add code to implement the OAuth2 flow. * Add test fixtures for setting up users and OAuth2 clients * Add tests for token generation with the "Password Grant" flow * Fix some issues with test due to changes in the database connection's row_factory
Diffstat (limited to 'gn3/auth/authentication/oauth2/endpoints/utilities.py')
-rw-r--r--gn3/auth/authentication/oauth2/endpoints/utilities.py30
1 files changed, 30 insertions, 0 deletions
diff --git a/gn3/auth/authentication/oauth2/endpoints/utilities.py b/gn3/auth/authentication/oauth2/endpoints/utilities.py
new file mode 100644
index 0000000..299f151
--- /dev/null
+++ b/gn3/auth/authentication/oauth2/endpoints/utilities.py
@@ -0,0 +1,30 @@
+"""endpoint utilities"""
+from typing import Any, Optional
+
+from flask import current_app
+from pymonad.maybe import Nothing
+
+from gn3.auth import db
+from gn3.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."""
+ __identity__ = lambda val: val
+ token = Nothing
+ with db.connection(current_app.config["AUTH_DB"]) as conn:
+ if token_type_hint == "access_token":
+ token = token_by_access_token(conn, token_str)
+ if token_type_hint == "access_token":
+ token = token_by_refresh_token(conn, token_str)
+
+ 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