lint

Author: ninthreezy

datatorch/agent/client.py

@@ -78,9 +78,7 @@ def agent_steps(self):
             }
         """ % STEP_DISPATCH_FIELDS)
         # fmt: on
-        return cast(
-            AsyncGenerator[AgentStepRequest, None], self.session.subscribe(sub)
-        )
+        return cast(AsyncGenerator[AgentStepRequest, None], self.session.subscribe(sub))
 
     def agent_step_cancels(self):
         """Subscribe to stop signals for steps this agent is running.
@@ -156,9 +154,7 @@ async def complete_step(
             step_input["renderedInput"] = rendered_input
         if error_message is not None:
             step_input["errorMessage"] = error_message
-        result = await self.execute(
-            mutate, params={"id": step_id, "input": step_input}
-        )
+        result = await self.execute(mutate, params={"id": step_id, "input": step_input})
         return bool(result.get("completed"))
 
     def initial_metrics(self, metrics):

datatorch/agent/pipelines/resolver.py

@@ -20,6 +20,7 @@
 raises rather than passing a literal ``${{ }}`` to the action. A yaml that
 uses job dependencies is a server-executed workflow.
 """
+
 import json
 import re
 from typing import Any, Dict, List, Set, Tuple
@@ -29,7 +30,9 @@
 # The name group is ``[^}]+?`` (not ``.+?``) so it can't backtrack across a
 # ``}}`` boundary — a string that both starts and ends with a ref must not be
 # mistaken for one spanning ref.
-STEP_REF = r"\$\{\{\s*steps\.(?P<stepName>[^}]+?)\.outputs?\.(?P<stepKey>[\w.\-]+)\s*\}\}"
+STEP_REF = (
+    r"\$\{\{\s*steps\.(?P<stepName>[^}]+?)\.outputs?\.(?P<stepKey>[\w.\-]+)\s*\}\}"
+)
 INPUT_REF = r"\$\{\{\s*inputs?\.(?P<inputKey>[\w.\-]+)\s*\}\}"
 JOB_REF = r"\$\{\{\s*jobs\.(?P<jobName>[^}]+?)\.outputs?\.(?P<jobKey>[\w.\-]+)\s*\}\}"
 
@@ -123,9 +126,7 @@ def replace(match: "re.Match") -> str:
         ]
     if isinstance(value, dict):
         return {
-            k: _resolve(
-                v, outputs_by_name, trigger_input, job_outputs_by_name, misses
-            )
+            k: _resolve(v, outputs_by_name, trigger_input, job_outputs_by_name, misses)
             for k, v in value.items()
         }
     return value

datatorch/agent/pipelines/step/step.py

@@ -25,9 +25,7 @@ class Step(object):
     """
 
     @classmethod
-    def from_dispatch(
-        cls, dispatch: "AgentStepDispatch", api: "AgentApiClient" = None
-    ):
+    def from_dispatch(cls, dispatch: "AgentStepDispatch", api: "AgentApiClient" = None):
         return cls(
             id=dispatch.get("stepId"),
             action=dispatch.get("action", ""),

datatorch/agent/pipelines/template.py

@@ -17,9 +17,7 @@
 # data (trigger payloads, fetched step outputs); splicing them into a shell
 # string is command injection, so shell/cmd runners must read them from
 # $INPUT_<NAME> env vars instead.
-_INPUT_REF_IN_TEMPLATE = re.compile(
-    r"\$\{[{%][^}]*?(?<![.\w])(?:input|variable)(?!\w)"
-)
+_INPUT_REF_IN_TEMPLATE = re.compile(r"\$\{[{%][^}]*?(?<![.\w])(?:input|variable)(?!\w)")
 
 
 class InputInjectionError(Exception):
@@ -146,9 +144,7 @@ def render_command(self, string: str):
                 "policy; see docs/Pipelines.md.)"
             )
         context = {
-            k: v
-            for k, v in self.variables.items()
-            if k not in ("input", "variable")
+            k: v for k, v in self.variables.items() if k not in ("input", "variable")
         }
         return self._template(string).render({**global_variables, **context})
 

test/agents/pipelines/test_local_pipeline.py

@@ -4,6 +4,7 @@
 the same dataflow the server orchestrator drives in production, exercised
 through the cmd runner's ``::key::value`` output protocol.
 """
+
 import asyncio
 import os
 import unittest

test/agents/pipelines/test_resolver.py

@@ -64,9 +64,7 @@ def test_recurses_into_arrays_and_objects(self):
                 "nested": {"deep": {"v": "${{ steps.Fetch.output.ok }}"}},
             }
         )
-        self.assertEqual(
-            value, {"list": [42, "x"], "nested": {"deep": {"v": True}}}
-        )
+        self.assertEqual(value, {"list": [42, "x"], "nested": {"deep": {"v": True}}})
 
     def test_collects_unresolved_and_leaves_untouched(self):
         value, unresolved = resolve(
@@ -117,9 +115,7 @@ def test_job_output_ref_resolves_when_supplied(self):
             {},
             {"build": {"artifact": "app.tar"}},
         )
-        self.assertEqual(
-            value, {"raw": "app.tar", "msg": "built app.tar"}
-        )
+        self.assertEqual(value, {"raw": "app.tar", "msg": "built app.tar"})
         self.assertEqual(unresolved, [])
 
     def test_job_outputs_alias(self):
@@ -131,8 +127,6 @@ def test_job_outputs_alias(self):
     def test_job_output_unresolvable_in_local_mode(self):
         # Local mode supplies no job outputs (single-job execution), so a
         # jobs.* reference is a strict miss rather than a silent pass-through.
-        value, unresolved = resolve(
-            {"x": "${{ jobs.build.outputs.artifact }}"}
-        )
+        value, unresolved = resolve({"x": "${{ jobs.build.outputs.artifact }}"})
         self.assertEqual(value["x"], "${{ jobs.build.outputs.artifact }}")
         self.assertEqual(unresolved, ["${{ jobs.build.outputs.artifact }}"])

test/agents/pipelines/test_runner_cancel.py

@@ -4,6 +4,7 @@
 the runner must kill its child process instead of orphaning it, and let the
 CancelledError propagate so the step aborts.
 """
+
 import asyncio
 
 import pytest

test/agents/pipelines/test_template.py

@@ -34,9 +34,7 @@ def test_variable_is_an_alias_of_input(self):
     def test_string_inputs_render_machine_local_refs(self):
         v = Variables()
         v.add_input("path", "${{ directory.temp }}/x")
-        self.assertEqual(
-            v.inputs["path"], global_variables["directory"]["temp"] + "/x"
-        )
+        self.assertEqual(v.inputs["path"], global_variables["directory"]["temp"] + "/x")
 
     def test_retired_namespaces_raise(self):
         # job/run/pipeline/trigger/event are server-side concepts now; the