consistent formatting

Author: hanstrompert

products/product_blocks/sap.py

@@ -12,12 +12,12 @@
 # limitations under the License.
 
 
+from nwastdlib.vlans import VlanRanges
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 from pydantic import computed_field
 
 from products.product_blocks.port import PortBlock, PortBlockInactive, PortBlockProvisioning
-from nwastdlib.vlans import VlanRanges
 
 
 class SAPBlockInactive(ProductBlockModel, product_block_name="SAP"):

products/product_types/node.py

@@ -12,9 +12,9 @@
 # limitations under the License.
 
 
-from pydantic_forms.types import strEnum
 from orchestrator.domain.base import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle
+from pydantic_forms.types import strEnum
 
 from products.product_blocks.node import NodeBlock, NodeBlockInactive, NodeBlockProvisioning
 

products/product_types/nsistp.py

@@ -12,11 +12,11 @@
 # limitations under the License.
 
 
+from nwastdlib.vlans import VlanRanges
 from orchestrator.domain.base import SubscriptionModel
 from orchestrator.types import SubscriptionLifecycle
 
 from products.product_blocks.nsistp import NsistpBlock, NsistpBlockInactive, NsistpBlockProvisioning
-from nwastdlib.vlans import VlanRanges
 
 
 class NsistpInactive(SubscriptionModel, is_base=True):

products/services/netbox/payload/sap.py

@@ -39,11 +39,15 @@ def build_sap_vlans_payload(model: SAPBlockProvisioning, subscription: Subscript
     """
     assert model.ims_id
     name = f"{model.port.node.node_name} {model.port.port_name}"
-    vlan_list = [vlan for vlan_start, vlan_end in model.vlan.to_list_of_tuples() for vlan in range(vlan_start, vlan_end+1)]
+    vlan_list = [
+        vlan for vlan_start, vlan_end in model.vlan.to_list_of_tuples() for vlan in range(vlan_start, vlan_end + 1)
+    ]
     return [netbox.VlanPayload(vid=vlan, group=model.ims_id, name=f"{name} - {vlan}") for vlan in vlan_list]
 
 
-def build_sap_vlan_group_payload(model: SAPBlockProvisioning, subscription: SubscriptionModel) -> netbox.VlanGroupPayload:
+def build_sap_vlan_group_payload(
+    model: SAPBlockProvisioning, subscription: SubscriptionModel
+) -> netbox.VlanGroupPayload:
     """Create and return a Netbox VlanGroupPayload object for a :class:`~products.product_blocks.sap.SAPBlockProvisioning`.
 
     Example payload::
@@ -85,4 +89,4 @@ def build_sap_payload(model: SAPBlockProvisioning, subscription: SubscriptionMod
 
     """
     vlan_payloads = build_sap_vlans_payload(model, subscription)
-    return netbox.VlansPayload(vlans=vlan_payloads)
+    return netbox.VlansPayload(vlans=vlan_payloads)

services/lso_client.py

@@ -15,7 +15,6 @@
 :term:`LSO` is responsible for executing Ansible playbooks, that deploy subscriptions.
 """
 
-from pydantic_forms.types import State
 import json
 import logging
 from os import getenv
@@ -25,9 +24,9 @@
 from orchestrator import step
 from orchestrator.config.assignee import Assignee
 from orchestrator.utils.errors import ProcessFailureError
-from orchestrator.workflow import conditional, Step, StepList, begin, callback_step, inputstep
+from orchestrator.workflow import Step, StepList, begin, callback_step, conditional, inputstep
 from pydantic_forms.core import FormPage
-from pydantic_forms.types import FormGenerator
+from pydantic_forms.types import FormGenerator, State
 from pydantic_forms.validators import LongText
 
 logger = logging.getLogger(__name__)
@@ -157,17 +156,14 @@ def lso_interaction(provisioning_step: Step) -> StepList:
     :rtype: :class:`StepList`
     """
     lso_is_enabled = conditional(lambda _: getenv("LSO_ENABLED") == "True")
-    return (
+    return begin >> lso_is_enabled(
         begin
-        >> lso_is_enabled(
-            begin
-            >> callback_step(
-                name=provisioning_step.name,
-                action_step=provisioning_step,
-                validate_step=_evaluate_results,
-            )
-            >> _show_results
+        >> callback_step(
+            name=provisioning_step.name,
+            action_step=provisioning_step,
+            validate_step=_evaluate_results,
         )
+        >> _show_results
     )
 
 
@@ -188,15 +184,12 @@ def indifferent_lso_interaction(provisioning_step: Step) -> StepList:
     :rtype: :class:`StepList`
     """
     lso_is_enabled = conditional(lambda _: getenv("LSO_ENABLED") == "True")
-    return (
+    return begin >> lso_is_enabled(
         begin
-        >> lso_is_enabled(
-            begin
-            >> callback_step(
-                name=provisioning_step.name,
-                action_step=provisioning_step,
-                validate_step=_ignore_results,
-            )
-            >> _show_results
+        >> callback_step(
+            name=provisioning_step.name,
+            action_step=provisioning_step,
+            validate_step=_ignore_results,
         )
+        >> _show_results
     )

services/netbox.py

@@ -147,8 +147,6 @@ class VlansPayload(NetboxPayload):
     vlans: list[VlanPayload]
 
 
-
-
 @dataclass
 class L2vpnPayload(NetboxPayload):
     name: str
@@ -286,6 +284,7 @@ def delete_l2vpn(**kwargs) -> None:
 def delete_vlan(**kwargs) -> None:
     delete_from_netbox(api.ipam.vlans, **kwargs)
 
+
 def delete_vlan_group(**kwargs) -> None:
     delete_from_netbox(api.ipam.vlan_groups, **kwargs)
 

workflows/core_link/create_core_link.py

@@ -12,22 +12,21 @@
 # limitations under the License.
 
 
-from pydantic_forms.types import UUIDstr
-import uuid
 import json
+import uuid
 from random import randrange
 from typing import TypeAlias, cast
 
 from orchestrator.services.products import get_product_by_id
 from orchestrator.targets import Target
 from orchestrator.types import SubscriptionLifecycle
+from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, begin, step
 from orchestrator.workflows.steps import store_process_subscription
 from orchestrator.workflows.utils import create_workflow
-from orchestrator.utils.json import json_dumps
 from pydantic import ConfigDict, model_validator
 from pydantic_forms.core import FormPage
-from pydantic_forms.types import FormGenerator, State
+from pydantic_forms.types import FormGenerator, State, UUIDstr
 from pydantic_forms.validators import Choice
 
 from products.product_types.core_link import CoreLinkInactive, CoreLinkProvisioning

workflows/core_link/modify_core_link.py

@@ -12,11 +12,10 @@
 # limitations under the License.
 
 
-from pydantic_forms.types import UUIDstr
 from orchestrator.workflow import StepList, begin, step
-from orchestrator.workflows.utils import modify_workflow, ensure_provisioning_status
+from orchestrator.workflows.utils import ensure_provisioning_status, modify_workflow
 from pydantic_forms.core import FormPage
-from pydantic_forms.types import FormGenerator, State
+from pydantic_forms.types import FormGenerator, State, UUIDstr
 
 from products.product_types.core_link import CoreLink, CoreLinkProvisioning
 from products.services.description import description
@@ -58,8 +57,4 @@ def update_core_link_in_nrm(subscription: CoreLink) -> State:
 
 @modify_workflow("Modify core_link", initial_input_form=initial_input_form_generator)
 def modify_core_link() -> StepList:
-    return (
-        begin
-        >> update_subscription
-        >> update_core_link_in_nrm
-    )
+    return begin >> update_subscription >> update_core_link_in_nrm

workflows/core_link/terminate_core_link.py

@@ -11,23 +11,21 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from pydantic_forms.types import UUIDstr
-from pydantic_forms.types import InputForm
 import json
+
+from orchestrator.utils.json import json_dumps
 from orchestrator.workflow import StepList, begin, step
 from orchestrator.workflows.utils import terminate_workflow
 from pydantic_forms.core import FormPage
-from pydantic_forms.types import State
+from pydantic_forms.types import InputForm, State, UUIDstr
 from pydantic_forms.validators import DisplaySubscription
-from orchestrator.utils.json import json_dumps
 
 from products.product_types.core_link import CoreLink
 from products.services.netbox.netbox import build_payload
 from services import netbox
 from services.lso_client import execute_playbook, lso_interaction
 
 
-#def terminate_initial_input_form_generator(subscription_id: UUIDstr, organisation: UUIDstr) -> InputForm:
 def terminate_initial_input_form_generator(subscription_id: UUIDstr) -> InputForm:
     temp_subscription_id = subscription_id
 

workflows/core_link/validate_core_link.py

@@ -12,9 +12,9 @@
 # limitations under the License.
 
 
-from pydantic_forms.types import State
 from orchestrator.workflow import StepList, begin, step
 from orchestrator.workflows.utils import validate_workflow
+from pydantic_forms.types import State
 
 from products.product_types.core_link import CoreLink