[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

paraffin/cli.py

@@ -3,7 +3,7 @@
 import typer
 
 from paraffin.submit import submit_node_graph
-from paraffin.utils import get_stage_graph, get_custom_queue
+from paraffin.utils import get_custom_queue, get_stage_graph
 
 app = typer.Typer()
 
@@ -26,7 +26,11 @@ def main(
 
     subgraph = get_stage_graph(names=names, glob=glob)
     custom_queues = get_custom_queue()
-    submit_node_graph(subgraph, shutdown_after_finished=shutdown_after_finished, custom_queues=custom_queues)
+    submit_node_graph(
+        subgraph,
+        shutdown_after_finished=shutdown_after_finished,
+        custom_queues=custom_queues,
+    )
 
     typer.echo(f"Submitted all (n = {len(subgraph)})  tasks.")
     if concurrency > 0:

paraffin/submit.py

@@ -1,16 +1,31 @@
+import fnmatch
+import typing as t
+
 import networkx as nx
 from celery import chord, group
-import typing as t
+
 from paraffin.worker import repro, shutdown_worker
-import fnmatch
 
 
-def submit_node_graph(subgraph: nx.DiGraph, shutdown_after_finished: bool = False, custom_queues: t.Optional[dict] = None):
+def submit_node_graph(
+    subgraph: nx.DiGraph,
+    shutdown_after_finished: bool = False,
+    custom_queues: t.Optional[dict] = None,
+):
     task_dict = {}
     custom_queues = custom_queues or {}
     for node in subgraph.nodes:
-        if (matched_pattern := next((pattern for pattern in custom_queues if fnmatch.fnmatch(node.name, pattern)), None)):
-            task_dict[node.name] = repro.s(name=node.name).set(queue=custom_queues[matched_pattern])
+        if matched_pattern := next(
+            (
+                pattern
+                for pattern in custom_queues
+                if fnmatch.fnmatch(node.name, pattern)
+            ),
+            None,
+        ):
+            task_dict[node.name] = repro.s(name=node.name).set(
+                queue=custom_queues[matched_pattern]
+            )
         else:
             task_dict[node.name] = repro.s(name=node.name)
 

paraffin/utils.py

@@ -1,8 +1,8 @@
 import fnmatch
+import pathlib
 
 import dvc.api
 import networkx as nx
-import pathlib
 import yaml
 
 
@@ -41,9 +41,12 @@ def get_stage_graph(names, glob=False):
 
     return subgraph
 
+
 def get_custom_queue():
     with pathlib.Path("paraffin.yaml").open() as f:
         config = yaml.safe_load(f)
-    
+
     return config.get("queue", {})
-# TODO: what about lists, shutdown, ?
+
+
+# TODO: what about lists, shutdown, ?