This repository contains a Python package which features:

• Depth First Search implementation of topological sorting for acyclic oriented graphs written in Python 3
• an application of this sorting - managing pipelines of your functions with dependencies: each function invokes automatically all its dependencies

• Python 3.8.x: get at python.org

• Download these files from GitHub or clone this repository via https:

git clone https://github.com/zuevval/topological-sorting.git
cd topological-sorting

• If you wish to run tests and static code checks:
  • install dependencies for tests:

    pip install test/requirements.txt

  • install GraphViz for test visualization. If you're running Windows, make sure to add <graphviz/installation/path>/bin to PATH environmental variable.

If you invoke Python from the root folder of this repository, a package pipeline_manager will be in your Python path.

Otherwise, you need to add this folder to PYTHONPATH, e. g. on Linux

export PYTHONPATH="${PYTHONPATH}:/path/containing/repository/pipeline-management-zuevval"

You may sort oriented acyclic graphs with strings or integers as vertices.

Add imports at the top of your script:

...
from pipeline_manager import Vertex, toposort

where Vertex is a shorthand for Union[str, int]

Each graph must be of type DefaultDict[Vertex, List[Vertex]].

Then call toposort passing a graph and a starting vertex to it:

...
sorted = toposort(v, my_graph)

Example:

from collections import defaultdict

from pipeline_manager import toposort


def main():
    graph = defaultdict(list, {0:[], 1:[0], 2:[0, 1]})
    print(toposort(2, graph))

if __name__ == '__main__':
    main()

Consider a pipeline - a set of steps where each one is a function invocation. A step may depend on other steps. You may build a pipeline using a decorator which allows passing an optional argument - names of methods on which this step depends:

from pipeline_manager import pipeline

register = pipeline()

@register
def step1():
    print("Step 1")

@register(depends_on=["step1"])
def step2():
    print("Step 2")

step1()
"Step 1"

When a method decorated with @register (or another name that you give to a result of the pipeline()) is called, all its dependencies are called, too. The order of invocation follows the inverse topological sort of the graph where each vertex is a step and each edge is a dependency, therefore, none of the steps are executed before their dependencies and all dependencies for every step are executed.

step2()
"Step 1"
"Step 2"

step1()
"Step 1"

There are some tests under the test folder. To run them:

• install test dependencies (see "Installation")
• [optionally] set up the environmental variable OUT_PATH - the output path for tests artifacts, e. g. on Linux

  export OUT_PATH=test/out

• run pytest from the root directory of the repository:

  python -m pytest

Some tests produce files that are not stored in the repository but are auto-generated and uploaded to artifacts when a CI workflow is triggered (e. g. on a pull request to develop or master). This includes PDF illustration to graphs and code coverage reports. You may download these files:

1. Go to the workflow webpadge
2. Navigate to the latest successful run
3. Download the archive "test output" and unpack it

Simply remove the root folder of the repository from your filesystem.

To report problems (or for any questions whatsoever), create an issue on GitHub.

Alternatively, you may e-mail me at valera.zuev.zva@gmail.com