flowkitx 1.0.1

Fluent async workflows with lazy execution for Python

🚀 FlowKit - Simplify Async Workflows

FlowKit provides fluent async workflows with lazy execution and minimal boilerplate.

⚡ Why FlowKit?

Before (Traditional Async):

async def fetch_user_data(user_id):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as response:
            user = await response.json()
            async with session.get("https://api.example.com/posts", params={"userId": user_id}) as posts_response:
                posts = await posts_response.json()
                return {"user": user, "posts": posts}

After (FlowKit):

@flowkit.simple
def fetch_user_data(user_id):
    user = flowkit.get(f"https://api.example.com/users/{user_id}").json()
    posts = flowkit.get("https://api.example.com/posts", params={"userId": user_id}).json()
    return {"user": user, "posts": posts}

🎯 Key Features

• 🔗 Fluent API - Chain operations with .pipe() method
• 🛡️ Type Safe - Full IDE support and type hints
• 🚀 Modern Foundation - Built on battle-tested httpx
• 🔧 Framework Ready - FastAPI, Django, Flask examples
• 📦 Simple Installation - pip install flowkit-async

📦 Installation

pip install flowkit-async

🔧 How FlowKit Works

FlowKit provides lazy async pipelines - operations are queued and executed when awaited.

import flowkit

# FlowKit operations return awaitable Flow objects
user_flow = flowkit.get("https://api.example.com/users/1")
posts_flow = flowkit.get("https://api.example.com/posts")

# Execute both operations
user = await user_flow.json()
posts = await posts_flow.json()

Core Concept: Flow Objects

Flow objects are awaitable pipelines that:

• Chain operations without immediate execution
• Execute when awaited in the current event loop
• Enable fluent, readable async patterns

This is structured async workflows, not sync-to-async magic.

🔧 How FlowKit Works

FlowKit provides lazy async pipelines. API calls return awaitable Flow objects that record operations and execute asynchronously when awaited.

FlowKit does NOT:

• convert blocking code into async
• use threads or greenlets
• bypass the event loop

This is structured async composition, not async magic.

🏆 Real-World Examples

Data Processing Pipeline

@flowkit.simple
def fetch_user_data(user_id: int):
    """Fetch and process user data."""
    return (flowkit.get(f"https://jsonplaceholder.typicode.com/users/{user_id}")
              .json()
              .pipe(lambda user: {**user, "processed": True}))

Note: @flowkit.simple is intended for synchronous functions. If used on async functions, it acts as a pass-through wrapper.

Concurrent Operations

@flowkit.simple
def fetch_dashboard():
    """Fetch multiple data sources concurrently."""
    users = flowkit.get("https://jsonplaceholder.typicode.com/users").json()
    posts = flowkit.get("https://jsonplaceholder.typicode.com/posts").json()
    comments = flowkit.get("https://jsonplaceholder.typicode.com/comments").json()
    
    # These execute concurrently when awaited
    return {"users": users, "posts": posts, "comments": comments}

🛠 When to Use FlowKit

FlowKit is ideal for:

• API-heavy applications with multiple calls
• Data pipelines and processing workflows
• Teams wanting consistent async patterns
• Rapid prototyping with clean syntax

⚠️ Important Notes

FlowKit does NOT:

• Block the event loop or use thread pools
• Make synchronous code magically non-blocking
• Replace understanding of asyncio fundamentals
• Work with CPU-bound operations

Always use async/await with CPU-intensive tasks.

🆚 How FlowKit Compares

Feature	FlowKit	aiohttp	httpx	requests
Fluent API	✅	❌	❌	❌
Type Safe	✅	✅	✅	✅
Sessions	✅	✅	✅	❌
Pipelines	✅	❌	❌	❌

FlowKit provides structured async workflows with minimal overhead on top of httpx and is comparable in performance for typical I/O-bound workloads.

📋 License

MIT License - see LICENSE file for details.

🤝 Contributing

We welcome contributions! Please see our Contributing Guide for details.

---

🚀 Try FlowKit today for cleaner async workflows:

pip install flowkit-async

⭐ Star us on GitHub! github.com/flowkit/flowkit