Flow Logo

API Documentation

Python CLI Commands

Overview

The flowbio Python client provides a comprehensive set of commands for interacting with Flow. This page provides a quick reference for the most commonly used commands.

For detailed documentation and advanced usage, see the Python Client Guide.

Installation & Setup

# Install via pip
pip install flowbio

# Install via conda
conda install -c bioconda flowbio

Basic Setup

import flowbio

# Initialize client
client = flowbio.Client()

# Login
client.login("username", "password")

Environment Variables

export FLOW_API_URL="https://api.flow.bio"
export FLOW_USERNAME="myusername"
export FLOW_PASSWORD="mypassword"

Authentication Commands

# Login with credentials
client.login("username", "password")

# Login with environment variables
client.login()

# Get current user info
print(client.user.username)
print(client.user.email)

# Logout
client.logout()

Project Commands

# List all your projects
projects = client.get_projects()

# Create a new project
project = client.create_project(
    name="My Project",
    description="Project description"
)

# Get specific project
project = client.get_project("project-id")

# Update project
project.update(description="New description")

# Search projects
results = client.search_projects("keyword")

# Delete project
project.delete()

Sample Commands

Upload Samples

# Upload single-end sample
sample = client.upload_sample(
    name="Sample Name",
    read1="/path/to/reads.fastq.gz",
    metadata={"organism": "Homo sapiens"}
)

# Upload paired-end sample
sample = client.upload_sample(
    name="Sample Name",
    read1="/path/to/reads_R1.fastq.gz",
    read2="/path/to/reads_R2.fastq.gz",
    project_id=project.id,
    metadata={"organism": "Homo sapiens"}
)

# Upload with progress bar
sample = client.upload_sample(
    name="Large Sample",
    read1="/path/to/file.fastq.gz",
    progress=True
)

Query Samples

# Get all your samples
samples = client.get_samples()

# Get samples from a project
samples = client.get_samples(project_id=project.id)

# Search samples
results = client.search_samples("liver")

# Get specific sample
sample = client.get_sample("sample-id")

# Delete sample
sample.delete()

Pipeline Commands

List and Run Pipelines

# List all available pipelines
pipelines = client.get_pipelines()

# Get specific pipeline
pipeline = client.get_pipeline("RNA-seq")

# Run pipeline with default parameters
execution = client.run_pipeline(
    pipeline="RNA-seq",
    samples=[sample1.id, sample2.id]
)

# Run with custom parameters
execution = client.run_pipeline(
    pipeline="RNA-seq",
    version="3.14.0",
    samples=[sample.id],
    params={
        "aligner": "star_salmon",
        "skip_trimming": False
    },
    genome_id="GRCh38"
)

Monitor Executions

# Get execution status
execution = client.get_execution("execution-id")
print(execution.status)
print(execution.progress)

# List all your executions
executions = client.get_executions()

# Wait for completion
import time
while execution.status in ["pending", "running"]:
    time.sleep(30)
    execution.refresh()

# Get execution logs
logs = execution.get_logs()

Download Results

# Download all results to directory
execution.download_results("/path/to/output/")

# Download specific file
output = execution.get_output("multiqc_report.html")
output.download("/path/to/report.html")

# Get output as stream
output = execution.get_output("gene_counts.csv")
with output.get_stream() as f:
    content = f.read()

Data File Commands

# Upload data file
data = client.upload_data(
    "/path/to/file.gtf",
    description="Annotation file"
)

# List data files
data_files = client.get_data()

# Search data files
results = client.search_data("*.gtf")

# Download data file
data.download("/path/to/output.gtf")

# Delete data file
data.delete()

Genome Commands

# List available genomes
genomes = client.get_genomes()

# Get specific genome
genome = client.get_genome("GRCh38")

# Create custom genome
genome = client.create_genome(
    name="Custom Genome",
    species="Homo sapiens",
    fasta_id=fasta_data.id,
    gtf_id=gtf_data.id
)

Bulk Operations

# Bulk upload samples from CSV
import pandas as pd

df = pd.read_csv("samples.csv")
samples = []
for _, row in df.iterrows():
    sample = client.upload_sample(
        name=row["name"],
        read1=row["read1"],
        read2=row["read2"],
        metadata=row.to_dict()
    )
    samples.append(sample)

# Bulk delete samples
client.bulk_delete_samples([s.id for s in samples])

# Create bulk download
job = client.create_bulk_download(
    sample_ids=[s.id for s in samples]
)

# Download when ready
while job.status != "completed":
    time.sleep(10)
    job.refresh()
job.download("/path/to/download.zip")

Sharing & Permissions

# Share project with user
project.share(
    user="collaborator@email.com",
    permission="edit"  # read, edit, or share
)

# Share with group
project.share(
    group="lab-members",
    permission="read"
)

# Update privacy
sample.update(private=False)  # Make public

Error Handling

from flowbio.exceptions import (
    FlowError,
    AuthenticationError,
    PermissionError,
    NotFoundError,
    ValidationError
)

try:
    sample = client.upload_sample(
        name="Test",
        read1="/missing/file.fastq.gz"
    )
except FileNotFoundError:
    print("File not found")
except PermissionError:
    print("Permission denied")
except FlowError as e:
    print(f"API error: {e}")

Common Workflows

Complete RNA-seq Analysis

# 1. Create project
project = client.create_project(name="RNA-seq Study")

# 2. Upload samples
samples = []
for file_pair in sample_files:
    sample = client.upload_sample(
        name=file_pair["name"],
        read1=file_pair["r1"],
        read2=file_pair["r2"],
        project_id=project.id,
        progress=True
    )
    samples.append(sample)

# 3. Run pipeline
execution = client.run_pipeline(
    pipeline="RNA-seq",
    samples=[s.id for s in samples],
    genome_id="GRCh38"
)

# 4. Monitor progress
while execution.status == "running":
    time.sleep(60)
    execution.refresh()
    print(f"Progress: {execution.progress}%")

# 5. Download results
if execution.status == "completed":
    execution.download_results("./results/")

Batch Processing

# Process samples in batches
def process_batch(sample_files, batch_size=10):
    for i in range(0, len(sample_files), batch_size):
        batch = sample_files[i:i+batch_size]
        
        # Upload batch
        samples = []
        for f in batch:
            sample = client.upload_sample(
                name=f["name"],
                read1=f["path"],
                project_id=project.id
            )
            samples.append(sample)
        
        # Run pipeline on batch
        execution = client.run_pipeline(
            pipeline="RNA-seq",
            samples=[s.id for s in samples]
        )
        
        yield execution

# Process all samples
executions = list(process_batch(all_samples))

Performance Tips

# Use larger chunk size for big files
client = flowbio.Client(chunk_size=50*1024*1024)  # 50MB

# Enable parallel uploads
sample = client.upload_sample(
    name="Large Sample",
    read1="/path/to/100GB.fastq.gz",
    parallel_uploads=4,
    chunk_size=100*1024*1024
)

# Reuse session for multiple requests
import requests
session = requests.Session()
client = flowbio.Client(session=session)

Debugging

# Enable debug logging
import logging
logging.basicConfig(level=logging.DEBUG)

# Debug mode client
client = flowbio.Client(debug=True)

# Custom request logging
def log_request(request):
    print(f"{request.method} {request.url}")
    
client.add_request_hook(log_request)

Next Steps

Previous
Authentication
Next
REST API Reference