//flow

Quick Start Guide

Build your first DigitalFrontier Flow pipeline in minutes

Docs are being migrated. These are the getting-started guides. Product names have been updated to DigitalFrontier, but the SDK, CLI and config identifiers in code samples still reference their current names — we publish the surface that actually exists rather than a renamed one. The full API reference is being moved to its own documentation site.

Get started with DigitalFrontier Flow and build your first asynchronous pipeline in just a few minutes.

Prerequisites

Python 3.11 or higher
Basic understanding of Python async/await

That's it! DigitalFrontier Flow uses our managed SaaS infrastructure - no Redis, Docker, or infrastructure setup required.

Installation

Install DigitalFrontier Flow using pip or uv:

# Using pip
pip install blazing

# Using uv (recommended)
uv pip install blazing

Or install from source:

git clone https://github.com/your-org/blazing.git
cd blazing
uv sync
source .venv/bin/activate

Your First Pipeline

Let's create a simple data processing pipeline that demonstrates the core concepts of DigitalFrontier Flow.

Step 1: Define a Service

Services encapsulate your business logic and can be reused across different steps:

from blazing.base import BaseService

class DataProcessor(BaseService):
    """A service for processing data"""

    async def fetch_data(self, source_id: str):
        """Fetch data from a source"""
        # Simulate fetching data
        return {"id": source_id, "data": [1, 2, 3, 4, 5]}

    async def transform_data(self, data: dict):
        """Transform the data"""
        # Apply some transformation
        transformed = {
            "id": data["id"],
            "sum": sum(data["data"]),
            "count": len(data["data"])
        }
        return transformed

    async def save_result(self, result: dict):
        """Save the processed result"""
        # Save to database, file, or API
        print(f"Saved result: {result}")
        return result

Step 2: Create a DigitalFrontier Application

Initialize your DigitalFrontier application - it uses our SaaS infrastructure by default:

from blazing import Blazing

# Create the app (uses DigitalFrontier by default)
app = Blazing()

# Register your service
@app.service
class DataProcessor(BaseService):
    # ... your service implementation

Step 3: Define Steps

Steps are individual processing units in your pipeline:

@app.step
async def fetch_step(source_id: str, services=None):
    """Step that fetches data"""
    processor = services['DataProcessor']
    data = await processor.fetch_data(source_id)
    return data

@app.step
async def transform_step(data: dict, services=None):
    """Step that transforms data"""
    processor = services['DataProcessor']
    result = await processor.transform_data(data)
    return result

@app.step
async def save_step(result: dict, services=None):
    """Step that saves the result"""
    processor = services['DataProcessor']
    final = await processor.save_result(result)
    return final

Step 4: Define a Workflow

Workflows orchestrate steps into a complete pipeline:

@app.workflow
async def data_pipeline(source_id: str, services=None):
    """Complete data processing pipeline"""
    # Fetch data
    data = await fetch_step(source_id, services=services)

    # Transform data
    result = await transform_step(data, services=services)

    # Save result
    final = await save_step(result, services=services)

    return final

Step 5: Run Your Pipeline

Put it all together and execute your workflow:

import asyncio

async def main():
    # Publish your app to DigitalFrontier
    await app.publish()

    # Execute the pipeline - three equivalent ways:

    # 1. One-liner with wait_result() - SIMPLEST! ⭐
    result = await app.data_pipeline("source_123").wait_result()

    # 2. Using RemoteRun handle (more explicit)
    run = await app.data_pipeline("source_123")
    result = await run.result()

    # 3. Using run() method (by name)
    run = await app.run("data_pipeline", "source_123")
    result = await run.result()

    print(f"Pipeline result: {result}")

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

Complete Example

Here's the complete code in one file:

import asyncio
from blazing import Blazing
from blazing.base import BaseService

# Create app (uses DigitalFrontier by default)
app = Blazing()

# Define service
@app.service
class DataProcessor(BaseService):
    async def fetch_data(self, source_id: str):
        return {"id": source_id, "data": [1, 2, 3, 4, 5]}

    async def transform_data(self, data: dict):
        return {
            "id": data["id"],
            "sum": sum(data["data"]),
            "count": len(data["data"])
        }

    async def save_result(self, result: dict):
        print(f"Saved result: {result}")
        return result

# Define steps
@app.step
async def fetch_step(source_id: str, services=None):
    processor = services['DataProcessor']
    return await processor.fetch_data(source_id)

@app.step
async def transform_step(data: dict, services=None):
    processor = services['DataProcessor']
    return await processor.transform_data(data)

@app.step
async def save_step(result: dict, services=None):
    processor = services['DataProcessor']
    return await processor.save_result(result)

# Define workflow
@app.workflow
async def data_pipeline(source_id: str, services=None):
    data = await fetch_step(source_id, services=services)
    result = await transform_step(data, services=services)
    final = await save_step(result, services=services)
    return final

# Run it
async def main():
    await app.publish()

    # Execute pipeline (using simplest one-liner syntax)
    result = await app.data_pipeline("source_123").wait_result()
    print(f"Pipeline result: {result}")

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

Running the Example

Save the code to my_pipeline.py and run it:

python my_pipeline.py

You should see output like:

Saved result: {'id': 'source_123', 'sum': 15, 'count': 5}
Pipeline result: {'id': 'source_123', 'sum': 15, 'count': 5}

Learning Without Async? Try SyncBlazing

Note: SyncBlazing is designed for learning and prototyping only. For production, we strongly recommend the async Blazing class for better performance.

If you're learning DigitalFrontier Flow and don't want to deal with async/await yet, use SyncBlazing:

from blazing import SyncBlazing
from blazing.base import BaseService

# Create sync app
app = SyncBlazing()

@app.service
class DataProcessor(BaseService):
    async def fetch_data(self, source_id: str):
        return {"id": source_id, "data": [1, 2, 3, 4, 5]}

    async def transform_data(self, data: dict):
        return {"id": data["id"], "sum": sum(data["data"]), "count": len(data["data"])}

@app.step
async def fetch_step(source_id: str, services=None):
    return await services['DataProcessor'].fetch_data(source_id)

@app.step
async def transform_step(data: dict, services=None):
    return await services['DataProcessor'].transform_data(data)

@app.workflow
async def data_pipeline(source_id: str, services=None):
    data = await fetch_step(source_id, services=services)
    return await transform_step(data, services=services)

# No async/await, no asyncio.run() - just regular Python!
app.publish()
result = app.data_pipeline("source_123")  # Returns result directly
print(f"Result: {result}")

See the Synchronous API guide for more details.

Next Steps

Now that you have a basic pipeline running, explore more advanced features:

Core Examples - See more workflow patterns
Services & Connectors - Connect to databases, APIs, and more
Worker Optimization - Tune performance for your workloads

Need Help?

Join our Discord community
Contact support at admin@digitalfrontier.so

DigitalFrontier Flow

Flow Sandbox