3D-printing/Cura
1
0
Fork 0
mirror of https://github.com/Ultimaker/Cura.git synced 2025-07-12 01:07:52 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Fixed up final bit of mypy issues

Browse source 
CURA-5744
This commit is contained in:
Jaime van Kessel 2018-09-21 14:12:31 +02:00
parent d0fc4878c2
commit 060ea0b762
5 changed files with 39 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

27
cura/OAuth2/AuthorizationRequestHandler.py
View file

@ -1,12 +1,15 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional, Callable, Tuple, Dict, Any, List
from typing import Optional, Callable, Tuple, Dict, Any, List, TYPE_CHECKING
from http.server import BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
from cura.OAuth2.Models import AuthenticationResponse, ResponseData, HTTP_STATUS, ResponseStatus
from cura.OAuth2.Models import AuthenticationResponse, ResponseData, HTTP_STATUS
if TYPE_CHECKING:
from cura.OAuth2.Models import ResponseStatus
class AuthorizationRequestHandler(BaseHTTPRequestHandler):
@ -15,15 +18,15 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
It also requests the access token for the 2nd stage of the OAuth flow.
"""
def __init__(self, request, client_address, server):
def __init__(self, request, client_address, server) -> None:
super().__init__(request, client_address, server)
# These values will be injected by the HTTPServer that this handler belongs to.
self.authorization_helpers = None # type: AuthorizationHelpers
self.authorization_callback = None # type: Callable[[AuthenticationResponse], None]
self.verification_code = None # type: str
self.authorization_helpers = None # type: Optional[AuthorizationHelpers]
self.authorization_callback = None # type: Optional[Callable[[AuthenticationResponse], None]]
self.verification_code = None # type: Optional[str]
def do_GET(self):
def do_GET(self) -> None:
"""Entry point for GET requests"""
# Extract values from the query string.
@ -44,7 +47,7 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
# If there is data in the response, we send it.
self._sendData(server_response.data_stream)
if token_response:
if token_response and self.authorization_callback is not None:
# Trigger the callback if we got a response.
# This will cause the server to shut down, so we do it at the very end of the request handling.
self.authorization_callback(token_response)
@ -56,7 +59,7 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
:return: HTTP ResponseData containing a success page to show to the user.
"""
code = self._queryGet(query, "code")
if code:
if code and self.authorization_helpers is not None and self.verification_code is not None:
# If the code was returned we get the access token.
token_response = self.authorization_helpers.getAccessTokenUsingAuthorizationCode(
code, self.verification_code)
@ -74,6 +77,8 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
success=False,
error_message="Something unexpected happened when trying to log in, please try again."
)
if self.authorization_helpers is None:
return ResponseData(), token_response
return ResponseData(
status=HTTP_STATUS["REDIRECT"],
@ -83,7 +88,7 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
), token_response
@staticmethod
def _handleNotFound() -> "ResponseData":
def _handleNotFound() -> ResponseData:
"""Handle all other non-existing server calls."""
return ResponseData(status=HTTP_STATUS["NOT_FOUND"], content_type="text/html", data_stream=b"Not found.")
@ -100,6 +105,6 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
self.wfile.write(data)
@staticmethod
def _queryGet(query_data: Dict[Any, List], key: str, default=None) -> Optional[str]:
def _queryGet(query_data: Dict[Any, List], key: str, default: Optional[str]=None) -> Optional[str]:
"""Helper for getting values from a pre-parsed query string"""
return query_data.get(key, [default])[0]

13
cura/OAuth2/AuthorizationRequestServer.py
View file

@ -1,8 +1,11 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from http.server import HTTPServer
from typing import Callable, Any, TYPE_CHECKING
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
if TYPE_CHECKING:
from cura.OAuth2.Models import AuthenticationResponse
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
class AuthorizationRequestServer(HTTPServer):
@ -14,12 +17,12 @@ class AuthorizationRequestServer(HTTPServer):
def setAuthorizationHelpers(self, authorization_helpers: "AuthorizationHelpers") -> None:
"""Set the authorization helpers instance on the request handler."""
self.RequestHandlerClass.authorization_helpers = authorization_helpers
self.RequestHandlerClass.authorization_helpers = authorization_helpers # type: ignore
def setAuthorizationCallback(self, authorization_callback) -> None:
def setAuthorizationCallback(self, authorization_callback: Callable[["AuthenticationResponse"], Any]) -> None:
"""Set the authorization callback on the request handler."""
self.RequestHandlerClass.authorization_callback = authorization_callback
self.RequestHandlerClass.authorization_callback = authorization_callback # type: ignore
def setVerificationCode(self, verification_code: str) -> None:
"""Set the verification code on the request handler."""
self.RequestHandlerClass.verification_code = verification_code
self.RequestHandlerClass.verification_code = verification_code # type: ignore

16
cura/OAuth2/AuthorizationService.py
View file

@ -2,17 +2,18 @@
# Cura is released under the terms of the LGPLv3 or higher.
import json
import webbrowser
from typing import Optional
from typing import Optional, TYPE_CHECKING
from urllib.parse import urlencode
# As this module is specific for Cura plugins, we can rely on these imports.
from UM.Logger import Logger
from UM.Signal import Signal
# Plugin imports need to be relative to work in final builds.
from cura.OAuth2.LocalAuthorizationServer import LocalAuthorizationServer
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
from cura.OAuth2.Models import OAuth2Settings, AuthenticationResponse, UserProfile
from cura.OAuth2.Models import AuthenticationResponse
if TYPE_CHECKING:
from cura.OAuth2.Models import UserProfile, OAuth2Settings
class AuthorizationService:
@ -32,7 +33,7 @@ class AuthorizationService:
self._auth_helpers = AuthorizationHelpers(settings)
self._auth_url = "{}/authorize".format(self._settings.OAUTH_SERVER_URL)
self._auth_data = None # type: Optional[AuthenticationResponse]
self._user_profile = None # type: Optional[UserProfile]
self._user_profile = None # type: Optional["UserProfile"]
self._cura_preferences = preferences
self._server = LocalAuthorizationServer(self._auth_helpers, self._onAuthStateChanged, daemon=True)
self._loadAuthData()
@ -75,7 +76,6 @@ class AuthorizationService:
def getAccessToken(self) -> Optional[str]:
"""
Get the access token response data.
:return: Dict containing token data.
"""
if not self.getUserProfile():
# We check if we can get the user profile.
@ -130,7 +130,7 @@ class AuthorizationService:
# Start a local web server to receive the callback URL on.
self._server.start(verification_code)
def _onAuthStateChanged(self, auth_response: "AuthenticationResponse") -> None:
def _onAuthStateChanged(self, auth_response: AuthenticationResponse) -> None:
"""Callback method for an authentication flow."""
if auth_response.success:
self._storeAuthData(auth_response)
@ -150,7 +150,7 @@ class AuthorizationService:
except ValueError as err:
Logger.log("w", "Could not load auth data from preferences: %s", err)
def _storeAuthData(self, auth_data: Optional["AuthenticationResponse"] = None) -> None:
def _storeAuthData(self, auth_data: Optional[AuthenticationResponse] = None) -> None:
"""Store authentication data in preferences and locally."""
self._auth_data = auth_data
if auth_data:

13
cura/OAuth2/LocalAuthorizationServer.py
View file

@ -1,22 +1,21 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import threading
from http.server import HTTPServer
from typing import Optional, Callable, Any
from typing import Optional, Callable, Any, TYPE_CHECKING
# As this module is specific for Cura plugins, we can rely on these imports.
from UM.Logger import Logger
# Plugin imports need to be relative to work in final builds.
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
from cura.OAuth2.AuthorizationRequestServer import AuthorizationRequestServer
from cura.OAuth2.AuthorizationRequestHandler import AuthorizationRequestHandler
from cura.OAuth2.Models import AuthenticationResponse
if TYPE_CHECKING:
from cura.OAuth2.Models import AuthenticationResponse
class LocalAuthorizationServer:
def __init__(self, auth_helpers: "AuthorizationHelpers",
auth_state_changed_callback: "Callable[[AuthenticationResponse], Any]",
auth_state_changed_callback: Callable[["AuthenticationResponse"], Any],
daemon: bool) -> None:
"""
:param auth_helpers: An instance of the authorization helpers class.
@ -62,7 +61,7 @@ class LocalAuthorizationServer:
def stop(self) -> None:
""" Stops the web server if it was running. Also deletes the objects. """
Logger.log("d", "Stopping local web server...")
Logger.log("d", "Stopping local oauth2 web server...")
if self._web_server:
self._web_server.server_close()

2
cura/OAuth2/Models.py
View file

@ -46,7 +46,7 @@ class ResponseStatus(BaseModel):
# Response data template.
class ResponseData(BaseModel):
status = None # type: Optional[ResponseStatus]
status = None # type: ResponseStatus
data_stream = None # type: Optional[bytes]
redirect_uri = None # type: Optional[str]
content_type = "text/html" # type: str