Metadata-Version: 2.4
Name: bugseq-client
Version: 0.1.8
Summary: Python client for BugSeq (https://bugseq.com)
Project-URL: Homepage, https://gitlab.com/bugseq-open/bugseq-client-py
Project-URL: Issues, https://gitlab.com/bugseq-open/bugseq-client-py/-/issues
Author-email: Josh Chorlton <josh@bugseq.com>
License: Apache 2.0
License-File: LICENSE
Requires-Python: >=3.11
Requires-Dist: authlib>=1.6.1
Requires-Dist: keyring>=25.6.0
Requires-Dist: pydantic>=2.11.7
Requires-Dist: python-dateutil>=2.9.0
Requires-Dist: requests>=2.32.4
Provides-Extra: test
Requires-Dist: pytest-cov>=4.1.0; extra == 'test'
Requires-Dist: pytest>=8.0.0; extra == 'test'
Description-Content-Type: text/markdown

# BugSeq Python Client Library

Authentication and client code for interacting with the [BugSeq](https://bugseq.com) API.

## Installation

`pip install bugseq-client`

## Example Usage

```python
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.
