cloudos-cb-py 1.3.0

Python client for CloudOS Cohort Browser API

cloudos-cb-py

Python client for the CloudOS Cohort Browser API. Provides functions for schema discovery, table exploration, and SQL query execution with team-based access control.

Requirements

• Python >= 3.9
• requests >= 2.28.0
• pandas >= 1.5.0

Prerequisites

IMPORTANT: Before using this package, ensure the following requirements are met:

• Bastion must be enabled for your workspace
• You are running the package from within an interactive session
• The interactive session and the cohort queried must be in the same workspace

Without these prerequisites, API calls will fail even with valid credentials.

Installation

From PyPI (recommended)

pip install cloudos-cb-py

From PyPI source distribution

You can download and install the source distribution (.tar.gz) from PyPI:

# Download the latest source distribution from PyPI
pip download --no-binary :all: cloudos-cb-py

# This downloads a file like: cloudos-cb-py-1.2.2.tar.gz
# Extract it
tar -xzf cloudos-cb-py-*.tar.gz

# Navigate into the extracted folder
cd cloudos-cb-py-*/

# Install
pip install .

From GitHub (Private Repository - Requires Access)

Note: The GitHub repository is private and requires authentication. Only proceed if you have been granted access to the repository.

Option A: Install directly with authentication

Note: You'll need a GitHub Personal Access Token with repo scope. Generate one at: GitHub Settings → Developer settings → Personal access tokens → Tokens (classic)

# Replace YOUR_TOKEN with your GitHub Personal Access Token
pip install git+https://[e-mail]/lifebit-ai/cloudos-cb-py.git

Option B: Download and install manually

If you have access to the private GitHub repository, follow these steps:

Step 1: Download the Package from GitHub

1. Navigate to the GitHub repository: https://github.com/lifebit-ai/cloudos-cb-py
2. Click the green "Code" button
3. Select "Download ZIP" from the dropdown menu
4. Save the ZIP file to a location on your computer (e.g., your Downloads folder)

Step 2: Extract the ZIP File

# Navigate to where you downloaded the ZIP
cd ~/Downloads

# Extract the ZIP file (the exact name may vary, e.g., cloudos-cb-py-main.zip)
unzip cloudos-cb-py-main.zip

# Navigate into the extracted folder
cd cloudos-cb-py-main

Step 3: Install from the Extracted Folder

# Install from the current directory
pip install .

Development install (includes test dependencies)

pip install -e ".[dev]"

Quick Start

1. Configure a profile

import cloudos_cb

cloudos_cb.configure(
    profilename="production",
    apikey="your-api-key-here",
    workspace_id="953h453uhr73894hhr9348h9",
    base_url="https://cloudos.lifebit.ai",
    set_default=True,
)

Credentials are stored in ~/.cloudos-cb/config.json with 0600 permissions. Set CLOUDOS_CONFIG_DIR to store the file elsewhere.

2. List configured profiles

profiles = cloudos_cb.profile_list()
print(profiles)
# Returns a pandas DataFrame with columns:
# profile_name, workspace_id, base_url, default, created_at, updated_at

3. Discover cohort tables

tables = cloudos_cb.cohort_tables(cohort_id="1a2b3c4d5e6f7g8h9i10j11k")
print(tables)
# Cohort 1a2b3c4d5e6f7g8h9i10j11k:
#   - omop_data.person
#       - person_id (integer)
#       - year_of_birth (integer)
#       - gender_concept_id (integer)
#       ...
#   - omop_data.observation
#       ...
#
# Total: 1 database(s), 5 table(s)

# Access raw data
schema_list = tables.schemas

4. Validate SQL (optional but recommended)

result = cloudos_cb.sql_validate(
    sql="SELECT person_id FROM omop_data.person WHERE year_of_birth >= 1960"
)

if result["isValid"]:
    print("SQL is valid")
else:
    print("SQL invalid:", result["error"]["message"])

5. Execute a query (high-level)

df = cloudos_cb.query(
    cohort_id="1a2b3c4d5e6f7g8h9i10j11k",
    sql="SELECT person_id, gender_concept_id FROM omop_data.person LIMIT 100",
)
print(df.head())
print(f"Total rows: {df.attrs['total_rows']}")

By default query() fetches all pages automatically. To return only the first page:

df = cloudos_cb.query(
    cohort_id="1a2b3c4d5e6f7g8h9i10j11k",
    sql="SELECT person_id FROM omop_data.person",
    all_pages=False,
    page_size=500,
)

6. Manual workflow

For fine-grained control over the submit / poll / fetch cycle. The data and count operations are served by two separate endpoints, so the manual workflow submits both and combines them — this reproduces exactly what query() returns, including the total_rows/total_pages metadata.

import time

cohort_id = "1a2b3c4d5e6f7g8h9i10j11k"
sql = "SELECT person_id FROM omop_data.person"

# Step 1: Submit a count task and a data task
count_task = cloudos_cb.query_submit_count_async(cohort_id=cohort_id, sql=sql)
data_task = cloudos_cb.query_submit_async(
    cohort_id=cohort_id,
    sql=sql,
    pagination={"pageNumber": 0, "pageSize": 100},
)

# Step 2: Poll each task until completed before fetching.
# status is one of: "pending", "running", "completed", "failed". Normalise the
# case (the server may return e.g. "Completed") before comparing.
# Bound the wait, and surface "failed" immediately instead of looping on it.
max_wait, poll_interval = 600, 2
for task in (data_task, count_task):
    deadline = time.monotonic() + max_wait
    while True:
        status = cloudos_cb.query_status(task_id=task["task_id"])["status"].lower().strip()
        if status == "completed":
            break
        if status == "failed":
            raise RuntimeError(f"Task {task['task_id']} failed")
        if time.monotonic() >= deadline:
            raise TimeoutError(f"Task {task['task_id']} did not complete in {max_wait}s")
        time.sleep(poll_interval)

# Step 3: Fetch the total from the count task, then the data page. Passing
# total_rows lets query_results() populate total_rows/total_pages for you —
# you never compute pages or set .attrs by hand.
total = cloudos_cb.query_count_results(task_id=count_task["task_id"])
df = cloudos_cb.query_results(task_id=data_task["task_id"], total_rows=total)

print(df)
print(f"Total rows: {df.attrs['total_rows']}, pages: {df.attrs['total_pages']}")

Why two tasks? The data endpoint (query_submit_async) returns rows only and no longer reports a total, for performance. The count is computed by a separate task. If you fetch data without passing total_rows, then df.attrs["total_rows"] and total_pages are left as None. The shortcut cloudos_cb.query_count(cohort_id, sql) runs the full submit→poll→fetch count cycle in one call. The high-level query() does all of this for you.

API Reference

configure(profilename, apikey, workspace_id, base_url=..., set_default=False)

Create or update a named credential profile.

Parameter	Type	Description
profilename	str	Profile name (required)
apikey	str	API key (required)
workspace_id	str	Workspace/team ID (required)
base_url	str	CloudOS base URL (default: https://cloudos.lifebit.ai)
set_default	bool	Mark this profile as the default

---

profile_list()

Return a pandas.DataFrame of all configured profiles.

---

cohort_tables(cohort_id, profilename="")

Retrieve schemas, tables, and columns for a cohort.

Returns a CohortTables object. Print it for a human-readable tree, or access .schemas for the raw list.

---

sql_validate(sql, profilename="")

Validate SQL syntax and references before execution.

Returns a dict with isValid (bool), tableReferences, columnReferences, and on failure an error dict with a message key.

---

query_submit_async(cohort_id, sql, pagination=None, profilename="", *, cursor=None)

Submit an async SQL data task (the query-results/data/async endpoint). Returns rows only — the total row count is no longer included. Returns a dict with:

Key	Description
task_id	Use this to poll status and fetch results
status	Initial status (typically "pending")
query	Echo of the submitted SQL
type	Task type string
sync_execution_timeout	Server-side timeout hint in ms
full_response	Raw API response

pagination is an optional dict with pageNumber (int >= 0) and pageSize (int >= 1). cursor is an optional opaque cursor (from a previous page's .attrs["cursor"]) for cursor-based pagination.

---

query_submit_count_async(cohort_id, sql, profilename="")

Submit an async SQL count task (the query-results/count/async endpoint). The completed task's result holds the total row count and no data rows. Returns the same dict shape as query_submit_async(). Fetch the total with query_count_results().

---

query_status(task_id, profilename="")

Check task status (works for both data and count tasks). Returns a dict with task_id, status, type, count_of_results, query, created_at, started_at, ended_at, user, full_response.

---

query_results(task_id, profilename="", *, total_rows=None)

Fetch data results for a completed data task. Returns a pandas.DataFrame with metadata in .attrs:

Attribute	Description
total_rows	Total rows across all pages, or None (the data endpoint omits it — pass total_rows)
page	Page index returned
page_size	Rows in this page
total_pages	Total number of pages, or None when total_rows is unknown
cursor	Opaque cursor for the next page, when provided by the server

Pass total_rows (from query_count() / query_count_results()) to populate total_rows and total_pages using the same logic as query(), so a manual workflow reproduces the high-level result exactly.

---

query_count_results(task_id, profilename="")

Fetch the total row count (an int) from a completed count task submitted via query_submit_count_async().

---

query_count(cohort_id, sql, poll_interval=2, max_wait=600, profilename="")

High-level helper that submits a count task, polls it to completion, and returns the total number of rows matching the query as an int.

---

query(cohort_id, sql, poll_interval=2, max_wait=600, page_size=1000, all_pages=True, profilename="")

High-level orchestrator. Submits the count task and the page-0 data task together, polls them, and derives total_rows/total_pages from the count. When all_pages=True, submits one data task per remaining page and concatenates them.

Parameter	Default	Description
poll_interval	2	Seconds between status checks (minimum 1)
max_wait	600	Maximum seconds to wait per task
page_size	1000	Rows per page
all_pages	True	Fetch all pages and combine them

---

Using multiple profiles

# Configure multiple profiles
cloudos_cb.configure(
    profilename="production",
    apikey="prod-key",
    workspace_id="prod-workspace",
    base_url="https://cloudos.lifebit.ai",
    set_default=True,
)
cloudos_cb.configure(
    profilename="staging",
    apikey="stage-key",
    workspace_id="stage-workspace",
    base_url="https://cloudos.lifebit.ai",
)

# Use default profile (production)
df = cloudos_cb.query(cohort_id="cohort-prod", sql="SELECT 1")

# Explicitly use staging profile
df = cloudos_cb.query(
    cohort_id="cohort-stage",
    sql="SELECT 1",
    profilename="staging",
)

Configuration storage

The config file is located at:

• $CLOUDOS_CONFIG_DIR/config.json when the env var is set
• ~/.cloudos/config.json otherwise (home directory)

File permissions are set to 0600 (user read/write only). The default location (~/.cloudos/) is outside any repository. If you override CLOUDOS_CONFIG_DIR to a path inside a project, add that directory to your .gitignore.

Error handling

from cloudos_cb import (
    CloudOSAuthError,
    CloudOSAccessError,
    CloudOSServerError,
    CloudOSConfigError,
    CloudOSValidationError,
)

try:
    df = cloudos_cb.query(cohort_id="...", sql="SELECT 1")
except CloudOSAuthError:
    print("Authentication failed - check your API key.")
except CloudOSAccessError:
    print("Access denied or resource not found.")
except CloudOSServerError:
    print("Server error - try again later.")
except CloudOSConfigError:
    print("Profile not configured - run configure() first.")
except CloudOSValidationError as e:
    print(f"Invalid input: {e}")

Logging

The package uses Python's standard logging module under the cloudos_cb namespace. To see informational messages:

import logging
logging.basicConfig(level=logging.INFO)

Running tests

pip install -e ".[dev]"
pytest

To check code style:

flake8 cloudos_cb tests

Package structure

cloudos-cb-py/
├── pyproject.toml        # Package metadata and build config
├── CHANGELOG.md
├── README.md
├── LICENSE
├── cloudos_cb/           # Package source
│   ├── __init__.py       # Public API
│   ├── exceptions.py     # Custom exception classes
│   ├── config.py         # Profile management
│   ├── http.py           # Authenticated HTTP helpers
│   ├── utils.py          # Shared utilities
│   └── queries.py        # Cohort Browser query functions
└── tests/
    ├── test_config.py
    ├── test_http.py
    ├── test_utils.py
    └── test_query.py