Format code with ruff

Author: fabi200123

coriolis/api-refs/source/conf.py

@@ -1,30 +1,26 @@
+extensions = ["openstackdocstheme", "os_api_ref"]
 
-extensions = [
-    'openstackdocstheme',
-    'os_api_ref'
-]
-
-source_suffix = '.rst'
+source_suffix = ".rst"
 
-master_doc = 'index'
+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'
-bug_tag = 'api-ref'
+repository_name = "cloudbase/coriolis"
+bug_project = "coriolis"
+bug_tag = "api-ref"
 
-pygments_style = 'sphinx'
+pygments_style = "sphinx"
 
-html_theme = 'openstackdocs'
+html_theme = "openstackdocs"
 
 html_theme_options = {
     "sidebar_mode": "toc",
 }
 
-html_last_updated_fmt = '%Y-%m-%d %H:%M'
+html_last_updated_fmt = "%Y-%m-%d %H:%M"
 
 latex_documents = [
-    ('index'),
+    ("index"),
 ]

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__)
 
@@ -38,7 +36,7 @@ def root_app_factory(loader, global_conf, **local_conf):
 
 class APIMapper(routes.Mapper):
     def routematch(self, url=None, environ=None):
-        if url is "":
+        if url == "":
             result = self._match("", environ)
             return result[0], result[1]
         return routes.Mapper.routematch(self, url, environ)
@@ -47,30 +45,30 @@ def connect(self, *args, **kwargs):
         # 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.
-        kwargs.setdefault('requirements', {})
-        if not kwargs['requirements'].get('format'):
-            kwargs['requirements']['format'] = 'json|xml'
+        kwargs.setdefault("requirements", {})
+        if not kwargs["requirements"].get("format"):
+            kwargs["requirements"]["format"] = "json|xml"
         return routes.Mapper.connect(self, *args, **kwargs)
 
 
 class ProjectMapper(APIMapper):
     def resource(self, member_name, collection_name, **kwargs):
-        if 'parent_resource' not in kwargs:
-            kwargs['path_prefix'] = '{project_id}/'
+        if "parent_resource" not in kwargs:
+            kwargs["path_prefix"] = "{project_id}/"
         else:
-            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)
+            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)
 
 
 class APIRouter(base_wsgi.Router):
     """Routes requests on the API to the appropriate controller and method."""
+
     ExtensionManager = None  # override in subclasses
 
     @classmethod
@@ -83,7 +81,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,18 +93,18 @@ 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
+                kargs["parent_resource"] = resource.parent
 
             mapper.resource(resource.collection, resource.collection, **kargs)
 
@@ -118,16 +117,25 @@ 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
@@ -34,10 +34,10 @@ def get_sort_params(req,
     params = req.params.copy()
     sort_keys = []
     sort_dirs = []
-    while 'sort_key' in params:
-        sort_keys.append(params.pop('sort_key').strip())
-    while 'sort_dir' in params:
-        sort_dirs.append(params.pop('sort_dir').strip())
+    while "sort_key" in params:
+        sort_keys.append(params.pop("sort_key").strip())
+    while "sort_dir" in params:
+        sort_dirs.append(params.pop("sort_dir").strip())
     if len(sort_keys) == 0 and default_keys:
         sort_keys.extend(default_keys)
     if len(sort_dirs) == 0 and default_dirs:

coriolis/api/middleware/auth.py

@@ -1,36 +1,38 @@
 # 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__)
 
 
 class CoriolisKeystoneContext(wsgi.Middleware):
     def _get_project_id(self, req):
-        if 'X_TENANT_ID' in req.headers:
+        if "X_TENANT_ID" in req.headers:
             # This is the new header since Keystone went to ID/Name
-            return req.headers['X_TENANT_ID']
-        elif 'X_TENANT' in req.headers:
+            return req.headers["X_TENANT_ID"]
+        elif "X_TENANT" in req.headers:
             # This is for legacy compatibility
-            return req.headers['X_TENANT']
+            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)
+        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)
@@ -40,42 +42,45 @@ def __call__(self, req):
         project_id = self._get_project_id(req)
 
         # get the roles
-        roles = [r.strip() for r in req.headers.get('X_ROLE', '').split(',')]
+        roles = [r.strip() for r in req.headers.get("X_ROLE", "").split(",")]
 
-        project_name = req.headers.get('X_TENANT_NAME')
-        project_domain_name = req.headers.get('X-Project-Domain-Name')
-        user_domain_name = req.headers.get('X-User-Domain-Name')
+        project_name = req.headers.get("X_TENANT_NAME")
+        project_domain_name = req.headers.get("X-Project-Domain-Name")
+        user_domain_name = req.headers.get("X-User-Domain-Name")
 
         req_id = req.environ.get(request_id.ENV_REQUEST_ID)
         # TODO(alexpilotti): Check why it's not str
         if isinstance(req_id, bytes):
             req_id = req_id.decode()
 
         # Get the auth token
-        auth_token = req.headers.get('X_AUTH_TOKEN')
+        auth_token = req.headers.get("X_AUTH_TOKEN")
 
         # Build a context, including the auth_token...
         remote_address = req.remote_addr
 
         service_catalog = None
-        if req.headers.get('X_SERVICE_CATALOG') is not None:
+        if req.headers.get("X_SERVICE_CATALOG") is not None:
             try:
-                catalog_header = req.headers.get('X_SERVICE_CATALOG')
+                catalog_header = req.headers.get("X_SERVICE_CATALOG")
                 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)
-
-        req.environ['coriolis.context'] = ctx
+                    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,15 +37,17 @@ 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})
-        safe = getattr(inner, 'safe', False)
-        headers = getattr(inner, 'headers', None)
-        status = getattr(inner, 'code', 500)
+        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)
         if status is None:
             status = 500
 
@@ -57,11 +57,16 @@ 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}
-            outer.explanation = _('%(exception)s: %(explanation)s') % params
+            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)
 
     @webob.dec.wsgify(RequestClass=wsgi.Request)