diff options
author | Frederick Muriuki Muriithi | 2022-12-19 16:02:19 +0300 |
---|---|---|
committer | Frederick Muriuki Muriithi | 2022-12-22 09:05:53 +0300 |
commit | b0641272491eb51d321b1b8a7d062e395e70800f (patch) | |
tree | c9b2065ea60399579c4c4d84c648b61ed67402ba /gn3/auth/authentication/oauth2/endpoints/utilities.py | |
parent | e9031e28594fcd21371adb2b9b26e17a1df95599 (diff) | |
download | genenetwork3-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.py | 30 |
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 |