Format code with Ruff + Flake8

Author: fabi200123

coriolis/api-refs/source/conf.py

@@ -1,15 +1,11 @@
-
-extensions = [
-    'openstackdocstheme',
-    'os_api_ref'
-]
+extensions = ['openstackdocstheme', 'os_api_ref']
 
 source_suffix = '.rst'
 
 master_doc = 'index'
 
-project = u'Coriolis API Reference'
-copyright = u'2018-present, Cloudbase Solutions S.R.L'
+project = 'Coriolis API Reference'
+copyright = '2018-present, Cloudbase Solutions S.R.L'
 
 repository_name = 'cloudbase/coriolis'
 bug_project = 'coriolis'

coriolis/api/__init__.py

@@ -18,16 +18,14 @@
 WSGI middleware for OpenStack API controllers.
 """
 
-from paste import urlmap
 import routes
-
 from oslo_log import log as logging
 from oslo_service import wsgi as base_wsgi
+from paste import urlmap
 
-from coriolis.api import wsgi
 from coriolis import exception
-from coriolis.i18n import _, _LW  # noqa
-
+from coriolis.api import wsgi
+from coriolis.i18n import _LW, _  # noqa
 
 LOG = logging.getLogger(__name__)
 
@@ -61,16 +59,13 @@ def resource(self, member_name, collection_name, **kwargs):
             parent_resource = kwargs['parent_resource']
             p_collection = parent_resource['collection_name']
             p_member = parent_resource['member_name']
-            kwargs['path_prefix'] = '{project_id}/%s/:%s_id' % (p_collection,
-                                                                p_member)
-        routes.Mapper.resource(self,
-                               member_name,
-                               collection_name,
-                               **kwargs)
+            kwargs['path_prefix'] = '{project_id}/%s/:%s_id' % (p_collection, p_member)
+        routes.Mapper.resource(self, member_name, collection_name, **kwargs)
 
 
 class APIRouter(base_wsgi.Router):
     """Routes requests on the API to the appropriate controller and method."""
+
     ExtensionManager = None  # override in subclasses
 
     @classmethod
@@ -83,7 +78,8 @@ def __init__(self, ext_mgr=None):
                 ext_mgr = self.ExtensionManager()
             else:
                 raise exception.CoriolisException(
-                    _("Must specify an ExtensionManager class"))
+                    _("Must specify an ExtensionManager class")
+                )
 
         mapper = ProjectMapper()
         self.resources = {}
@@ -94,15 +90,15 @@ def __init__(self, ext_mgr=None):
 
     def _setup_ext_routes(self, mapper, ext_mgr):
         for resource in ext_mgr.get_resources():
-            LOG.debug('Extended resource: %s',
-                      resource.collection)
+            LOG.debug('Extended resource: %s', resource.collection)
 
             wsgi_resource = wsgi.Resource(resource.controller)
             self.resources[resource.collection] = wsgi_resource
             kargs = dict(
                 controller=wsgi_resource,
                 collection=resource.collection_actions,
-                member=resource.member_actions)
+                member=resource.member_actions,
+            )
 
             if resource.parent:
                 kargs['parent_resource'] = resource.parent
@@ -118,16 +114,19 @@ def _setup_extensions(self, ext_mgr):
             controller = extension.controller
 
             if collection not in self.resources:
-                LOG.warning(_LW('Extension %(ext_name)s: Cannot extend '
-                                'resource %(collection)s: No such resource'),
-                            {'ext_name': extension.extension.name,
-                             'collection': collection})
+                LOG.warning(
+                    _LW(
+                        'Extension %(ext_name)s: Cannot extend '
+                        'resource %(collection)s: No such resource'
+                    ),
+                    {'ext_name': extension.extension.name, 'collection': collection},
+                )
                 continue
 
-            LOG.debug('Extension %(ext_name)s extending resource: '
-                      '%(collection)s',
-                      {'ext_name': extension.extension.name,
-                       'collection': collection})
+            LOG.debug(
+                'Extension %(ext_name)s extending resource: %(collection)s',
+                {'ext_name': extension.extension.name, 'collection': collection},
+            )
 
             resource = self.resources[collection]
             resource.register_actions(controller)

coriolis/api/common.py

@@ -12,9 +12,9 @@ def get_paging_params(req):
     return marker, limit
 
 
-def get_sort_params(req,
-                    default_keys=('created_at', 'id'),
-                    default_dirs=('desc', 'desc')):
+def get_sort_params(
+    req, default_keys=('created_at', 'id'), default_dirs=('desc', 'desc')
+):
     """Retrieves sort keys/directions parameters.
 
     Processes the parameters to create a list of sort keys and sort directions

coriolis/api/middleware/auth.py

@@ -1,13 +1,13 @@
 # Copyright 2016 Cloudbase Solutions Srl
 # All Rights Reserved.
 
+import webob
 from oslo_log import log as logging
 from oslo_middleware import request_id
 from oslo_serialization import jsonutils
-import webob
 
-from coriolis.api import wsgi
 from coriolis import context
+from coriolis.api import wsgi
 from coriolis.i18n import _
 
 LOG = logging.getLogger(__name__)
@@ -23,14 +23,16 @@ def _get_project_id(self, req):
             return req.headers['X_TENANT']
         else:
             raise webob.exc.HTTPBadRequest(
-                explanation=_("No 'X_TENANT_ID' or 'X_TENANT' passed."))
+                explanation=_("No 'X_TENANT_ID' or 'X_TENANT' passed.")
+            )
 
     def _get_user(self, req):
         user = req.headers.get('X_USER')
         user = req.headers.get('X_USER_ID', user)
         if user is None:
             raise webob.exc.HTTPUnauthorized(
-                explanation=_("Neither X_USER_ID nor X_USER found in request"))
+                explanation=_("Neither X_USER_ID nor X_USER found in request")
+            )
         return user
 
     @webob.dec.wsgify(RequestClass=wsgi.Request)
@@ -64,18 +66,21 @@ def __call__(self, req):
                 service_catalog = jsonutils.loads(catalog_header)
             except ValueError:
                 raise webob.exc.HTTPInternalServerError(
-                    explanation=_('Invalid service catalog json.'))
-
-        ctx = context.RequestContext(user,
-                                     project_id,
-                                     project_name=project_name,
-                                     project_domain_name=project_domain_name,
-                                     user_domain_name=user_domain_name,
-                                     roles=roles,
-                                     auth_token=auth_token,
-                                     remote_address=remote_address,
-                                     service_catalog=service_catalog,
-                                     request_id=req_id)
+                    explanation=_('Invalid service catalog json.')
+                )
+
+        ctx = context.RequestContext(
+            user,
+            project_id,
+            project_name=project_name,
+            project_domain_name=project_domain_name,
+            user_domain_name=user_domain_name,
+            roles=roles,
+            auth_token=auth_token,
+            remote_address=remote_address,
+            service_catalog=service_catalog,
+            request_id=req_id,
+        )
 
         req.environ['coriolis.context'] = ctx
         return self.application

coriolis/api/middleware/fault.py

@@ -14,16 +14,14 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from oslo_log import log as logging
 import six
 import webob.dec
 import webob.exc
+from oslo_log import log as logging
 
+from coriolis import exception, utils
 from coriolis.api import wsgi
-from coriolis import exception
-from coriolis.i18n import _, _LE, _LI  # noqa
-from coriolis import utils
-
+from coriolis.i18n import _LE, _LI, _  # noqa
 
 LOG = logging.getLogger(__name__)
 
@@ -39,12 +37,14 @@ def status_to_type(status):
             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)()
+            status, webob.exc.HTTPInternalServerError
+        )()
 
     def _error(self, inner, req):
-        LOG.exception(_LE("Caught error: %(type)s %(error)s"),
-                      {'type': type(inner),
-                       'error': inner})
+        LOG.exception(
+            _LE("Caught error: %(type)s %(error)s"),
+            {'type': type(inner), 'error': inner},
+        )
         safe = getattr(inner, 'safe', False)
         headers = getattr(inner, 'headers', None)
         status = getattr(inner, 'code', 500)
@@ -57,10 +57,12 @@ def _error(self, inner, req):
         if headers:
             outer.headers = headers
         if safe:
-            msg = (inner.msg if isinstance(inner, exception.CoriolisException)
-                   else six.text_type(inner))
-            params = {'exception': inner.__class__.__name__,
-                      'explanation': msg}
+            msg = (
+                inner.msg
+                if isinstance(inner, exception.CoriolisException)
+                else six.text_type(inner)
+            )
+            params = {'exception': inner.__class__.__name__, 'explanation': msg}
             outer.explanation = _('%(exception)s: %(explanation)s') % params
         return wsgi.Fault(outer)
 

coriolis/api/v1/deployment_actions.py

@@ -3,9 +3,9 @@
 
 from webob import exc
 
+from coriolis import exception
 from coriolis.api import wsgi as api_wsgi
 from coriolis.deployments import api
-from coriolis import exception
 from coriolis.policies import deployments as deployment_policies
 
 

coriolis/api/v1/deployments.py

@@ -4,16 +4,15 @@
 from oslo_log import log as logging
 from webob import exc
 
+from coriolis import exception
 from coriolis.api import common
+from coriolis.api import wsgi as api_wsgi
 from coriolis.api.v1 import utils as api_utils
 from coriolis.api.v1.views import deployment_view
-from coriolis.api import wsgi as api_wsgi
 from coriolis.deployments import api
 from coriolis.endpoints import api as endpoints_api
-from coriolis import exception
 from coriolis.policies import deployments as deployment_policies
 
-
 LOG = logging.getLogger(__name__)
 
 
@@ -27,23 +26,24 @@ def show(self, req, id):
         context = req.environ["coriolis.context"]
         context.can(deployment_policies.get_deployments_policy_label("show"))
         include_task_info = api_utils.get_bool_url_arg(
-            req, "include_task_info", default=False)
+            req, "include_task_info", default=False
+        )
         deployment = self._deployment_api.get_deployment(
-            context, id,
-            include_task_info=include_task_info)
+            context, id, include_task_info=include_task_info
+        )
         if not deployment:
             raise exc.HTTPNotFound()
 
         return deployment_view.single(deployment)
 
     def _list(self, req):
-        show_deleted = api_utils.get_bool_url_arg(
-            req, "show_deleted", default=False)
+        show_deleted = api_utils.get_bool_url_arg(req, "show_deleted", default=False)
         context = req.environ["coriolis.context"]
         context.show_deleted = show_deleted
         context.can(deployment_policies.get_deployments_policy_label("list"))
         include_task_info = api_utils.get_bool_url_arg(
-            req, "include_task_info", default=False)
+            req, "include_task_info", default=False
+        )
 
         marker, limit = common.get_paging_params(req)
         sort_keys, sort_dirs = common.get_sort_params(req)
@@ -53,9 +53,12 @@ def _list(self, req):
                 context,
                 include_tasks=include_task_info,
                 include_task_info=include_task_info,
-                marker=marker, limit=limit,
-                sort_keys=sort_keys, sort_dirs=sort_dirs,
-            ))
+                marker=marker,
+                limit=limit,
+                sort_keys=sort_keys,
+                sort_dirs=sort_dirs,
+            )
+        )
 
     def index(self, req):
         return self._list(req)
@@ -72,36 +75,51 @@ def _validate_deployment_input(self, context, body):
         if not transfer_id:
             raise exc.HTTPBadRequest(
                 explanation="Missing 'transfer_id' field from deployment "
-                            "body. A deployment can be created strictly "
-                            "based on an existing Transfer.")
+                "body. A deployment can be created strictly "
+                "based on an existing Transfer."
+            )
 
         clone_disks = deployment.get("clone_disks", True)
         force = deployment.get("force", False)
         skip_os_morphing = deployment.get("skip_os_morphing", False)
         instance_osmorphing_minion_pool_mappings = deployment.get(
-            'instance_osmorphing_minion_pool_mappings', {})
+            'instance_osmorphing_minion_pool_mappings', {}
+        )
         user_scripts = deployment.get('user_scripts', {})
         api_utils.validate_user_scripts(user_scripts)
         return (
-            transfer_id, force, clone_disks, skip_os_morphing,
+            transfer_id,
+            force,
+            clone_disks,
+            skip_os_morphing,
             instance_osmorphing_minion_pool_mappings,
-            user_scripts)
+            user_scripts,
+        )
 
     def create(self, req, body):
         context = req.environ['coriolis.context']
         context.can(deployment_policies.get_deployments_policy_label("create"))
 
-        (transfer_id, force, clone_disks, skip_os_morphing,
-         instance_osmorphing_minion_pool_mappings,
-         user_scripts) = self._validate_deployment_input(
-            context, body)
+        (
+            transfer_id,
+            force,
+            clone_disks,
+            skip_os_morphing,
+            instance_osmorphing_minion_pool_mappings,
+            user_scripts,
+        ) = self._validate_deployment_input(context, body)
 
         # NOTE: destination environment for transfer should have been
         # validated upon its creation.
         deployment = self._deployment_api.deploy_transfer_instances(
-            context, transfer_id, instance_osmorphing_minion_pool_mappings,
-            clone_disks, force, skip_os_morphing,
-            user_scripts=user_scripts)
+            context,
+            transfer_id,
+            instance_osmorphing_minion_pool_mappings,
+            clone_disks,
+            force,
+            skip_os_morphing,
+            user_scripts=user_scripts,
+        )
 
         return deployment_view.single(deployment)