Session

The Session object provides a session-based alternative to the global initialize() pattern. Instead of initializing global singleton clients, you create a Session instance that holds its own API clients and exposes all resource operations as namespaced methods.

This is useful when you need to:

Work with multiple configurations (e.g. different organizations) in the same process
Avoid global state for cleaner dependency management
Use the SDK in applications where global singletons are undesirable

Note

You should use either the global initialize() pattern or session-based clients, not both. If runeq has already been globally initialized when you create a Session, a warning will be raised.

Getting Started

Create a Config and pass it to Session:

from runeq import Config, Session

config = Config()
session = Session(config)

All resource operations are available as namespaced methods on the session object:

# Organization metadata
orgs = session.org.get_orgs()
session.org.set_active_org(org_id)

# Patient metadata
patients = session.patient.get_all_patients()
patient = session.patient.get_patient(patient_id)
devices = session.patient.get_all_devices()

# User metadata
user = session.user.get_current_user()

# Project metadata
projects = session.project.get_projects()
project_patients = session.project.get_project_patients(project_id)

# Stream metadata and data
stream_metadata = session.stream_metadata.get_patient_stream_metadata(patient_id)
stream_df = session.stream_metadata.get_stream_dataframe(stream_ids, start_time=..., end_time=...)

# Raw stream data
data = session.stream.get_stream_data(stream_id, start_time=..., end_time=...)

# Events
events = session.event.get_patient_events(patient_id, start_time=..., end_time=...)

# Sleep metrics
from datetime import date
sleep = session.sleep.get_sleep_metrics(patient_id, start_date=date(2025, 1, 1), end_date=date(2025, 1, 10))

Namespaces

The session organizes methods into the following namespaces, each corresponding to a resource module:

session.org — Organization metadata (org)
session.patient — Patient and device metadata (patient)
session.project — Project and cohort metadata (project)
session.user — User metadata (user)
session.event — StrivePD events (event)
session.stream_metadata — Stream metadata and dataframes (stream_metadata)
session.stream — Raw stream data and availability (stream)
session.sleep — Sleep metrics (sleep)

Each namespace method has the same signature as the corresponding module-level function, minus the client parameter (which is provided by the session automatically).

Note

Object-level methods that make API calls (e.g. get_stream_dataframe(), get_batch_availability_dataframe()) are not yet supported in a session-based context. Use the corresponding namespace methods on the session instead.

API Reference

class runeq.Session(config: BaseConfig)

Session object to access Rune APIs.

__init__(config: BaseConfig)

event: EventNamespace

graph_client: GraphClient

org: OrgNamespace

patient: PatientNamespace

project: ProjectNamespace

sleep: SleepNamespace

stream: StreamNamespace

stream_client: StreamClient

stream_metadata: StreamMetadataNamespace

strive_client: StriveClient

user: UserNamespace

OrgNamespace

class runeq.resources.session.namespaces.OrgNamespace(client: GraphClient)

client: GraphClient

get_org(org_id: str) → Org

get_orgs() → OrgSet

set_active_org(org: Union[str, Org]) → Org

PatientNamespace

class runeq.resources.session.namespaces.PatientNamespace(client: GraphClient)

client: GraphClient

get_all_devices(patients: Union[PatientSet, List[str]] = None) → DeviceSet

get_all_patients() → PatientSet

get_device(patient: Union[Patient, str], device_id: str) → Device

get_patient(patient_id: str) → Patient

get_patient_devices(patient: Union[Patient, str]) → DeviceSet

ProjectNamespace

class runeq.resources.session.namespaces.ProjectNamespace(client: GraphClient)

client: GraphClient

get_cohort_patients(cohort_id: str) → ProjectPatientMetadataSet

get_project(project_id: str) → Project

get_project_patients(project_id: str) → ProjectPatientMetadataSet

get_projects() → ProjectSet

UserNamespace

class runeq.resources.session.namespaces.UserNamespace(client: GraphClient)

client: GraphClient

get_current_user() → User

EventNamespace

class runeq.resources.session.namespaces.EventNamespace(client: GraphClient)

client: GraphClient

get_patient_activity_events(patient_id: str, start_time: Union[int, float, date, datetime], end_time: Union[int, float, date, datetime]) → EventSet

get_patient_events(patient_id: str, start_time: Union[int, float, date, datetime], end_time: Union[int, float, date, datetime], include_filters: Optional[list] = None) → EventSet

get_patient_medication_events(patient_id: str, start_time: Union[int, float, date, datetime], end_time: Union[int, float, date, datetime]) → EventSet

get_patient_symptom_events(patient_id: str, start_time: Union[int, float, date, datetime], end_time: Union[int, float, date, datetime]) → EventSet

get_patient_wellbeing_events(patient_id: str, start_time: Union[int, float, date, datetime], end_time: Union[int, float, date, datetime]) → EventSet

StreamMetadataNamespace

class runeq.resources.session.namespaces.StreamMetadataNamespace(graph_client: GraphClient, stream_client: StreamClient)

get_all_stream_types() → StreamTypeSet

get_patient_stream_metadata(patient_id: str, device_id: Optional[str] = None, stream_type_id: Optional[str] = None, algorithm: Optional[str] = None, category: Optional[str] = None, measurement: Optional[str] = None, **parameters) → StreamMetadataSet

get_stream_availability_dataframe(stream_ids: Union[str, Iterable[str]], start_time: Union[float, date], end_time: Union[float, date], resolution: int, batch_operation: Optional[str] = None, limit: Optional[int] = None, page_token: Optional[str] = None, timestamp: Optional[str] = 'iso', timezone: Optional[int] = None) → DataFrame

get_stream_dataframe(stream_ids: Union[str, Iterable[str]], start_time: Optional[Union[float, date]] = None, start_time_ns: Optional[int] = None, end_time: Optional[Union[float, date]] = None, end_time_ns: Optional[int] = None, limit: Optional[int] = None, page_token: Optional[str] = None, timestamp: Optional[str] = 'iso', timezone: Optional[int] = None, translate_enums: Optional[bool] = True) → DataFrame

get_stream_metadata(stream_ids: Union[str, Iterable[str]]) → Union[StreamMetadata, StreamMetadataSet]

graph_client: GraphClient

stream_client: StreamClient

StreamNamespace

class runeq.resources.session.namespaces.StreamNamespace(client: StreamClient)

client: StreamClient

get_stream_aggregate_window(stream_id: str, start_time: Union[int, float, date, datetime], end_time: Union[int, float, date, datetime], resolution: int, aggregate_function: Literal['sum', 'mean'], timestamp: Optional[str] = 'iso', timezone: Optional[int] = None, timezone_name: Optional[str] = None) → dict

get_stream_availability(stream_ids: Union[str, Iterable[str]], start_time: Union[int, float, date, datetime], end_time: Union[int, float, date, datetime], resolution: int, batch_operation: Optional[str] = None, format: Optional[str] = 'csv', limit: Optional[int] = None, page_token: Optional[str] = None, timestamp: Optional[str] = 'iso', timezone: Optional[int] = None, timezone_name: Optional[str] = None) → Iterator[Union[str, dict]]

get_stream_daily_aggregate(stream_id: str, start_time: Union[int, float, date, datetime], resolution: int, n_days: int) → dict

get_stream_data(stream_id: str, start_time: Optional[Union[int, float, date, datetime]] = None, start_time_ns: Optional[int] = None, end_time: Optional[Union[int, float, date, datetime]] = None, end_time_ns: Optional[int] = None, format: Optional[str] = 'csv', limit: Optional[int] = None, page_token: Optional[str] = None, timestamp: Optional[str] = 'iso', timezone: Optional[int] = None, timezone_name: Optional[str] = None, translate_enums: Optional[bool] = True) → Iterator[Union[str, dict]]

SleepNamespace

class runeq.resources.session.namespaces.SleepNamespace(client: StriveClient)

client: StriveClient

get_sleep_metrics(patient_id: str, start_date: date, end_date: date) → List[Dict]