Low-Level API Access

The SDK's high-level Client covers the most common workflows — hierarchy traversal, file I/O, metadata updates, and more. When you need more control — to call endpoints the client does not expose, inspect raw HTTP responses, or extend the client with custom logic — you work directly with the underlying ApiClient and model layer. This page documents those lower-level primitives.

Direct API Calls

When to use the raw API

The high-level fw.<method>() calls satisfy the vast majority of use cases. Drop to the raw ApiClient when you need to:

• Call an endpoint the high-level client does not expose
• Inspect HTTP status codes or response headers alongside the response body
• Send a one-off custom header or query parameter without changing global defaults
• Receive the raw bytes from a response without SDK deserialization

The ApiClient instance for an existing client is accessible via fw.api_client.

Using call_api() directly

ApiClient.call_api() accepts any resource path, HTTP method, and optional request body. Pass a model class name as a string for response_type to trigger automatic deserialization, or pass None to skip deserialization entirely. After every call_api() invocation — including those made internally by high-level fw.* methods — api_client.last_response holds the raw RESTResponse; its .data attribute is the response body as bytes, which is the most direct way to read the body when response_type=None is passed.

import json

api_client = fw.api_client

_, status, headers = api_client.call_api(
    "/config",
    "GET",
    response_type=None,
    auth_settings=["ApiKey"],
    _return_http_data_only=False,
)

raw = json.loads(api_client.last_response.data)
print(f"HTTP {status} — site: {raw.get('site', {}).get('name')}")

Setting _return_http_data_only=False makes call_api() return a three-tuple of (data, status_code, headers) instead of just the deserialized data. Use this when you need to branch on the HTTP status code or read a response header.

data, status, headers = api_client.call_api(
    "/users/self",
    "GET",
    response_type="User",
    auth_settings=["ApiKey"],
    _return_http_data_only=False,
)

print(f"Headers: {headers}")
print(f"User: {data.id}")

Pass additional query parameters or headers directly to call_api() for a single call. These do not affect the global defaults set on the ApiClient instance.

result, status, _ = api_client.call_api(
    "/sessions",
    "GET",
    query_params=[("limit", "10"), ("filter", "label=baseline")],
    header_params={"X-Debug-Trace": "my-script"},
    response_type="list[SessionOutput]",
    auth_settings=["ApiKey"],
    _return_http_data_only=False,
)

print(f"Returned {len(result)} sessions with status {status}")

Response parsing

By default, call_api() parses the response body and returns a typed SDK model. Set _preload_content=False to skip body loading and receive the raw requests.Response directly. This is useful when you need to stream a large binary response rather than hold it all in memory.

Some endpoints use a ticket-based flow: a POST returns a JSON ticket and a subsequent GET with that ticket delivers the file bytes. api_client.last_response.data holds the raw bytes of the most recent response, which is convenient for reading the ticket without a separate response_type parameter.

import json

# Step 1: POST to /download returns a JSON ticket, not the file.
api_client.call_api(
    "/download",
    "POST",
    body={"nodes": [{"level": "session", "_id": session_id}]},
    auth_settings=["ApiKey"],
    _return_http_data_only=True,
)
ticket = json.loads(api_client.last_response.data)["ticket"]

# Step 2: GET with the ticket streams the actual file bytes.
raw_resp = api_client.call_api(
    "/download",
    "GET",
    query_params=[("ticket", ticket)],
    auth_settings=["ApiKey"],
    _preload_content=False,
    _return_http_data_only=True,
)

with open("download.tar", "wb") as f:
    for chunk in raw_resp.iter_content(chunk_size=65536):
        f.write(chunk)

Custom API Client

Adding custom methods

For a single endpoint that does not warrant a full subclass, assign a method to an existing client instance at runtime. Python's types.MethodType binds a function to an object so that self resolves correctly, giving the function access to the client's _fw.api_client.

GET /readertasks/project/{project_id} is not exposed by the high-level client. Attaching it as a runtime method lets you call it on any existing flywheel.Client instance without defining a subclass.

import json
import types


def get_reader_tasks(self, project_id, filter=None):
    query_params = []
    if filter is not None:
        query_params.append(("filter", filter))

    self._fw.api_client.call_api(
        f"/readertasks/project/{project_id}",
        "GET",
        query_params=query_params,
        auth_settings=["ApiKey"],
        _return_http_data_only=True,
    )
    return json.loads(self._fw.api_client.last_response.data)


fw.get_reader_tasks = types.MethodType(get_reader_tasks, fw)

todo = fw.get_reader_tasks(project_id, filter="status=Todo")
for task in todo["results"]:
    print(task["_id"])

The filter parameter accepts comma-separated expressions in the same format as other Flywheel API filters — for example, "status=Todo,assignee=user@example.com".

Extending the client

When you need to add several related methods for the same resource, a subclass keeps them organized and makes the intent clear. Client delegates unknown attribute lookups to the internal Flywheel instance via __getattr__, so a subclass retains full access to all existing SDK methods alongside the new ones.

Multiple legacy reader task endpoints are a natural set to bundle into an ExtendedClient:

import json
import flywheel


class ExtendedClient(flywheel.Client):
    def get_reader_tasks(self, project_id, filter=None):
        query_params = []
        if filter is not None:
            query_params.append(("filter", filter))

        self._fw.api_client.call_api(
            f"/readertasks/project/{project_id}",
            "GET",
            query_params=query_params,
            auth_settings=["ApiKey"],
            _return_http_data_only=True,
        )
        return json.loads(self._fw.api_client.last_response.data)

    def get_reader_task(self, task_id):
        self._fw.api_client.call_api(
            f"/readertasks/{task_id}",
            "GET",
            auth_settings=["ApiKey"],
            _return_http_data_only=True,
        )
        return json.loads(self._fw.api_client.last_response.data)

With all needed methods on the class, the workflow stays on a single client instance and the methods self-document as a unit:

fw = ExtendedClient(api_key="your-api-key")

page = fw.get_reader_tasks(project_id, filter="status=Todo")
if page["results"]:
    task_ids = [task["_id"] for task in page["results"]]
    first_task = fw.get_reader_task(task_ids[0])
    print(first_task.get("assignee"))

Middleware patterns

ApiClient exposes two methods for attaching persistent request modifications:

Use set_default_header() to inject a tracing identifier or a custom User-Agent without repeating the header on every individual call.

import uuid

api_client = fw.api_client
api_client.set_default_header("X-Script-Run-Id", str(uuid.uuid4()))
api_client.user_agent = "my-pipeline/1.0 flywheel-sdk"

set_default_query_param() is lower-level and appended unconditionally — it does not merge with existing query parameters of the same name, so avoid calling it more than once for the same parameter key.

Request and response interceptors

ApiClient supports a one-shot version check function via set_version_check_fn(). The function is called before the next request is dispatched and is cleared afterwards.

def assert_minimum_version():
    version = fw.get_version()
    if version.release < "22.0":
        raise RuntimeError(f"Requires Flywheel >= 22.0, got {version.release}")

fw.api_client.set_version_check_fn(assert_minimum_version)

For persistent intercept logic — such as logging all request URLs or collecting timing metrics — subclass ApiClient and override call_api(), delegating to super() after your logic runs.

import time
from flywheel.api_client import ApiClient


class LoggingApiClient(ApiClient):
    def call_api(self, resource_path, method, *args, **kwargs):
        start = time.monotonic()
        result = super().call_api(resource_path, method, *args, **kwargs)
        elapsed = time.monotonic() - start
        print(f"{method} {resource_path} completed in {elapsed:.3f}s")
        return result

To use a custom ApiClient subclass, initialize flywheel.Client as normal and then replace its api_client. Copying the configuration from the original carries over authentication and host settings automatically.

import flywheel

fw = flywheel.Client()
config = fw.api_client.configuration
fw.api_client = LoggingApiClient(configuration=config)

Working with Raw Models

What SDK models are

SDK models are auto-generated Python classes produced from the OpenAPI specification by the Swagger code generator. They are not Pydantic models. Each model class carries two class-level dictionaries that describe its schema:

session = fw.get_session(session_id)

for attr, type_str in session.swagger_types.items():
    json_key = session.attribute_map[attr]
    print(f"{attr} ({type_str}) -> JSON key: {json_key}")

Constructing models directly

Construct any SDK model by importing it from flywheel and passing keyword arguments. Only fields with non-None values are serialized when the model is sent to the API.

session_input = flywheel.SessionInput(
    label="Baseline",
    subject=flywheel.SessionEmbeddedSubject(label=subject_label, age=30),
    project=project_id,
)

new_session_id = fw.add_session(session_input)

Type safety with model attributes

The swagger_types dict documents the expected Python type for each attribute. Assigning a value of the wrong type does not raise an error at assignment time — the mismatch only surfaces when the model is serialized or when the server rejects the request.

session_input = flywheel.SessionInput()

for attr, expected_type in session_input.swagger_types.items():
    print(f"  {attr}: {expected_type}")

Use this introspection when building models dynamically from external data sources to validate field types before making API calls.

EXPECTED_TYPES = flywheel.SessionInput.swagger_types

def validate_session_payload(payload):
    for key, value in payload.items():
        expected = EXPECTED_TYPES.get(key)
        if not isinstance(value, expected)
            raise TypeError(
                f"Field '{key}' must be of type '{expected}', got {type(value).__name__}"
            )

Model serialization

ApiClient.sanitize_for_serialization() converts a model instance — or any nested structure of models, lists, and dicts — into a plain Python dict suitable for JSON encoding. It follows nested models recursively and omits attributes whose value is None.

import json

api_client = fw.api_client

session = fw.get_session(session_id)

session_dict = api_client.sanitize_for_serialization(session)
print(json.dumps(session_dict, default=str, indent=2))

The output uses the JSON keys from attribute_map — for example, the Python attribute id maps to the JSON key _id.

Model deserialization

Pass a response_type string to call_api() to receive a fully deserialized model object. The response_type must match a class name exported from flywheel.models.

session = api_client.call_api(
    f"/sessions/{session_id}",
    "GET",
    response_type="SessionOutput",
    auth_settings=["ApiKey"],
    _return_http_data_only=True,
)

print(type(session).__name__)  # SessionOutput
print(session.label)

For list responses, wrap the type string in list[...]:

sessions = api_client.call_api(
    "/sessions",
    "GET",
    query_params=[("limit", "10")],
    response_type="list[SessionOutput]",
    auth_settings=["ApiKey"],
    _return_http_data_only=True,
)

for session in sessions:
    print(session.label)