python-quantumflow 2.0.0

Python QuantumFlow: Advanced type conversion for Python

⚛️ Python QuantumFlow 🔄

✨ A next-generation type-and-data-flow framework for Python with a lightweight footprint. 🚀

🌟 Overview

🔮 Python QuantumFlow is a powerful yet lightweight framework that enhances Python's data flow capabilities with:

• 🔄 Automatic type conversion with flow() and TypeFlowContext
• 🧩 Function decorators for creating intelligent flows with @qflow
• ⚡ Asynchronous operations with @async_flow
• 🛡️ Robust error handling with retry logic
• 🎨 Beautiful terminal output with color and styling
• 📊 Visualization of data flows
• 🔁 ETL pipeline creation and orchestration
• 🧪 Advanced testing support for flow validation
• 📈 Performance monitoring and optimization tools
• 🔍 Debugging and introspection capabilities
• 🌐 Distributed computing integration

🚀 What's New in Version 2.0

Python QuantumFlow 2.0 represents a major evolution from the original python-typeflow, with significant improvements:

Key Enhancements

• ⚡ Performance Boost: Up to 3x faster type conversions and flow execution
• 🧠 Smarter Type Inference: Improved algorithm for detecting and converting complex nested types
• 🔄 Flow Composition: Chain and combine flows with intuitive operators
• 🌐 Extended Ecosystem: New integrations with popular data science and ML frameworks
• 🧪 Enhanced Testing Tools: Built-in utilities for testing flows and mocking data sources

Version Comparison

| Feature | Version 1.x | Version 2.x | | ---------------- | ----------------- | ---------------------------------- | --- | | Type Conversion | Basic types only | Complex nested structures | | Error Handling | Manual try/except | Automatic with @retry | | Async Support | Limited | Full async/await with backpressure | | Flow Composition | Manual chaining | Operator-based (>>, +, |) | | Memory Usage | Moderate | Optimized with streaming support | | Visualization | None | Interactive flow diagrams | | CLI Tools | None | Complete development toolkit |

Migration from v1.x

Upgrading from python-typeflow 1.x to Python QuantumFlow 2.x is straightforward:

# Old way (python-typeflow 1.x)
from python_typeflow import convert_type, apply_flow

result = apply_flow(data, convert_type(str))

# New way (Python QuantumFlow 2.x)
from python_quantumflow.core import flow

result = flow(lambda x: str(x))(data)
# or even simpler
result = flow(str)(data)

📚 Complete migration guide

For a complete guide to migrating your code from python-typeflow 1.x to Python QuantumFlow 2.x, see our Migration Guide.

Key differences:

• Decorator syntax changes
• New context manager approach
• Enhanced error handling
• Parallel and distributed execution options

📦 Installation

# Install core package
pip install python-quantumflow

# Install with additional features
pip install python-quantumflow[viz]      # Visualization features
pip install python-quantumflow[async]    # Enhanced async capabilities
pip install python-quantumflow[ml]       # Machine learning integrations
pip install python-quantumflow[full]     # All features

🚀 Quick Start

Basic Type Flow

from python_quantumflow.core import flow, with_typeflow as TypeFlowContext

# Create a list of numbers
numbers = [1, 2, 3, 4, 5]

# Convert types using flow
numbers_str = flow(lambda x: str(x))(numbers)
sum_str = flow(lambda x: str(x))(sum(numbers))

# Combine the results
result = f"{numbers_str} items with sum: {sum_str}"
print(result)

👉 Click to see output

[1, 2, 3, 4, 5] items with sum: 15

Function Flows with @qflow

from python_quantumflow.core import qflow

@qflow
def process_data(items):
    """Process a list of items by doubling each value."""
    return [item * 2 for item in items]

data = [5, 10, 15, 20, 25]
result = process_data(data)
print(f"Processed data: {result}")

👉 Click to see output

Processed data: [10, 20, 30, 40, 50]

Async Flow with Retry

import asyncio
from python_quantumflow.core import async_flow, retry, fancy_print

@async_flow
@retry(max_attempts=3, backoff_factor=0.5)
async def fetch_remote_data(url):
    """Fetch data from a URL with automatic retry on failure."""
    await asyncio.sleep(0.2)  # Simulate network delay

    # Your network request logic here
    return f"Data from {url}"

async def main():
    urls = ["https://api.example.com/data/1", "https://api.example.com/data/2"]
    tasks = [fetch_remote_data(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Process results
    for url, result in zip(urls, results):
        if isinstance(result, Exception):
            fancy_print(f"Failed to fetch {url}: {result}", style="red")
        else:
            fancy_print(f"Success: {result}", style="green")

if __name__ == "__main__":
    asyncio.run(main())

👉 Click to see output

Success: Data from https://api.example.com/data/1
Success: Data from https://api.example.com/data/2

With failures (when network errors occur):

Request to https://api.example.com/data/1 failed, retrying...
Failed to fetch https://api.example.com/data/1: Failed to connect
Success: Data from https://api.example.com/data/2

Color and Styling

from python_quantumflow.core import fancy_print

# Styled output with explicit styling
fancy_print("Error: Connection failed", style="bold red")
fancy_print("Success: Operation completed", style="bold green")
fancy_print("Info: Processing data", style="cyan")

# Auto-styled output based on content
fancy_print("Error: This will be red automatically")
fancy_print("Success! This will be green automatically")
fancy_print("Warning: This will be yellow automatically")

👉 Click to see output

Note: Colors shown will vary based on your terminal

ETL Pipeline Example

from python_quantumflow.core import qflow, fancy_print

@qflow
def extract(source):
    """Extract data from a source."""
    fancy_print(f"Extracting from {source}...", style="dim")
    return [1, 2, 3, 4, 5]

@qflow
def transform(data):
    """Transform the raw data."""
    fancy_print(f"Transforming {data}...", style="dim")
    return [item * 2 for item in data]

@qflow
def load(transformed_data):
    """Load the transformed data."""
    fancy_print(f"Loading {transformed_data}...", style="dim")
    return f"Loaded {len(transformed_data)} items"

@qflow
def etl_pipeline(source):
    """Complete ETL pipeline."""
    raw_data = extract(source)
    transformed_data = transform(raw_data)
    result = load(transformed_data)
    return result

result = etl_pipeline("database://example/table1")
fancy_print(f"Pipeline result: {result}", style="bold green")

👉 Click to see output

Extracting from database://example/table1...
Transforming [1, 2, 3, 4, 5]...
Loading [2, 4, 6, 8, 10]...
Pipeline result: Loaded 5 items

🧰 Key Components

• Core Flow Functions:

  • flow(): Wraps objects for type conversion
  • TypeFlowContext: Context for automatic type conversion
  • qflow: Decorator for creating flow functions
  • async_flow: Decorator for asynchronous flows

• Error Handling:

  • retry: Decorator for automatic retry on failure
  • Exception handling within flows

• Terminal Enhancements:

  • fancy_print: Rich terminal output with colors and styles
  • Auto-styling based on message content
  • Progress indicators and animations

🖥️ Command Line Interface

Python QuantumFlow includes a CLI for rapid development and learning:

# Start interactive playground
pqflow play

# Start guided tutorial
pqflow tutor

# Generate documentation for flows
pqflow autodoc your_module

# Run a flow
pqflow run your_module.your_flow --input data.json

# Visualize a flow
pqflow visualize your_module.your_flow --output flow.png

# Create a new Python QuantumFlow project
pqflow init my_project

# View performance metrics for flows
pqflow metrics your_module.your_flow

# Run flow with default colorized output
pqflow run --color your_module.your_flow

# Apply type conversions to a JSON file
pqflow convert input.json output.json --schema schema.json

# Start a web dashboard for monitoring flows
pqflow dashboard --port 8080

# Generate sample code from templates
pqflow generate etl --name data_pipeline

# Benchmark flow performance
pqflow benchmark your_module.your_flow --iterations 1000

👉 Click to see CLI in action

The Python QuantumFlow CLI provides a fun, interactive way to learn and use the framework.

🔬 Advanced Features

• 📊 Metrics and Observability: Monitor and track flow execution
• ✅ Validation: Validate flow inputs and outputs
• ⚙️ Execution Backends: Run flows in different execution contexts (threads, processes, distributed)

📊 Performance

Python QuantumFlow is optimized for both performance and flexibility.

Feature	Performance	Memory Usage	Quantum Awesomeness
Type Conversion	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐
Flow Execution	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐
Async Operations	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐⭐
Fancy Output	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐

🧠 Why Python QuantumFlow?

• Lightweight: Minimal dependencies, small footprint
• Intuitive: Simple API with powerful capabilities
• Flexible: Works with your existing code
• Pretty: Makes your terminal output gorgeous
• Fast: Optimized for performance

🙌 Created By

Magi Sharma
Quantum-entangled code enthusiast
Version: 0.6.1

---

Made with ❤️ and a sprinkle of quantum magic