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
|
"""Common oauth2 client utilities."""
from typing import Optional
from urllib.parse import urljoin
from flask import session, current_app as app
from pymonad.maybe import Just, Maybe, Nothing
from pymonad.either import Left, Right, Either
from authlib.integrations.requests_client import OAuth2Session
SCOPE = "profile group role resource register-client user introspect migrate-data"
def oauth2_client():
config = app.config
return OAuth2Session(
config["OAUTH2_CLIENT_ID"], config["OAUTH2_CLIENT_SECRET"],
scope=SCOPE, token_endpoint_auth_method="client_secret_post",
token=session.get("oauth2_token"))
def oauth2_get(uri_path: str, data: dict = {}) -> Either:
token = session.get("oauth2_token")
config = app.config
client = OAuth2Session(
config["OAUTH2_CLIENT_ID"], config["OAUTH2_CLIENT_SECRET"],
token=token, scope=SCOPE)
resp = client.get(
urljoin(config["GN_SERVER_URL"], uri_path),
data=data)
if resp.status_code == 200:
return Right(resp.json())
return Left(resp)
def oauth2_post(
uri_path: str, data: Optional[dict] = None, json: Optional[dict] = None,
**kwargs) -> Either:
token = session.get("oauth2_token")
config = app.config
client = OAuth2Session(
config["OAUTH2_CLIENT_ID"], config["OAUTH2_CLIENT_SECRET"],
token=token, scope=SCOPE)
resp = client.post(
urljoin(config["GN_SERVER_URL"], uri_path), data=data, json=json,
**kwargs)
if resp.status_code == 200:
return Right(resp.json())
return Left(resp)
|
