Spaces:
Paused
Paused
| from authlib.common.encoding import json_loads | |
| from authlib.common.urls import urlparse, url_decode | |
| from .errors import InsecureTransportError | |
| class OAuth2Request: | |
| def __init__(self, method: str, uri: str, body=None, headers=None): | |
| InsecureTransportError.check(uri) | |
| #: HTTP method | |
| self.method = method | |
| self.uri = uri | |
| self.body = body | |
| #: HTTP headers | |
| self.headers = headers or {} | |
| self.client = None | |
| self.auth_method = None | |
| self.user = None | |
| self.authorization_code = None | |
| self.refresh_token = None | |
| self.credential = None | |
| def args(self): | |
| query = urlparse.urlparse(self.uri).query | |
| return dict(url_decode(query)) | |
| def form(self): | |
| return self.body or {} | |
| def data(self): | |
| data = {} | |
| data.update(self.args) | |
| data.update(self.form) | |
| return data | |
| def client_id(self) -> str: | |
| """The authorization server issues the registered client a client | |
| identifier -- a unique string representing the registration | |
| information provided by the client. The value is extracted from | |
| request. | |
| :return: string | |
| """ | |
| return self.data.get('client_id') | |
| def response_type(self) -> str: | |
| rt = self.data.get('response_type') | |
| if rt and ' ' in rt: | |
| # sort multiple response types | |
| return ' '.join(sorted(rt.split())) | |
| return rt | |
| def grant_type(self) -> str: | |
| return self.form.get('grant_type') | |
| def redirect_uri(self): | |
| return self.data.get('redirect_uri') | |
| def scope(self) -> str: | |
| return self.data.get('scope') | |
| def state(self): | |
| return self.data.get('state') | |
| class JsonRequest: | |
| def __init__(self, method, uri, body=None, headers=None): | |
| self.method = method | |
| self.uri = uri | |
| self.body = body | |
| self.headers = headers or {} | |
| def data(self): | |
| return json_loads(self.body) | |