apply linting

examples/pipelines/logistic_stats/pipeline.py

@@ -113,9 +113,9 @@ def model(dhis2_data: dict[str, typing.Any], gadm_data, worldpop_data):
     population_df = pd.Series(data=[stat["sum"] for stat in stats], index=administrative_areas.index)
     population_df = pd.DataFrame({"District": administrative_areas["NAME_2"], "Population": population_df})
     corrected_population_df = population_df.copy()
-    corrected_population_df.loc[
-        corrected_population_df["District"].str.startswith("Western"), "District"
-    ] = "Western Area"
+    corrected_population_df.loc[corrected_population_df["District"].str.startswith("Western"), "District"] = (
+        "Western Area"
+    )
 
     corrected_population_df = corrected_population_df.groupby("District").sum()
 

openhexa/cli/api.py

@@ -530,10 +530,10 @@ def generate_zip_file(pipeline_directory_path: typing.Union[str, Path]) -> io.By
 
 
 def upload_pipeline(
-        pipeline_directory_path: typing.Union[str, Path],
-        name: str = None,
-        description: str = None,
-        link: str = None,
+    pipeline_directory_path: typing.Union[str, Path],
+    name: str = None,
+    description: str = None,
+    link: str = None,
 ):
     """Upload the pipeline contained in the provided directory using the GraphQL API.
 

openhexa/sdk/pipelines/log_level.py

@@ -1,4 +1,5 @@
 """Log levels for the pipeline runs."""
+
 from enum import IntEnum
 
 

openhexa/sdk/pipelines/parameter.py

@@ -433,7 +433,6 @@ def validate(self, value: typing.Any) -> typing.Any:
         else:
             return self._validate_single(value)
 
-
     def to_dict(self) -> dict[str, typing.Any]:
         """Return a dictionary representation of the Parameter instance."""
         return {
@@ -518,12 +517,16 @@ def _validate_default(self, default: typing.Any, multiple: bool):
                     f"The default value for {self.code} is not included in the provided choices."
                 )
 
-def validate_connection_parameters(parameters : [Parameter]):
-    """Validate the provided connection parameters."""
+
+def validate_parameters_with_connection(parameters: [Parameter]):
+    """Validate the provided connection parameters if they relate to existing connection parameter."""
     for parameter in parameters:
         if parameter.connection is not None:
             if not any(p.code == parameter.connection for p in parameters):
-                raise InvalidParameterError(f"Connection parameter {parameter.code} references a non-existing parameter {parameter.connection}")
+                raise InvalidParameterError(
+                    f"Connection parameter {parameter.code} references a non-existing parameter {parameter.connection}"
+                )
+
 
 def parameter(
     code: str,

openhexa/sdk/pipelines/runtime.py

@@ -15,7 +15,7 @@
 import requests
 
 from openhexa.sdk.pipelines.exceptions import PipelineNotFound
-from openhexa.sdk.pipelines.parameter import TYPES_BY_PYTHON_TYPE, Parameter, validate_connection_parameters
+from openhexa.sdk.pipelines.parameter import TYPES_BY_PYTHON_TYPE, Parameter, validate_parameters_with_connection
 
 from .pipeline import Pipeline
 
@@ -161,7 +161,7 @@ def get_pipeline(pipeline_path: Path) -> Pipeline:
                 parameter = Parameter(type=type_class.expected_type, **parameter_args)
                 pipelines_parameters.append(parameter)
 
-            validate_connection_parameters(pipelines_parameters)
+            validate_parameters_with_connection(pipelines_parameters)
 
             pipeline = Pipeline(parameters=pipelines_parameters, function=None, **pipeline_decorator_spec["args"])
 

tests/test_ast.py

@@ -510,4 +510,4 @@ def test_pipeline_wit_wrong_connection_parameter(self):
                     )
                 )
             with self.assertRaises(InvalidParameterError):
-                pipeline = get_pipeline(tmpdirname)
+                pipeline = get_pipeline(tmpdirname)

tests/test_parameter.py

@@ -316,7 +316,6 @@ def test_parameter_validate_multiple():
         parameter_4.validate(["ab", "xy"])
 
 
-
 def test_parameter_decorator():
     """Ensure that the @parameter decorator behaves as expected (options and defaults)."""