resolve ruff linting errors

gso/services/ipam.py

@@ -4,6 +4,7 @@ from typing import Union
 from pydantic import BaseSettings
 
 from gso.services import _ipam
+from gso.services._ipam import V4HostAddress, V6HostAddress
 
 
 class V4ServiceNetwork(BaseSettings):
@@ -19,20 +20,14 @@ class ServiceNetworks(BaseSettings):
     v6: ipaddress.IPv6Network
 
 
-class V4HostAddress(BaseSettings):
-    v4: ipaddress.IPv4Address
-
-
-class V6HostAddress(BaseSettings):
-    v6: ipaddress.IPv6Address
-
-
 class HostAddresses(BaseSettings):
     v4: ipaddress.IPv4Address
     v6: ipaddress.IPv6Address
 
 
-def new_service_networks(service_type="", comment="", extattrs={}) -> ServiceNetworks:
+def new_service_networks(service_type="", comment="", extattrs=None) -> ServiceNetworks:
+    if extattrs is None:
+        extattrs = {}
     v4_service_network = _ipam.allocate_service_ipv4_network(
         service_type=service_type, comment=comment, extattrs=extattrs
     )
@@ -48,8 +43,10 @@ def new_service_host(
     service_networks: ServiceNetworks = None,
     host_addresses: HostAddresses = None,
     cname_aliases=None,
-    extattrs={},
+    extattrs=None,
 ) -> HostAddresses:
+    if extattrs is None:
+        extattrs = {}
     return _ipam.allocate_service_host(
         hostname=hostname,
         service_type=service_type,
@@ -67,8 +64,10 @@ def delete_service_network(
 
 
 def delete_service_host(
-    hostname="", host_addresses: HostAddresses = None, cname_aliases=[], service_type=""
-) -> HostAddresses:
+    hostname="", host_addresses: HostAddresses = None, cname_aliases=None, service_type=""
+) -> V4HostAddress | V6HostAddress:
+    if cname_aliases is None:
+        cname_aliases = []
     return _ipam.delete_service_host(
         hostname=hostname, host_addresses=host_addresses, cname_aliases=cname_aliases, service_type=service_type
     )

gso/services/provisioning_proxy.py

 """The Provisioning Proxy service, which interacts with LSO running externally.
+
 LSO is responsible for executing Ansible playbooks, that deploy subscriptions.
 """
 import json
@@ -8,8 +9,6 @@ import requests
 from orchestrator import inputstep
 from orchestrator.config.assignee import Assignee
 from orchestrator.domain import SubscriptionModel
-
-# noinspection PyProtectedMember
 from orchestrator.forms import FormPage, ReadOnlyField
 from orchestrator.forms.validators import Accept, Label, LongText
 from orchestrator.types import State, UUIDstr, strEnum
@@ -25,6 +24,7 @@ logger = logging.getLogger(__name__)
 
 class CUDOperation(strEnum):
     """Enum for different C(R)UD operations that the provisioning proxy supports.
+
     Read is not applicable, hence these become CUD and not CRUD operations.
     """
 
@@ -37,8 +37,7 @@ class CUDOperation(strEnum):
 
 
 def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr, operation: CUDOperation):
-    """Internal function for sending a request to LSO. The callback address is
-    derived using the process ID provided.
+    """Send a request to LSO. The callback address is derived using the process ID provided.
 
     :param str endpoint: The LSO-specific endpoint to call, depending on the
         type of service object that is acted upon.
@@ -70,12 +69,12 @@ def _send_request(endpoint: str, parameters: dict, process_id: UUIDstr, operatio
         request = requests.delete(url, json=parameters, timeout=10000)
 
     if request.status_code != 200:
-        print(request.content)
+        logger.debug(request.content)
         raise AssertionError(request.content)
 
 
 def provision_device(subscription: DeviceProvisioning, process_id: UUIDstr, dry_run: bool = True):
-    """Function that provisions a new device using LSO.
+    """Provision a new device using LSO.
 
     :param :class:`DeviceProvisioning` subscription: The subscription object
         that is to be provisioned.
@@ -91,7 +90,7 @@ def provision_device(subscription: DeviceProvisioning, process_id: UUIDstr, dry_
 def provision_ip_trunk(
     subscription: IptrunkProvisioning, process_id: UUIDstr, config_object: str, dry_run: bool = True
 ):
-    """Function that provisions an IP trunk service using LSO.
+    """Provision an IP trunk service using LSO.
 
     :param :class:`IptrunkProvisioning` subscription: The subscription object
         that is to be provisioned.
@@ -136,7 +135,7 @@ def provision_ip_trunk(
 
 
 def deprovision_ip_trunk(subscription: Iptrunk, process_id: UUIDstr, dry_run: bool = True):
-    """Function that provisions an IP trunk service using LSO.
+    """Deprovision an IP trunk service using LSO.
 
     :param :class:`IptrunkProvisioning` subscription: The subscription object
         that is to be provisioned.
@@ -162,7 +161,7 @@ def await_pp_results(subscription: SubscriptionModel, label_text: str) -> State:
         @validator("pp_run_results", allow_reuse=True, pre=True, always=True)
         def run_results_must_be_given(cls, run_results):
             if run_results is None:
-                raise ValueError("Run results may not be empty. " "Wait for the provisioning proxy to finish.")
+                raise ValueError("Run results may not be empty. Wait for the provisioning proxy to finish.")
             return run_results
 
     result_page = yield ProvisioningResultPage

gso/settings.py

 """GSO settings, ensuring that the required parameters are set correctly."""
 import ipaddress
 import json
+import logging
 import os
 
-# from pydantic import BaseSettings, Field
 from pydantic import BaseSettings
 
+logger = logging.getLogger(__name__)
+
 
 class GeneralParams(BaseSettings):
     """General parameters for a GSO configuration file."""
@@ -42,8 +44,9 @@ class V6NetworkParams(BaseSettings):
 
 
 class ServiceNetworkParams(BaseSettings):
-    """Parameters for InfoBlox that describe IPv4 and v6 networks, and the
-    corresponding domain name that should be used as a suffix.
+    """Parameters for InfoBlox.
+
+    The parameters describe IPv4 and v6 networks, and the corresponding domain name that should be used as a suffix.
     """
 
     V4: V4NetworkParams
@@ -82,12 +85,10 @@ class OSSParams(BaseSettings):
 
 
 def load_oss_params() -> OSSParams:
-    """Look for OSS_PARAMS_FILENAME in the environment and load the parameters
-    from that file.
-    """
+    """Look for OSS_PARAMS_FILENAME in the environment and load the parameters from that file."""
     with open(os.environ["OSS_PARAMS_FILENAME"], encoding="utf-8") as file:
         return OSSParams(**json.loads(file.read()))
 
 
 if __name__ == "__main__":
-    print(load_oss_params())
+    logger.debug(load_oss_params())

gso/workflows/device/terminate_device.py

 import ipaddress
+from typing import Type
 
 from orchestrator.forms import FormPage
 from orchestrator.forms.validators import Label
 from orchestrator.targets import Target
-from orchestrator.types import InputForm, SubscriptionLifecycle, UUIDstr
+from orchestrator.types import SubscriptionLifecycle, UUIDstr
 from orchestrator.workflow import done, init, step, workflow
 from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import BaseModel
 
 from gso.products.product_types.device import Device
 from gso.services import ipam
 
 
-def initial_input_form_generator(subscription_id: UUIDstr) -> InputForm:
+def initial_input_form_generator(subscription_id: UUIDstr) -> Type[BaseModel]:
     subscription = Device.from_subscription(subscription_id)
 
     class TerminateForm(FormPage):

pyproject.toml

@@ -17,6 +17,7 @@ exclude = '''
     | \.tox
     | venv
     | gso/migrations
+    | gso/services/_ipam\.py
   )/
 )
 '''
@@ -51,7 +52,8 @@ exclude = [
     "__pycache__",
     "htmlcov",
     "venv",
-    "gso/migrations"
+    "gso/migrations",
+    "_ipam.py"  # TODO: remove
 ]
 ignore = [
     "C417",
@@ -72,7 +74,8 @@ ignore = [
     "N803",
     "N801",
     "N815",
-    "N802"
+    "N802",
+    "S101"
 ]
 line-length = 120
 select = [
@@ -94,7 +97,7 @@ target-version = "py310"
 ban-relative-imports = "all"
 
 [tool.ruff.per-file-ignores]
-"test/*" = ["S101", "B033", "N816", "N802"]
+"test/*" = ["B033", "N816", "N802"]
 
 [tool.ruff.isort]
 known-third-party = ["pydantic", "migrations"]