Skip to content

Commit

Permalink
Appease ruff+mypy
Browse files Browse the repository at this point in the history
  • Loading branch information
nx10 committed Jul 18, 2024
1 parent b74b40b commit 8c201d2
Show file tree
Hide file tree
Showing 4 changed files with 63 additions and 21 deletions.
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,4 +21,4 @@ set_global_runner(GraphRunner(get_global_runner())) # Use GraphRunner middlewar

print(get_global_runner().mermaid()) # Print mermaid diagram

```
```
15 changes: 13 additions & 2 deletions poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ packages = [{include = "styxgraph", from = "src"}]

[tool.poetry.dependencies]
python = "~3.11"
styxdefs = "^0.1.5"

[tool.poetry.group.dev.dependencies]
pytest = "^8.2.2"
Expand Down
66 changes: 48 additions & 18 deletions src/styxgraph/__init__.py
Original file line number Diff line number Diff line change
@@ -1,48 +1,76 @@
""".. include:: ../../README.md""" # noqa: D415

import pathlib
from styxdefs import set_global_runner, get_global_runner, Runner, Execution, Metadata, InputPathType, OutputPathType

from styxdefs import (
Execution,
InputPathType,
Metadata,
OutputPathType,
Runner,
)


class _GraphExecution(Execution):
def __init__(self, base: Execution, graph_runner: 'GraphRunner', metadata: Metadata) -> None:
"""Graph execution."""

def __init__(
self, base: Execution, graph_runner: "GraphRunner", metadata: Metadata
) -> None:
"""Create a new GraphExecution."""
self.base = base
self.graph_runner = graph_runner
self.metadata = metadata
self.input_files = []
self.output_files = []
self.input_files: list[InputPathType] = []
self.output_files: list[OutputPathType] = []

def input_file(self, host_file: InputPathType) -> str:
"""Resolve input file."""
self.input_files.append(host_file)
return self.base.input_file(host_file)

def run(self, cargs: list[str]) -> None:
self.graph_runner.graph_append(self.metadata, self.input_files, self.output_files)
"""Run the command."""
self.graph_runner.graph_append(
self.metadata, self.input_files, self.output_files
)
return self.base.run(cargs)

def output_file(self, local_file: str, optional: bool = False) -> OutputPathType:
"""Resolve output file."""
output_file = self.base.output_file(local_file, optional)
self.output_files.append(output_file)
return output_file


# Define a new runner
class GraphRunner(Runner):
def __init__(self, base: Runner):
"""Graph runner."""

def __init__(self, base: Runner) -> None:
"""Create a new GraphRunner."""
self.base = base
self.graph: list[tuple[str, list[InputPathType], list[OutputPathType]]] = []

def start_execution(self, metadata: Metadata) -> Execution:
"""Start a new execution."""
return _GraphExecution(self.base.start_execution(metadata), self, metadata)

def graph_append(self, metadata: Metadata, input_file: list[InputPathType], output_file: list[OutputPathType]) -> None:
print(f'Appending {metadata.name} with {input_file} and {output_file}')

def graph_append(
self,
metadata: Metadata,
input_file: list[InputPathType],
output_file: list[OutputPathType],
) -> None:
"""Append a node to the graph."""
print(f"Appending {metadata.name} with {input_file} and {output_file}")
self.graph.append((metadata.name, input_file, output_file))

def mermaid(self) -> str:
connections = []
inputs_lookup = {}
outputs_lookup = {}
"""Generate a mermaid graph of the graph."""
connections: list[str] = []
inputs_lookup: dict[str, list[str]] = {}
outputs_lookup: dict[str, str] = {}
for id, inputs, outputs in self.graph:
for input in inputs:
if input not in inputs_lookup:
Expand All @@ -54,13 +82,15 @@ def mermaid(self) -> str:
for id, inputs, _ in self.graph:
for input in inputs:
for output, output_id in outputs_lookup.items():
if pathlib.Path(input).is_relative_to(output): # is subfolder/file in output root
connections.append(f'{output_id} --> {id}')

if pathlib.Path(input).is_relative_to(
output
): # is subfolder/file in output root
connections.append(f"{output_id} --> {id}")

# Generate mermaid
mermaid = 'graph TD\n'
mermaid = "graph TD\n"
for id, _, _ in self.graph:
mermaid += f' {id}\n'
mermaid += f" {id}\n"
for connection in connections:
mermaid += f' {connection}\n'
mermaid += f" {connection}\n"
return mermaid

0 comments on commit 8c201d2

Please sign in to comment.