One thing I'd mention is that we don't really imagine Pyper as a whole observability and orchestration platform. It's really a package for writing Python functions and executing them concurrently, in a flexible pattern that can be integrated with other tools.

For example, I'm personally a fan of Prefect as an observability platform-- you could define pipelines in Pyper then wrap it in a Prefect flow for orchestration logic.

Exception handling and logging can also be handled by orchestration tools (or in the business logic if appropriate, literally using try... except...)

For a simple progress bar, tqdm is probably the first thing to try. As it wraps anything iterable, applying it to a pipeline might look like:

  import time
  from pyper import task
  from tqdm import tqdm


  @task(branch=True)
  def func(limit: int):
      for i in range(limit):
          time.sleep(0.1)
          yield i


  def main():
      for _ in tqdm(func(limit=20), total=20):
          pass


  if __name__ == "__main__":
      main()

Have you tried multiprocessing.shared_memory to address this?

Concurrency in general isn't about parallelism. It's just about doing multiple things at the same time.

I suppose one excellent thing about this would be if you could just change 1 parameter and switch from multiprocessing to threaded.

you could reassign every line, but it would look nicer with chained functions.

  pipeline = task(get_data, branch=True)
  pipeline = pipeline | task(step1, workers=20)
  pipeline = pipeline |  task(step2, workers=20)
  pipeline = pipeline |  task(step3, workers=20, multiprocess=True)

I would be tempted to do something like this:

  steps = [task(step1, workers=20),
           task(step2, workers=20),
           task(step3, workers=20, multiprocess=True)]
  pipeline = task(get_data, branch=True)

  for step in steps:
      pipeline =   pipeline.__or__(step)