Quick Start Guide

Flow is in preview. df-flow is not published yet, so the install commands below describe the intended flow rather than one you can run today. Talk to us if you want access while Flow is in preview.

Do not run the install commands below yet. These package names are not reserved, so anything currently published under them is not ours. They document the intended flow; we will announce when they are real.

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 df-flow

# Using uv (recommended)
uv pip install df-flow

Or install from source:

git clone https://github.com/Digital-Frontier-LDA/df-flow.git
cd df-flow
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 df_flow.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 df_flow import Flow

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

# 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 df_flow import Flow
from df_flow.base import BaseService

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

# 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 SyncFlow

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

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

from df_flow import SyncFlow
from df_flow.base import BaseService

# Create sync app
app = SyncFlow()

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

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

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

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

@app.workflow
def data_pipeline(source_id: str, services=None):
    data = fetch_step(source_id, services=services)
    return 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}")

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?

• Email us at support@digitalfrontier.so