chore: Adds typing annotations to Coriolis

coriolis/api/__init__.py

@@ -35,13 +35,13 @@ def root_app_factory(loader, global_conf, **local_conf):
 
 
 class APIMapper(routes.Mapper):
-    def routematch(self, url=None, environ=None):
+    def routematch(self, url: str | None = None, environ=None):
         if url is "":
             result = self._match("", environ)
             return result[0], result[1]
         return routes.Mapper.routematch(self, url, environ)
 
-    def connect(self, *args, **kwargs):
+    def connect(self, *args, **kwargs) -> None:
         # NOTE(inhye): Default the format part of a route to only accept json
         #             and xml so it doesn't eat all characters after a '.'
         #             in the url.
@@ -88,7 +88,7 @@ def __init__(self, ext_mgr=None):
         self._setup_extensions(ext_mgr)
         super(APIRouter, self).__init__(mapper)
 
-    def _setup_ext_routes(self, mapper, ext_mgr):
+    def _setup_ext_routes(self, mapper: APIMapper, ext_mgr) -> None:
         for resource in ext_mgr.get_resources():
             LOG.debug('Extended resource: %s', resource.collection)
 
@@ -108,7 +108,7 @@ def _setup_ext_routes(self, mapper, ext_mgr):
             if resource.custom_routes_fn:
                 resource.custom_routes_fn(mapper, wsgi_resource)
 
-    def _setup_extensions(self, ext_mgr):
+    def _setup_extensions(self, ext_mgr) -> None:
         for extension in ext_mgr.get_controller_extensions():
             collection = extension.collection
             controller = extension.controller
@@ -132,5 +132,5 @@ def _setup_extensions(self, ext_mgr):
             resource.register_actions(controller)
             resource.register_extensions(controller)
 
-    def _setup_routes(self, mapper, ext_mgr):
+    def _setup_routes(self, mapper: APIMapper, ext_mgr):
         raise NotImplementedError

coriolis/api/common.py

@@ -4,7 +4,7 @@
 from coriolis import utils
 
 
-def get_paging_params(req):
+def get_paging_params(req) -> tuple:
     marker = req.GET.get("marker")
     limit = req.GET.get("limit")
     if limit is not None:
@@ -13,8 +13,10 @@ def get_paging_params(req):
 
 
 def get_sort_params(
-    req, default_keys=('created_at', 'id'), default_dirs=('desc', 'desc')
-):
+    req,
+    default_keys: tuple[str, str] = ('created_at', 'id'),
+    default_dirs: tuple[str, str] = ('desc', 'desc'),
+) -> tuple[list, list]:
     """Retrieves sort keys/directions parameters.
 
     Processes the parameters to create a list of sort keys and sort directions

coriolis/api/middleware/fault.py

@@ -32,15 +32,15 @@ class FaultWrapper(wsgi.Middleware):
     _status_to_type = {}
 
     @staticmethod
-    def status_to_type(status):
+    def status_to_type(status: int) -> webob.exc.HTTPNotFound:
         if not FaultWrapper._status_to_type:
             for clazz in utils.walk_class_hierarchy(webob.exc.HTTPError):
                 FaultWrapper._status_to_type[clazz.code] = clazz
         return FaultWrapper._status_to_type.get(
             status, webob.exc.HTTPInternalServerError
         )()
 
-    def _error(self, inner, req):
+    def _error(self, inner: exception.NotFound, req: wsgi.Request) -> wsgi.Fault:
         LOG.exception(
             _LE("Caught error: %(type)s %(error)s"),
             {'type': type(inner), 'error': inner},
@@ -67,7 +67,7 @@ def _error(self, inner, req):
         return wsgi.Fault(outer)
 
     @webob.dec.wsgify(RequestClass=wsgi.Request)
-    def __call__(self, req):
+    def __call__(self, req: wsgi.Request) -> wsgi.Fault | webob.Response:
         try:
             return req.get_response(self.application)
         except Exception as ex:

coriolis/api/v1/deployment_actions.py

@@ -1,6 +1,8 @@
 # Copyright 2024 Cloudbase Solutions Srl
 # All Rights Reserved.
 
+from typing import NoReturn
+
 from webob import exc
 
 from coriolis import exception
@@ -15,7 +17,9 @@ def __init__(self):
         super(DeploymentActionsController, self).__init__()
 
     @api_wsgi.action('cancel')
-    def _cancel(self, req, id, body):
+    def _cancel(
+        self, req: api_wsgi.Request, id: str, body: dict[str, dict]
+    ) -> NoReturn:
         context = req.environ['coriolis.context']
         context.can(deployment_policies.get_deployments_policy_label("cancel"))
         try:
@@ -29,5 +33,5 @@ def _cancel(self, req, id, body):
             raise exc.HTTPNotFound(explanation=ex.msg)
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(DeploymentActionsController())

coriolis/api/v1/deployments.py

@@ -1,6 +1,8 @@
 # Copyright 2024 Cloudbase Solutions Srl
 # All Rights Reserved.
 
+from typing import NoReturn
+
 from oslo_log import log as logging
 from webob import exc
 
@@ -22,7 +24,7 @@ def __init__(self):
         self._endpoints_api = endpoints_api.API()
         super(DeploymentsController, self).__init__()
 
-    def show(self, req, id):
+    def show(self, req: api_wsgi.Request, id: str) -> dict[str, dict]:
         context = req.environ["coriolis.context"]
         context.can(deployment_policies.get_deployments_policy_label("show"))
         include_task_info = api_utils.get_bool_url_arg(
@@ -36,7 +38,7 @@ def show(self, req, id):
 
         return deployment_view.single(deployment)
 
-    def _get_filters(self, req) -> dict:
+    def _get_filters(self, req: api_wsgi.Request) -> dict:
         filters = {}
         # For simplicity and consistency, we'll use "status" to search for a
         # given "last_execution_status".
@@ -49,7 +51,7 @@ def _get_filters(self, req) -> dict:
             filters["status"] = status
         return filters
 
-    def _list(self, req):
+    def _list(self, req: api_wsgi.Request) -> dict[str, list]:
         show_deleted = api_utils.get_bool_url_arg(req, "show_deleted", default=False)
         context = req.environ["coriolis.context"]
         context.show_deleted = show_deleted
@@ -75,14 +77,14 @@ def _list(self, req):
             )
         )
 
-    def index(self, req):
+    def index(self, req: api_wsgi.Request) -> dict[str, list]:
         return self._list(req)
 
     def detail(self, req):
         return self._list(req)
 
     @api_utils.format_keyerror_message(resource='deployment', method='create')
-    def _validate_deployment_input(self, context, body):
+    def _validate_deployment_input(self, context, body) -> tuple:
         deployment = body["deployment"]
 
         transfer_id = deployment.get("transfer_id", "")
@@ -111,7 +113,7 @@ def _validate_deployment_input(self, context, body):
             user_scripts,
         )
 
-    def create(self, req, body):
+    def create(self, req: api_wsgi.Request, body: dict[str, dict]) -> dict[str, dict]:
         context = req.environ['coriolis.context']
         context.can(deployment_policies.get_deployments_policy_label("create"))
 
@@ -138,7 +140,7 @@ def create(self, req, body):
 
         return deployment_view.single(deployment)
 
-    def delete(self, req, id):
+    def delete(self, req: api_wsgi.Request, id: str) -> NoReturn:
         context = req.environ['coriolis.context']
         context.can(deployment_policies.get_deployments_policy_label("delete"))
         try:
@@ -148,5 +150,5 @@ def delete(self, req, id):
             raise exc.HTTPNotFound(explanation=ex.msg)
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(DeploymentsController())

coriolis/api/v1/diagnostics.py

@@ -16,12 +16,12 @@ def __init__(self):
         self._diag_api = api.API()
         super(DiagnosticsController, self).__init__()
 
-    def index(self, req):
+    def index(self, req: api_wsgi.Request) -> dict[str, list]:
         context = req.environ['coriolis.context']
         context.can(diagnostics.get_diagnostics_policy_label("get"))
 
         return diagnostic_view.collection(self._diag_api.get(context))
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(DiagnosticsController())

coriolis/api/v1/endpoint_actions.py

@@ -15,7 +15,9 @@ def __init__(self):
         super(EndpointActionsController, self).__init__()
 
     @api_wsgi.action('validate-connection')
-    def _validate_connection(self, req, id, body):
+    def _validate_connection(
+        self, req: api_wsgi.Request, id: str, body: dict[str, None]
+    ) -> dict[str, dict]:
         context = req.environ['coriolis.context']
         context.can(
             "%s:validate_connection" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX)
@@ -29,5 +31,5 @@ def _validate_connection(self, req, id, body):
             raise exc.HTTPNotFound(explanation=ex.msg)
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointActionsController())

coriolis/api/v1/endpoint_destination_minion_pool_options.py

@@ -17,7 +17,7 @@ def __init__(self):
         self._minion_pool_options_api = api.API()
         super(EndpointDestinationMinionPoolOptionsController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> dict[str, list]:
         context = req.environ['coriolis.context']
         context.can(
             "%s:list_destination_minion_pool_options"
@@ -43,5 +43,5 @@ def index(self, req, endpoint_id):
         )
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointDestinationMinionPoolOptionsController())

coriolis/api/v1/endpoint_destination_options.py

@@ -17,7 +17,7 @@ def __init__(self):
         self._destination_options_api = api.API()
         super(EndpointDestinationOptionsController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> dict[str, list]:
         context = req.environ['coriolis.context']
         context.can(
             "%s:list_destination_options" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX)
@@ -42,5 +42,5 @@ def index(self, req, endpoint_id):
         )
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointDestinationOptionsController())

coriolis/api/v1/endpoint_instances.py

@@ -19,7 +19,7 @@ def __init__(self):
         self._instance_api = api.API()
         super(EndpointInstanceController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> dict[str, list]:
         context = req.environ['coriolis.context']
         context.can("%s:list_instances" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX))
         marker, limit = common.get_paging_params(req)
@@ -44,7 +44,7 @@ def index(self, req, endpoint_id):
             )
         )
 
-    def show(self, req, endpoint_id, id):
+    def show(self, req: api_wsgi.Request, endpoint_id: str, id: str) -> dict[str, dict]:
         context = req.environ['coriolis.context']
         context.can("%s:get_instance" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX))
 
@@ -65,5 +65,5 @@ def show(self, req, endpoint_id, id):
         )
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointInstanceController())

coriolis/api/v1/endpoint_inventory.py

@@ -18,7 +18,7 @@ def __init__(self):
         self._endpoint_resources_api = api.API()
         super(EndpointInventoryController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> api_wsgi.ResponseObject:
         context = req.environ['coriolis.context']
         context.can("%s:export_inventory" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX))
 
@@ -35,5 +35,5 @@ def index(self, req, endpoint_id):
         return api_wsgi.ResponseObject(csv_content)
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointInventoryController())

coriolis/api/v1/endpoint_networks.py

@@ -17,7 +17,7 @@ def __init__(self):
         self._network_api = api.API()
         super(EndpointNetworkController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> dict[str, list]:
         context = req.environ['coriolis.context']
         context.can("%s:list_networks" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX))
         env = req.GET.get("env")
@@ -31,5 +31,5 @@ def index(self, req, endpoint_id):
         )
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointNetworkController())

coriolis/api/v1/endpoint_source_minion_pool_options.py

@@ -17,7 +17,7 @@ def __init__(self):
         self._minion_pool_options_api = api.API()
         super(EndpointSourceMinionPoolOptionsController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> dict[str, list]:
         context = req.environ['coriolis.context']
         context.can(
             "%s:list_source_minion_pool_options"
@@ -43,5 +43,5 @@ def index(self, req, endpoint_id):
         )
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointSourceMinionPoolOptionsController())

coriolis/api/v1/endpoint_source_options.py

@@ -17,7 +17,7 @@ def __init__(self):
         self._source_options_api = api.API()
         super(EndpointSourceOptionsController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> dict[str, list]:
         context = req.environ['coriolis.context']
         context.can(
             "%s:list_source_options" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX)
@@ -42,5 +42,5 @@ def index(self, req, endpoint_id):
         )
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointSourceOptionsController())

coriolis/api/v1/endpoint_storage.py

@@ -17,7 +17,7 @@ def __init__(self):
         self._storage_api = api.API()
         super(EndpointStorageController, self).__init__()
 
-    def index(self, req, endpoint_id):
+    def index(self, req: api_wsgi.Request, endpoint_id: str) -> dict[str, dict]:
         context = req.environ['coriolis.context']
         context.can("%s:list_storage" % (endpoint_policies.ENDPOINTS_POLICY_PREFIX))
         env = req.GET.get("env")
@@ -31,5 +31,5 @@ def index(self, req, endpoint_id):
         )
 
 
-def create_resource():
+def create_resource() -> api_wsgi.Resource:
     return api_wsgi.Resource(EndpointStorageController())