Skip to content

Welcome to Chorey: The Asynchronous, Type-Safe Python Pipeline Framework

Chorey is a modern, lightweight, and highly testable framework for building complex data and control flows in Python. It allows you to chain asynchronous functions, manage retries, handle parallel processing, and implement conditional routing, all while preserving end-to-end type safety.

Key Features

  • Asynchronous by Design: Built on asyncio, Chorey ensures your pipelines run efficiently and non-blockingly, perfect for I/O-bound tasks like calling external APIs or interacting with databases.
  • Intuitive Chaining (.next()): Construct complex sequences of operations using a clean, fluent API.
  • Advanced Control Flow:
  • Parallelism (.branch()): Run multiple independent steps concurrently, merging the results into an order-aware, typed tuple.
  • Conditionals (.route()): Define dynamic execution paths based on the output of a previous step that can be piped into further processing with the union type of all possible outputs.
  • Type Safety & Introspection: Chorey enforces strict type contracts between steps and can introspect the required input type of the entire pipeline.
  • Resilience: Built-in support for retries and custom failure callbacks (.retry(), .on_failure_do()).
  • Visualization: Automatically generate Mermaid diagrams for any pipeline, providing clear visual debugging for even the most complex flows.

Quick Start Example

from chorey import step
from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str

@dataclass
class EnrichedUser(User):
    comments_made: int
    # more fields that may enrich the user

async def fetch_user(user_id: int) -> User:
    # Simulate fetching user from a database or API
    return User(id=user_id, name="John Doe")

async def enrich_user(user: User) -> EnrichedUser:
    # Simulate enriching user data
    return EnrichedUser(id=user.id, name=user.name, comments_made=42)

# define the pipeline
pipeline = step(fetch_user).next(enrich_user)

if __name__ == "__main__":
    import asyncio

    result = asyncio.run(pipeline.feed(1))
    print(result)  # Output: EnrichedUser(id=1, name='John Doe', comments_made=42)

Next Steps

Guide Description
Core Concepts Dive deeper into Chorey's foundational concepts like Feeder, type arguments, and pipeline structure
Data Flow Understand how data and state are passed between steps
Branching - Concurrent Execution Learn how to run multiple steps in parallel and combine their results
Routing - Conditional Execution Discover how to implement conditional logic in your pipelines
ChoreyContext - Shared State Explore how to manage shared state across your pipeline steps
Resilience Configure steps to automatically retry on failure or execute custom side effects
Visualization Generate Mermaid diagrams to visualize your pipeline structure
CLI Learn how to run pipelines directly from the command line