Chore/formatting

.github/workflows/github-python-workflow.yml

@@ -3,9 +3,9 @@ name: Python Workflow
 
 on:
   push:
-    branches: [main, develop]
+    branches: [ main, develop ]
   pull_request:
-    types: [opened, synchronize, reopened]
+    types: [ opened, synchronize, reopened ]
 
 jobs:
   lint:
@@ -30,12 +30,10 @@ jobs:
       - name: Run flake8 lint checks
         run: |
           # stop the build if there are Python syntax errors or undefined names
-          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics --output-file=flake8-report.txt
-          # exit-zero treats all errors as warnings
-          flake8 . --count --exit-zero --max-complexity=10 --max-line-length=120 --statistics --output-file=flake8-report.txt
+          flake8 . --count --ignore=E,W,D,I,N806,N815,N818,Q000,Q001,Q002,S001,B008,B028 --show-source --statistics --max-line-length=120 --output-file=flake8-report.txt
 
       - name: Run black format checks
-        run: black --check --diff ogc example > black-report.txt  || true # For now pass even if files need formatting
+        run: black --check --diff -l 120 ogc example > black-report.txt
 
       - name: Artifact flake8 report
         uses: actions/upload-artifact@v4
@@ -86,7 +84,7 @@ jobs:
 
   sonarqube_scan:
     name: SonarQube Scan
-    needs: [unit_testing]
+    needs: [ unit_testing ]
     runs-on: ubuntu-latest
 
     steps:

example/app.py

@@ -115,7 +115,7 @@ def check_layers(layer):
 
     if match_object:
         clean_layer = match_object.group(0)
-        if clean_layer in [l.identifier for l in all_layers]:
+        if clean_layer in [available_layer.identifier for available_layer in all_layers]:
             return "{} is an available layer id".format(clean_layer)
         else:
             return "No layer available with that id"

ogc/core.py

@@ -38,8 +38,9 @@ class OGC(tl.HasTraits):
     def base_url(self):
         return "{}{}?".format(self.server_address, self.endpoint)
 
-    def __init__(self, layers=[], **kwargs):
+    def __init__(self, layers=None, **kwargs):
         super().__init__(**kwargs)
+        layers = layers if layers is not None else []
         coverages = [
             wcs_response_1_0_0.Coverage(
                 layer=layer,

ogc/edr/edr_formatter.py

@@ -17,13 +17,13 @@ def __init__(self, formatter_def: dict):
         super().__init__(formatter_def)
         self.mimetype = formatter_def["mimetype"]
 
-    def write(self, options: dict = {}, data: dict | None = None) -> str:
+    def write(self, options: dict | None = None, data: dict | None = None) -> str:
         """Generate data in the specified format.
 
         Parameters
         ----------
         options : dict, optional
-            Formatting options, by default {}.
+            Formatting options, by default None.
         data : dict | None, optional
             Dictionary representation of the data, by default None.
 

ogc/edr/test/test_edr_routes.py

@@ -12,20 +12,21 @@
 from ogc.edr.edr_routes import EdrRoutes
 
 
-def mock_request(request_args: Dict[str, Any] = {}) -> APIRequest:
+def mock_request(request_args: Dict[str, Any] | None = None) -> APIRequest:
     """Creates a mock request for EDR routes to use.
 
 
     Parameters
     ----------
-    request_args: Dict[str, Any], optional
-        The dictionary for query string arguments.
+    request_args: Dict[str, Any] | None, optional
+        The dictionary for query string arguments, by default None.
 
     Returns
     -------
     APIRequest
         Mock API request for route testing.
     """
+    request_args = request_args if request_args is not None else {}
     environ = create_environ(base_url="http://127.0.0.1:5000/ogc/edr")
     request = Request(environ)
     request.args = ImmutableMultiDict(request_args.items())

ogc/ogc_common.py

@@ -1,6 +1,7 @@
 import logging
 
-import lxml, lxml.etree
+import lxml
+import lxml.etree
 import numpy as np
 import traitlets as tl
 
@@ -102,7 +103,7 @@ def __init__(
         """
         exception_code: 'NoApplicableCode', 'InvalidFormat', 'CoverageNotDefined', 'MissingParameterValue', 'InvalidParameterValue'
         """
-        super(WCSException, self).__init__(exception_text)
+        super().__init__(exception_text, exception_code, locator)
 
         self.exception_text = exception_text
         self.exception_code = exception_code
@@ -123,7 +124,5 @@ def to_xml(self):
         {exception_text}
     </ExceptionText>
 </ExceptionReport>
-""".format(
-            self=self, exception_text=exception_text
-        )
+""".format(self=self, exception_text=exception_text)
         return xml

ogc/servers.py

@@ -98,16 +98,17 @@ def __init__(self, *args, ogcs=None, home_func=None):
             ogcs = []
 
         self.ogcs = ogcs
+
+        def make_method(idx):
+            def method():
+                return self.ogc_render(idx)
+
+            return method
+
         for idx, ogc in enumerate(ogcs):
             endpoint = ogc.endpoint  # should be a string, e.g. "/ogc"
             method_name = "render" + endpoint.replace("/", "_")  # e.g. "render_ogc"
 
-            def make_method(idx):
-                def method():
-                    return self.ogc_render(idx)
-
-                return method
-
             method = make_method(idx)
             setattr(self, method_name, method)
             method = getattr(self, method_name)
@@ -258,13 +259,13 @@ def ogc_render(self, ogc_idx):
             ee = WCSException()
             return respond_xml(ee.to_xml(), status=500)
 
-    def edr_render(self, callable: Callable) -> Callable:
+    def edr_render(self, request_handler: Callable) -> Callable:
         """Function which returns a wrapper for the provided callable.
         Filters arguments and handles any necessary exceptions.
 
         Parameters
         ----------
-        callable : Callable
+        request_handler : Callable
             The callable request handler to be wrapped.
         Returns
         -------
@@ -315,7 +316,7 @@ def wrapper(*args, **kwargs) -> Response:
                 request.args = ImmutableMultiDict(filtered_args)
                 pygeoapi_request = APIRequest.from_flask(request, ["en"])
                 # Build the flask response
-                headers, status, content = callable(pygeoapi_request, *args, **kwargs)
+                headers, status, content = request_handler(pygeoapi_request, *args, **kwargs)
                 response = make_response(content, status)
                 if headers:
                     response.headers = headers
@@ -343,6 +344,6 @@ class FastAPI(object):
     https://mangum.io/asgi-frameworks/
     """
 
-    def __init__(self, *args, ogcs=[]):
+    def __init__(self, *args, ogcs):
         super().__init__(*args)
         raise NotImplementedError

ogc/version.py

@@ -12,13 +12,13 @@
 import os
 from collections import OrderedDict
 
-#########################
-## UPDATE VERSION HERE ##
-#########################
+#######################
+# UPDATE VERSION HERE #
+#######################
 MAJOR = 0
 MINOR = 4
 HOTFIX = 2
-#########################
+#######################
 
 
 VERSION_INFO = OrderedDict([("MAJOR", MAJOR), ("MINOR", MINOR), ("HOTFIX", HOTFIX)])
@@ -64,14 +64,8 @@ def version():
             except Exception:
                 return version_full
 
-        version_full = (
-            subprocess.check_output([git, "describe", "--always"], cwd=CWD)
-            .strip()
-            .decode("ascii")
-        )
-        version_full = version_full.replace("-", "+", 1).replace(
-            "-", "."
-        )  # Make this consistent with PEP440
+        version_full = subprocess.check_output([git, "describe", "--always"], cwd=CWD).strip().decode("ascii")
+        version_full = version_full.replace("-", "+", 1).replace("-", ".")  # Make this consistent with PEP440
 
     except Exception as e:
         print("Could not determine Project version from git repo.\n" + str(e))

ogc/wcs_request_1_0_0.py

@@ -1,6 +1,7 @@
 import logging
 
-import lxml, lxml.etree
+import lxml
+import lxml.etree
 import dateutil.parser
 import traitlets as tl
 
@@ -149,7 +150,7 @@ def validate(self):
 
         assert self.identifier, "WCS Request validation error: no coverage specified"
         assert self.domain_subset_bbox, "WCS Request validation error: no bounding box specified"
-        self.output_format.validate(), "WCS Request validation error: output format"
+        self.output_format.validate()  # WCS Request validation error: output format
         assert self.height, "WCS Request validation error: no height specified"
         assert self.width, "WCS Request validation error: no width specified"
         lons = [
@@ -160,7 +161,7 @@ def validate(self):
             self.domain_subset_bbox.lower_corner[1],
             self.domain_subset_bbox.upper_corner[1],
         ]
-        if any([abs(l) > 361.000 for l in lons]) or any([abs(l) > 91.000 for l in lats]):
+        if any(abs(lon) > 361.000 for lon in lons) or any(abs(lat) > 91.000 for lat in lats):
             raise ogc_common.WCSException(
                 exception_code="InvalidParameterValue",
                 locator="BBOX",

ogc/wcs_response_1_0_0.py

@@ -122,9 +122,7 @@ def coverage_offering(self, coverage):
             xml += """            <wcs:lonLatEnvelope srsName="urn:ogc:def:crs:OGC:1.3:CRS84">
             <gml:pos>{coverage.wgs84_bounding_box_lower_corner_lat_lon[1]} {coverage.wgs84_bounding_box_lower_corner_lat_lon[0]}</gml:pos>
             <gml:pos>{coverage.wgs84_bounding_box_upper_corner_lat_lon[1]} {coverage.wgs84_bounding_box_upper_corner_lat_lon[0]}</gml:pos>
-        </wcs:lonLatEnvelope>""".format(
-                coverage=coverage
-            )
+        </wcs:lonLatEnvelope>""".format(coverage=coverage)
 
         xml += """\
     <wcs:domainSet>
@@ -235,9 +233,7 @@ def service(self):
         <wcs:fees>UNAVAILABLE</wcs:fees>
         <wcs:accessConstraints>{constraints}</wcs:accessConstraints>
     </wcs:Service>
-""".format(
-            self=self, constraints=settings.CONSTRAINTS
-        )
+""".format(self=self, constraints=settings.CONSTRAINTS)
 
     base_url = tl.Unicode(default_value=None, allow_none=True)  # e.g., http://hostname:port/path?
 
@@ -277,9 +273,7 @@ def capability(self):
       <wcs:Format>application/vnd.ogc.se_xml</wcs:Format>
     </wcs:Exception>
   </wcs:Capability>
-  """.format(
-            self=self
-        )
+  """.format(self=self)
 
     coverages = tl.List(tl.Instance(klass=Coverage))  # is populated via Traits in constructor
 
@@ -316,9 +310,7 @@ def contents(self):
                 <gml:pos>{coverage.wgs84_bounding_box_lower_corner_lat_lon[1]} {coverage.wgs84_bounding_box_lower_corner_lat_lon[0]}</gml:pos>
                 <gml:pos>{coverage.wgs84_bounding_box_upper_corner_lat_lon[1]} {coverage.wgs84_bounding_box_upper_corner_lat_lon[0]}</gml:pos>
             </wcs:lonLatEnvelope>
-""".format(
-                    coverage=coverage
-                )
+""".format(coverage=coverage)
             xml += "        </wcs:CoverageOfferingBrief>\n"
         xml += "    </wcs:ContentMetadata>\n"
 

ogc/wms_response_1_3_0.py

@@ -86,9 +86,7 @@ def request(self):
         </DCPType>
       </GetMap>
     </Request>
-  """.format(
-            self=self
-        )
+  """.format(self=self)
 
     def exception(self):
         return """\
@@ -241,11 +239,11 @@ def to_xml(self):
         )
 
     @staticmethod
-    def _format_number(input, float_decimals=9):
-        if isinstance(input, float):
-            return "{number:.{decimals}f}".format(number=input, decimals=float_decimals)
+    def _format_number(input_number, float_decimals=9):
+        if isinstance(input_number, float):
+            return "{number:.{decimals}f}".format(number=input_number, decimals=float_decimals)
         else:
-            return "{}".format(input)
+            return "{}".format(input_number)
 
     def _get_CRS_and_BoundingBox(self, depth=3):
         indent = "    "

pyproject.toml

@@ -63,6 +63,8 @@ dev = [
     "pre_commit",
     "black",
     "flake8",
+    "flake8-bugbear",
+    "flake8-builtins",
 ]
 
 [tool.setuptools.packages.find]