DigitalFrontier Flow Documentation

Serverless, Persistent async pipelines with the power that Lambda can't provide.

Get Started

• Quick Start 🚀 - Build your first async pipeline in 5 minutes
• Guide 📚 - Complete guide with examples

Serverless, Persistent: The Full Stack

Lambda gives you serverless. DigitalFrontier Flow gives you serverless and persistence. The workflow/step/service architecture unlocks capabilities that traditional serverless cannot provide:

⚡ Full Python

Lambda/Cloud Run: Rigid DAGs, Step Functions, JSON state machines. No if statements, no loops, no real programming.

DigitalFrontier Flow: Write real Python. If statements, loops, dynamic branching. Unlimited complexity.

@app.workflow
async def pipeline(user_id: str):
    user = await fetch_user(user_id)

    # Real Python - if statements, loops, everything
    if user['premium']:
        result = await premium_processing(user)
    else:
        result = await standard_processing(user)

    for item in result['items']:
        await process_item(item)

    return result

🔄 Full Reusability

Lambda/Cloud Run: Cold starts every time. Connections rebuilt. State lost. 50-200ms overhead per invocation.

DigitalFrontier Flow: Connections pooled. State cached. Warmup once, reuse forever. 5ms per task.

class DatabaseService(BaseService):
    def __init__(self, connector_instances):
        # Created ONCE - persists across thousands of tasks
        self.db_pool = connector_instances['postgres']
        self.cache = {}

    async def query(self, sql: str):
        # Connection already warm - no cold start
        return await self.db_pool.execute(sql)

🔒 Full Security

Lambda/Cloud Run: All code has full access. User code can steal credentials, access databases, leak secrets.

DigitalFrontier Flow: Sandboxed steps run isolated. Service calls automatically execute in trusted environment. Credentials never leak - architecture prevents it.

@app.step(sandboxed=True)
async def user_code(x: int, y: int, services=None):
    # Runs isolated - no credential access

    # Service call → executes in TRUSTED environment
    result = await services['Database'].query(f"SELECT {x} + {y}")
    # Credentials never enter sandbox

    return result

Architecture: Sandboxed code → Service call → Trusted execution → Safe result

📈 Full Scalability

Lambda/Cloud Run: Limited concurrency, timeout limits, regional restrictions.

DigitalFrontier Flow: Unlimited workers, no timeouts, scale to millions of tasks.

🏛️ Full Sovereignty

Lambda/Cloud Run: Locked to AWS/GCP. No control over where code runs.

DigitalFrontier Flow: Deploy on GCP, Akash Network, or hybrid. Your infrastructure, your control, your data residency.

⚡ The Comparison

	AWS Lambda	Cloud Run	DigitalFrontier Flow
Python	❌ DAGs only	❌ Step Functions	✅ Full Python
Reusability	❌ Cold starts	❌ Per request	✅ Pool & Cache
Security	❌ Full access	❌ Full access	✅ Sandboxed
Scalability	⚠️ Limited	⚠️ Limited	✅ Unlimited
Sovereignty	❌ AWS only	❌ GCP only	✅ Multi-cloud

When to Use DigitalFrontier Flow

✅ Complex multi-stage workflows - Pipelines with branching logic and multiple steps

✅ Stateful computations - Automatic checkpointing and partial recovery

✅ Mixed workloads - Combination of I/O-bound and CPU-bound tasks

✅ Long-running pipelines - Workflows that run for hours or days with automatic recovery

✅ Dynamic routing - Pipeline structure that changes based on data

Quick Example

from blazing import Blazing

# Create your app
app = Blazing()

# Define processing steps with @step
@app.step
async def fetch_data(source_id: str):
    # Fetch from database or API
    return {"id": source_id, "data": [1, 2, 3]}

@app.step
async def transform_data(raw_data: dict):
    # Transform the data
    total = sum(raw_data["data"])
    return {"id": raw_data["id"], "result": total}

# Define the pipeline with @route
@app.workflow
async def data_pipeline(source_id: str):
    # Orchestrate the workflow
    data = await fetch_data(source_id)
    result = await transform_data(data)
    return result

# Publish and start
async def main():
    # Publish infrastructure (creates Redis indices, registers routes)
    await app.publish()

    # Start the app (makes routes available to workers)
    await app.start()

    # Submit a task
    unit = await app.create_workflow_task("data_pipeline", source_id="123")
    result = await unit.result()
    print(result)  # {"id": "123", "result": 6}

Key Features

📝 Declarative Pipelines

Define workflows using simple decorators - @workflow for pipelines, @step for steps.

🔄 Automatic Recovery

Pipelines resume automatically after interruptions. No lost work, no manual intervention.

⚡ Smart Workers

System automatically balances async and blocking workers based on your workload patterns.

🔌 Persistent Connections

Keep database and SSH connections alive across tasks for better performance.

📊 Built-in Monitoring

Real-time dashboard shows queue depths, worker states, and throughput metrics.

Learn More

• Quick Start Guide - Build your first pipeline in 5 minutes
• Complete Guide - In-depth examples and patterns