|
1 | 1 | import os |
| 2 | +import urllib.parse as urlparse |
| 3 | +import webbrowser |
2 | 4 | from http.server import BaseHTTPRequestHandler, HTTPServer |
| 5 | + |
3 | 6 | from requests_oauthlib import OAuth2Session |
4 | | -import webbrowser |
5 | | -import urllib.parse as urlparse |
6 | 7 |
|
7 | 8 | # Environment setup |
8 | 9 | os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1" |
9 | 10 | USABLE_PORT_RANGE = (29170, 292180) |
10 | 11 |
|
11 | 12 |
|
12 | 13 | class OAuthClient: |
13 | | - def __init__(self, server_url: str = "http://localhost:8000", client_id: str ='', scopes: list = None) -> None: |
| 14 | + def __init__(self, server_url: str = "http://localhost:8000", client_id: str = '', scopes: list = None) -> None: |
14 | 15 | self.server_url = server_url |
15 | 16 | self.client_id = client_id |
16 | 17 | self.scopes = scopes if scopes is not None else [] |
17 | 18 |
|
18 | 19 | self._handler_wrapper = RequestHandlerWrapper(self) |
19 | 20 | self._setup_callback() |
20 | 21 | self._poll_user() |
21 | | - |
| 22 | + |
22 | 23 | def get_url(self, path: str) -> str: |
23 | 24 | """Get the authorization URL.""" |
24 | 25 | return urlparse.urljoin(self.server_url, path) |
@@ -53,7 +54,7 @@ def callback(self, callback_url: str): |
53 | 54 |
|
54 | 55 |
|
55 | 56 | class RequestHandlerWrapper: |
56 | | - """Provides callback for OIDC endpint.""" |
| 57 | + """Provides callback for OIDC endpoint.""" |
57 | 58 | def __init__(self, oauth_client) -> None: |
58 | 59 | self.done = False |
59 | 60 | self.error = None |
@@ -97,6 +98,7 @@ def log_message(self, *args): |
97 | 98 |
|
98 | 99 | return RequestHandler |
99 | 100 |
|
| 101 | + |
100 | 102 | class OAuthError(Exception): |
101 | 103 | """Exception raised during the OAuth process.""" |
102 | 104 | def __init__(self, message: str) -> None: |
|
0 commit comments