Flow Logo

API Documentation

API Quickstart

Flow provides multiple ways to interact with the platform programmatically:

  • REST API - Comprehensive endpoints for all platform operations
  • Python Client - High-level interface for common workflows
  • Direct HTTP - For custom integrations and scripting

For complete API documentation, visit docs.api.flow.bio

Python Client (flowbio)

The easiest way to interact with Flow programmatically is through the flowbio Python client.

Installation

pip install flowbio

Basic Usage

import flowbio

# Initialize client and authenticate
client = flowbio.Client()
client.login("your_username", "your_password")

# Upload a sample with paired-end reads
sample = client.upload_sample(
    "My RNA-seq Sample",
    "/path/to/reads_R1.fastq.gz",
    "/path/to/reads_R2.fastq.gz",
    progress=True,
    metadata={
        "organism": "Homo sapiens",
        "tissue": "liver",
        "treatment": "control"
    }
)

print(f"Sample uploaded: {sample.id}")

Common Operations

Upload Different Data Types

# Standard data file (genomes, annotations, etc.)
data = client.upload_data(
    "/path/to/genome.fa",
    progress=True,
    retries=5
)

# Multiplexed sequencing data
multiplexed = client.upload_multiplexed(
    "/path/to/multiplexed.fastq.gz",
    progress=True
)

# Sample annotation sheet
annotation = client.upload_annotation(
    "/path/to/samples.csv",
    progress=True
)

Run Pipelines

# Run RNA-seq pipeline on samples
execution = client.run_pipeline(
    pipeline="RNA-seq",
    version="3.8.1",
    nextflow_version="23.04.3",
    params={
        "aligner": "star_salmon",
        "trimmer": "trimgalore"
    },
    sample_ids=[sample.id]
)

print(f"Pipeline execution started: {execution.id}")

Query Existing Data

# Get your samples
my_samples = client.get_samples(user="me")

# Get public samples
public_samples = client.get_samples(public=True, organism="Mus musculus")

# Get pipeline executions
executions = client.get_executions(status="completed")

REST API

The REST API provides comprehensive endpoints for all platform operations.

Base URL

https://api.flow.bio/api/

Authentication

Authenticate using JWT tokens:

# Login to get tokens
curl -X POST https://api.flow.bio/api/auth/login/ \
  -H "Content-Type: application/json" \
  -d '{"username": "your_username", "password": "your_password"}'

Include the access token in subsequent requests:

Authorization: Bearer <access_token>

Example Requests

Get Sample Details

curl -H "Authorization: Bearer <token>" \
  https://api.flow.bio/api/samples/<sample_id>/

List Available Pipelines

curl -H "Authorization: Bearer <token>" \
  https://api.flow.bio/api/pipelines/

Monitor Execution Progress

curl -H "Authorization: Bearer <token>" \
  https://api.flow.bio/api/executions/<execution_id>/

Example POST Requests

Create a Project

curl -X POST https://api.flow.bio/api/projects/ \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "My Project",
    "description": "RNA-seq analysis project"
  }'

Update Sample Metadata

curl -X PATCH https://api.flow.bio/api/samples/<sample_id>/ \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{
    "metadata": {
      "tissue": "brain",
      "age": 42
    }
  }'

File Upload

Upload files using multipart/form-data:

curl -X POST https://api.flow.bio/api/upload/ \
  -H "Authorization: Bearer <token>" \
  -F "file=@/path/to/file.fastq.gz" \
  -F "name=My Sample" \
  -F "project_id=12345"

File Download

Download execution outputs:

curl -X GET https://api.flow.bio/api/download/<file_id>/ \
  -H "Authorization: Bearer <token>" \
  -o output.zip

Health Check

Check API status:

curl https://api.flow.bio/health/

Rate Limits and Best Practices

Rate Limits

  • API Requests: 1000 requests per hour
  • File Upload: 100 uploads per hour
  • File Download: 500 downloads per hour

Best Practices

  1. Use pagination for large result sets:

    # Get paginated results
page = 1
while True:
    response = requests.get(
        f'https://api.flow.bio/api/samples/?page={page}',
        headers={'Authorization': f'Bearer {token}'}
    )
    data = response.json()
    process_samples(data['results'])
    if not data['next']:
        break
    page += 1

  2. Request only needed fields to reduce response size

  3. Batch operations when possible:

    # Good - single request
samples = client.upload_samples([
    {"name": "Sample1", "path": "file1.fastq"},
    {"name": "Sample2", "path": "file2.fastq"}
])

# Avoid - multiple requests
for sample in samples_list:
    client.upload_sample(sample["name"], sample["path"])

  4. Handle errors gracefully with retries:

    try:
    result = client.run_pipeline(...)
except flowbio.APIError as e:
    if e.status_code == 429:  # Rate limited
        time.sleep(60)
        result = client.run_pipeline(...)

  5. Use webhooks for long-running operations instead of polling

Advanced Examples

Bulk Processing Pipeline

import flowbio
import pandas as pd

client = flowbio.Client()
client.login("username", "password")

# Read sample sheet
samples_df = pd.read_csv("samples.csv")

# Upload all samples
uploaded_samples = []
for _, row in samples_df.iterrows():
    sample = client.upload_sample(
        name=row["sample_name"],
        r1_path=row["r1_file"],
        r2_path=row["r2_file"],
        metadata={
            "condition": row["condition"],
            "replicate": row["replicate"]
        }
    )
    uploaded_samples.append(sample)

# Run pipeline on all samples
execution = client.run_pipeline(
    pipeline="RNA-seq",
    version="latest",
    sample_ids=[s.id for s in uploaded_samples],
    params={
        "genome": "GRCh38",
        "aligner": "star_salmon"
    }
)

# Monitor execution
while execution.status not in ["completed", "failed"]:
    time.sleep(60)
    execution = client.get_execution(execution.id)
    print(f"Status: {execution.status}, Progress: {execution.progress}%")

Custom REST Client

import requests

class FlowRESTClient:
    def __init__(self, token):
        self.base_url = "https://api.flow.bio/api"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
    
    def get(self, endpoint, params=None):
        response = requests.get(
            f"{self.base_url}{endpoint}",
            params=params,
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()
    
    def post(self, endpoint, data=None):
        response = requests.post(
            f"{self.base_url}{endpoint}",
            json=data,
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()

# Use custom client
client = FlowRESTClient(token="your_token")

# Get user's projects
projects = client.get("/projects/", params={"user": "me"})
for project in projects["results"]:
    print(f"{project['name']}: {project['sample_count']} samples")

Next Steps

Previous
Common file formats
Next
API Overview