biotailor 0.1.17

Python SDK for the Biotailor bioinformatics pipeline platform

biotailor

Python SDK for the Biotailor bioinformatics pipeline platform.

Table of Contents

• Installation
• Quickstart
• API Reference
  • Client: BiotailorClient
    • Tool discovery
    • Dataset discovery
    • Job management
    • Running and monitoring pipelines
  • Tools: ToolConfig
  • Pipelines: Pipeline
    • Building multi-process pipelines
    • Batch inputs
  • Models
  • Exceptions
• Development
• Publishing to PyPI

Installation

pip install biotailor

Quickstart

from biotailor import BiotailorClient, Pipeline

# 1. Create a client (get your API key from the Biotailor website)
client = BiotailorClient(api_key="btk_xxx.sk_yyy")

# 2. Discover available tools
client.print_tools()

# 3. Build a pipeline
tool = client.get_tool("fastp-single")
tool.print_tool_guide()  # Shows toolId, parameterId, outputId, and option values.

pipeline = (
    Pipeline(name="QC my reads", tool=tool)
    .set_input("input", "reads.fastq")
    .set_param("qualified_quality_phred", 20)
)

# 4. Run and wait
job = client.run_and_wait(pipeline, output_dir="./results")
print(f"Job {job.jobid} finished with status: {job.job_status.value}")

API Reference

Import the main SDK objects from biotailor:

from biotailor import BiotailorClient, Pipeline

Client: BiotailorClient

Create one client per API key. By default, it uses the production API at https://api.biotailor.org.

client = BiotailorClient(
    api_key="btk_xxx.sk_yyy",
    base_url="https://api.biotailor.org",
    timeout=30,
    debug=False,
)

Tool discovery

Method	Returns	Description
list_tools()	list[ToolConfig]	Fetch all available tool configurations.
get_tools()	list[ToolConfig]	Alias for list_tools().
print_tools()	None	Print all tools as a human-readable table.
get_tool(tool_id)	ToolConfig	Fetch one tool by ID, such as "fastp-single" or "bowtie2".

Dataset discovery

Method	Returns	Description
list_datasets(type=None)	list[Dataset]	Fetch available datasets, optionally filtered by dataset type.
print_datasets(type=None)	None	Print datasets grouped by dataset type.

Example:

client.print_datasets(type="bowtie2_db")
datasets = client.list_datasets()

Workflow management

Method	Returns	Description
list_workflows(orgid=None, is_private=None, order="desc")	list[Workflow]	Fetch workflows with optional filters.
print_workflows(orgid=None, is_private=None, order="desc")	None	Print workflows as a human-readable table.
get_workflow(workflow_id)	Workflow	Fetch one workflow by ID.
create_workflow(workflow_name, orgid="null", is_private=True)	Workflow	Create a reusable workflow template.
update_workflow(workflow_id, workflow_name=None, react_flow_workflow=None, is_private=None)	None	Update workflow metadata or graph state.
delete_workflow(workflow_id)	None	Delete a workflow.

Job management

Method	Returns	Description
list_jobs(sortBy="latestStatusTimestamp", order="desc", jobNameFilter=None)	list[Job]	Fetch jobs, with optional sorting and name filtering.
print_jobs(sortBy="latestStatusTimestamp", order="desc", jobNameFilter=None)	None	Print jobs as a human-readable table.
get_job(job_id)	Job	Fetch current status and metadata for one job.
cancel_job(job_id)	None	Cancel a running job.

sortBy supports "latestStatusTimestamp" and "startTimestamp". order supports "asc" and "desc".

Running and monitoring pipelines

Method	Returns	Description
run(pipeline, show_progress=True, workflow_id=None)	Job	Create a workflow (unless workflow_id is given), run it to create a job, upload inputs, and return the job.
wait_for_completion(job_id, poll_interval=10, timeout=None, verbose=True)	Job	Poll until the job reaches a terminal status. Prints status updates and running logs when verbose=True.
run_and_wait(pipeline, output_dir=None, poll_interval=10, timeout=None, show_progress=True, workflow_id=None)	Job	Run a pipeline, wait for completion, and optionally download outputs on success.
get_job_logs(job_id, process_id)	str	Fetch formatted logs for a specific process in a job.
get_output_file_sizes(job_id)	list[dict]	Fetch output file size metadata for a completed job.
download_outputs(job_id, output_dir, process_id=None, show_progress=True)	list[str]	Download output files. With no process_id, each process is saved under output_dir/{processId}/. Batch outputs are nested under {batchId}/.

Tools: ToolConfig

ToolConfig describes one Biotailor tool. You usually get it from client.get_tool() or client.list_tools().

Important fields:

Field	Description
toolid	Tool identifier used by the API.
display_name	Human-readable tool name.
category	Tool category, such as quality control or alignment.
description	Optional long-form tool description.
has_sepe	Whether the tool supports both single-end and paired-end modes.
default_hardware	Default DefaultHardware with CPU and RAM settings.
parameters	List of ToolParameter objects.
outputs	List of Output objects.

Methods:

Method	Returns	Description
tool_guide()	str	Return a readable guide with tool info, ID labels, parameters, outputs, and sample pipeline code.
print_tool_guide()	None	Print the same guide returned by tool_guide().

Example:

tool = client.get_tool("bowtie2")
tool.print_tool_guide()

Use tool_guide() to find exact values for pipeline construction:

• toolId identifies the tool.
• parameterId is passed to set_param(), set_input(), set_input_batch(), or set_dataset().
• outputId is passed to step.output() or connect() for multi-process links.
• option value is the exact value passed to set_param() for option-style parameters.

Pipelines: Pipeline

Pipeline is a chainable builder for one tool run or a multi-process workflow. For a single tool, pass either a ToolConfig object or a tool ID string.

pipeline = Pipeline(name="my-job", tool=tool)

Builder methods:

Method	Returns	Description
set_param(param_id, value)	Pipeline	Set one non-file tool parameter.
set_params(**kwargs)	Pipeline	Set multiple non-file parameters at once.
set_input(param_id, file_path)	Pipeline	Set one local file input.
set_input_batch(param_id, file_paths)	Pipeline	Set multiple local files for a batch input.
set_dataset(param_id, dataset_type, dataset_id)	Pipeline	Use a platform dataset as an input source.
set_hardware(cpu, ram_gb)	Pipeline	Override default hardware. CPU must be one of the supported powers of two; RAM must be cpu * 2, cpu * 4, or cpu * 8.
set_pair_end()	Pipeline	Mark the run as paired-end. Required for tools that support both single-end and paired-end modes.
set_single_end()	Pipeline	Mark the run as single-end. Required for tools that support both single-end and paired-end modes.

Utility methods:

Method	Returns	Description
validate(available_datasets=None)	None	Validate file paths, hardware, required parameters, parameter value types, SE/PE mode compatibility, and optional dataset catalog matches.
build_workflow(available_datasets=None)	dict	Build the workflow JSON sent to the API. Calls validate() first.
get_file_map()	dict[str, str]	Return the local file map used by the uploader.
has_dataset_inputs()	bool	Return whether the pipeline uses any platform datasets.

For tools with has_sepe=True, always call either set_single_end() or set_pair_end() before running.

Building multi-process pipelines

A multi-process pipeline contains named process steps. Each step wraps one tool. To connect tools, pass a source step's outputId into a downstream file-input parameterId.

Use tool.print_tool_guide() to find the exact IDs before building the workflow:

• source tool outputId: used in source_step.output("outputId")
• target tool file-input parameterId: used in target_step.set_input("parameterId", ...)

Example with two connected processes:

fastp = client.get_tool("fastp-single")
bowtie2 = client.get_tool("bowtie2")

fastp.print_tool_guide()
bowtie2.print_tool_guide()

pipeline = Pipeline("trim-then-align")

trim = (
    pipeline.add_process("trim", fastp)
    .set_input("input", "reads.fastq")
    .set_param("qualified_quality_phred", 20)
)

align = (
    pipeline.add_process("align", bowtie2)
    .set_single_end()
    .set_dataset("x", "bowtie2_db", "bowtie2_yeast_r64_1_1")
    .set_input("-U", trim.output("filtered_reads"))
)

job = client.run_and_wait(pipeline, output_dir="./results")

The connection above means: use the filtered_reads output from process trim as the -U file input for process align.

Equivalent explicit connection syntax:

pipeline = Pipeline("trim-then-align")

pipeline.add_process("trim", fastp).set_input("input", "reads.fastq")
pipeline.add_process("align", bowtie2).set_single_end().set_dataset(
    "x", "bowtie2_db", "bowtie2_yeast_r64_1_1"
)
pipeline.connect("trim", "filtered_reads", "align", "-U")

For multi-process downloads, download_outputs(job_id, output_dir) creates one folder per process:

results/
  trim/
    ...trim outputs...
  align/
    ...alignment outputs...

To download only one process's outputs directly into a folder, pass process_id:

client.download_outputs(job.jobid, "./align-results", process_id="align")

Batch inputs

What is batch input?

Batch input lets you run the same process once per sample, in parallel, without building separate pipelines. Instead of calling set_input() with one file, call set_input_batch() with a list of local file paths. Biotailor submits one job that fans out over every file (or file pair) in the batch.

Each sample in the batch gets a batch ID derived from its filename. That ID is used to organize outputs on disk and to distinguish results when downloading.

Single-end batch input

Use set_input_batch() on a single file-input parameter. Each file in the list is treated as one independent single-end sample.

pipeline = Pipeline("trim-batch", fastp)
pipeline.set_single_end().set_input_batch(
    "--in1",
    ["sampleA_1.fastq", "sampleB_1.fastq"],
)

Rules:

• Call set_single_end() for tools that support both single-end and paired-end modes.
• Every filename must produce a unique batch ID.
• Batch ID = filename without extension (e.g. sampleA_1.fastq → sampleA_1).

Paired-end batch input

Use set_pair_end() and call set_input_batch() on both read-input parameters (--in1 and --in2, or the IDs shown in tool.print_tool_guide()). Both lists must have the same length — one R1/R2 pair per sample.

pipeline = Pipeline("trim-batch-pe", fastp)
pipeline.set_pair_end()
pipeline.set_input_batch("--in1", ["sampleA_1.fastq", "sampleB_1.fastq"])
pipeline.set_input_batch("--in2", ["sampleA_2.fastq", "sampleB_2.fastq"])

Rules:

• Call set_pair_end() before setting batch inputs.
• Both --in1 and --in2 lists must contain the same number of files.
• Each index across the two lists must represent the same sample (R1 at index 0 pairs with R2 at index 0, and so on).

Paired-end batch file naming convention

Paired-end batch inputs must follow a _1 / _2 suffix convention so Biotailor can match R1 and R2 files to the same sample:

Read	Filename pattern	Example
R1 (--in1)	{sample}_1.{ext}	sampleA_1.fastq
R2 (--in2)	{sample}_2.{ext}	sampleA_2.fastq

The batch ID is the shared sample prefix with the _1 / _2 suffix stripped:

• sampleA_1.fastq + sampleA_2.fastq → batch ID sampleA
• sampleB_1.fastq + sampleB_2.fastq → batch ID sampleB

Requirements enforced by the API:

• All R1 files must end with _1; all R2 files must end with _2 (or the reverse assignment across the two parameters).
• After stripping the suffix, batch IDs must be unique and must match across --in1 and --in2.
• Every sample must have both an R1 and an R2 file.

Example script

Multi-process paired-end batch: trim two samples with fastp, then align each trimmed output with bowtie2.

from biotailor import BiotailorClient, Pipeline

client = BiotailorClient(api_key="btk_xxx.sk_yyy")
fastp = client.get_tool("fastp")
bowtie2 = client.get_tool("bowtie2")

pipeline = Pipeline("sdk-fastp-bowtie2-batch")
trim = (
    pipeline.add_process("trim", fastp)
    .set_pair_end()
    .set_input_batch("--in1", ["inputA_1.fastq", "inputB_1.fastq"])
    .set_input_batch("--in2", ["inputA_2.fastq", "inputB_2.fastq"])
    .set_param("--qualified_quality_phred", 20)
)
pipeline.add_process("align", bowtie2).set_single_end().set_dataset(
    "x", "bowtie2_db", "bowtie2_yeast_r64_1_1"
).set_param("format", "fastq").set_input("-U", trim.output("output1"))

job = client.run_and_wait(pipeline, output_dir="./results")
print(f"Job {job.jobid} finished with status: {job.job_status.value}")

Single-end batch on one process:

from biotailor import BiotailorClient, Pipeline

client = BiotailorClient(api_key="btk_xxx.sk_yyy")
fastp = client.get_tool("fastp")

pipeline = (
    Pipeline("sdk-fastp-batch", fastp)
    .set_single_end()
    .set_input_batch("--in1", ["inputA_1.fastq", "inputA_2.fastq"])
    .set_param("--qualified_quality_phred", 20)
)

job = client.run_and_wait(pipeline, output_dir="./results")

Download output folder structure

When you pass output_dir to run_and_wait() or call download_outputs(), batch results are saved under {processId}/{batchId}/:

Paired-end batch (batch IDs inputA, inputB):

results/
  trim/
    inputA/
      output1
      report.html
      report.json
    inputB/
      output1
      report.html
      report.json
  align/
    inputA/
      alignedReads1.fastq
      output.sam
      unalignedReads1.fastq
    inputB/
      alignedReads1.fastq
      output.sam
      unalignedReads1.fastq

Single-end batch (batch IDs inputA_1, inputA_2):

results/
  trim/
    inputA_1/
      output1
      report.html
      report.json
    inputA_2/
      output1
      report.html
      report.json

For a single-process pipeline, omit the process folder — files go directly under {batchId}/ inside output_dir.

Models

The SDK uses dataclasses for API responses and tool definitions.

Model	Purpose	Key fields
Workflow	Reusable workflow template metadata.	workflow_id, workflow_name, userid, orgid, is_private, latest_updated_timestamp, email, react_flow_workflow, raw
Job	Job metadata and status.	jobid, workflow_id, job_name, job_status, processes_status, start_timestamp, latest_status_timestamp, end_timestamp, estimate_cost, processes_name, workflow, raw
JobStatus	Job status enum.	BUILDING, UPLOADING, STARTING, RUNNING, FAILED, SUCCEEDED, CANCELLED; use JobStatus.terminal_statuses() for terminal states.
ProcessStatus	Per-process status metadata.	process_id, status
ProcessState	Per-process status enum.	WAITING, STARTING, RUNNING, SUCCEEDED, FAILED
Dataset	Platform dataset metadata.	dataset_id, display_name, description, dataset_type, compression_method, s3uri, raw
ToolConfig	Full tool configuration.	toolid, display_name, category, has_sepe, default_hardware, parameters, outputs, description
ToolParameter	One configurable tool input or parameter.	id, display_name, type, description, default_value, validation, is_pair_end_only, is_single_end_only, alias
Validation	Validation rules for a parameter.	required, minimum, maximum, options, accepted_input_types, accepted_compression_method
Option	One allowed option for a parameter.	display_name, value, description
Output	One declared tool output.	id, display_name, description, file_type, file_name, compression_method, is_pair_end_only, is_single_end_only, alias
DefaultHardware	Default tool hardware.	cpu, ram_in_gb

Exceptions

All SDK exceptions inherit from BiotailorError.

Exception	When it is raised
BiotailorAPIError	The API returns a non-2xx response. Includes status_code, message, and response_body.
BiotailorValidationError	Client-side validation fails before a request or upload.
BiotailorUploadError	Uploading an input file fails.

Development

# Install in dev mode
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Lint
ruff check src/ tests/

Publishing to PyPI

# 1. Install build tools
pip install build twine

# 2. Clean old builds and build the package
rm -rf dist/
python -m build

# 4. Upload to PyPI
twine upload dist/*

Note: Bump version in pyproject.toml before each release — PyPI rejects duplicate versions. Use a PyPI API token for authentication (username: __token__).