BugSeq Python Client Library

Authentication and client code for interacting with the BugSeq API.

Installation

pip install bugseq-client

Example Usage

from pathlib import Path

import requests
from bugseq_client import Client
from bugseq_client.openapi_client import (
    AppModelsJobRunOptionsSampleType,
    JobRunSubmitRequest,
    Kit,
    MoleculeType,
    Platform,
    RunOptions,
)

# you should replace this with your lab ID
LAB_ID = "lab_123"


def submit_test_job(client: Client, file_ids: list[str]):
    job_run_submit_request = JobRunSubmitRequest(
        user_provided_name="My awesome analysis",
        file_ids=file_ids,
        run_options=RunOptions(
            platform=Platform.NANOPORE,
            kit=Kit.MINION_R10_4_1,
            sample_type=AppModelsJobRunOptionsSampleType.RESPIRATORY_UPPER,
            molecule_type=MoleculeType.DNA,
        ),
        lab_id=LAB_ID,
        testmode=True,
    )
    client.jobs.submit_analysis_v1_jobs_post(job_run_submit_request)


def download_url_to_file(download_url: str, dst: Path):
    with requests.get(download_url, stream=True) as r:
        r.raise_for_status()
        with open(dst, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                if chunk:  # filter out keep-alive chunks
                    f.write(chunk)


def fetch_and_download_job_results(client: Client):
    print("fetching jobs")
    jobs_resp = client.jobs.list_analyses_v1_jobs_get()
    for job in jobs_resp.job_runs:
        print(f"id={job.id} name={job.user_provided_name} status={job.job_status}")

    print()

    assert len(jobs_resp.job_runs) > 0, "No jobs found, no results to pull"

    job_id = jobs_resp.job_runs[0].id

    results_dir = Path(f"downloaded-results-{job_id}")

    print(f"fetching outputs for job {job_id} to {results_dir}")

    results_resp = client.jobs.get_analysis_results_v1_jobs_job_id_results_get(job_id)
    for output in results_resp.outputs:
        print(f"  output filename={output.filename} size={output.size}")

        if not output.filename.startswith("sample_reports/"):
            print("  does not match summary_reports filter, skipping")
            continue

        download_resp = (
            client.jobs.download_analysis_result_v1_jobs_job_id_results_download_get(
                job_id, output.filename
            )
        )

        download_dst = results_dir / output.filename
        download_dst.parent.mkdir(exist_ok=True, parents=True)
        download_url_to_file(download_resp.url, download_dst)
        print(f"  downloaded filename={output.filename} dst={str(download_dst)}")


def main():
    # Client.default() authenticates as a machine client if
    # BUGSEQ_MACHINE_CREDENTIALS_JSON or BUGSEQ_MACHINE_CREDENTIALS_FILE is
    # set, otherwise it falls back to Client.login(): reusing a valid
    # stored token, or opening a browser for the interactive login flow if
    # nothing is stored yet. Either way, the resulting token is refreshed
    # automatically for as long as the client is in use.
    with Client.default() as client:
        file_id = client.upload_file("sample.fastq.gz")
        submit_test_job(client, file_ids=[file_id])
        fetch_and_download_job_results(client)


if __name__ == "__main__":
    main()

Machine credentials

To authenticate as a machine client instead of a human user, download a credentials JSON file from the API Credentials page in the BugSeq web app, then either:

• point BUGSEQ_MACHINE_CREDENTIALS_FILE at it (or set BUGSEQ_MACHINE_CREDENTIALS_JSON to its contents directly) and call Client.default(), or
• call Client.from_machine_credentials_file("bugseq-credentials.json") directly.