cloudos-cb-py 1.2.2

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 source

git clone https://github.com/lifebit-ai/cloudos-cb-py
cd cloudos-cb-py
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",
    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:

# Step 1: Submit
task = cloudos_cb.query_submit_async(
    cohort_id="1a2b3c4d5e6f7g8h9i10j11k",
    sql="SELECT person_id FROM omop_data.person",
    pagination={"pageNumber": 0, "pageSize": 100},
)
print("Task ID:", task["task_id"])

# Step 2: Poll
status = cloudos_cb.query_status(task_id=task["task_id"])
print("Status:", status["status"])
# status["status"] is one of: "pending", "running", "completed", "failed"

# Step 3: Fetch results when completed
df = cloudos_cb.query_results(task_id=task["task_id"])
print(df)

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="")

Submit an async SQL task. 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).

---

query_status(task_id, profilename="")

Check task status. 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="")

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

Attribute	Description
total_rows	Total rows across all pages
page	Page index returned
page_size	Rows in this page
total_pages	Total number of pages available

---

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

High-level orchestrator. Submits, polls, and fetches results automatically. When all_pages=True, submits one async task per 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",
    set_default=True,
)
cloudos_cb.configure(
    profilename="staging",
    apikey="stage-key",
    workspace_id="stage-workspace",
)

# 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