Skip to content

oura

oura

Oura Ring connector — sleep, readiness, and activity via REST API v2.

Uses a Personal Access Token (PAT) stored in the connector config dir. All API calls are in module-level functions for easy mocking in tests.

Classes

OuraConnector

OuraConnector(*, token_path: str = _DEFAULT_TOKEN_PATH)

Bases: BaseConnector

Sync sleep, readiness, and activity data from Oura Ring.

Source code in src/openjarvis/connectors/oura.py
def __init__(self, *, token_path: str = _DEFAULT_TOKEN_PATH) -> None:
    self._token_path = Path(token_path)
    self._status = SyncStatus()
Functions
sync
sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]

Yield Documents for sleep, readiness, and activity.

Source code in src/openjarvis/connectors/oura.py
def sync(
    self, *, since: Optional[datetime] = None, cursor: Optional[str] = None
) -> Iterator[Document]:
    """Yield Documents for sleep, readiness, and activity."""
    token = self._load_token()
    start = (since or datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
    end = datetime.now().strftime("%Y-%m-%d")

    for data_type in ("sleep", "daily_readiness", "daily_activity"):
        data = _oura_api_get(
            token, data_type, params={"start_date": start, "end_date": end}
        )
        for item in data.get("data", []):
            day = item.get("day", start)
            yield Document(
                doc_id=f"oura-{data_type}-{day}",
                source="oura",
                doc_type=data_type,
                content=json.dumps(item),
                title=f"Oura {data_type.replace('_', ' ').title()}{day}",
                timestamp=datetime.fromisoformat(day),
                metadata={"data_type": data_type, "day": day},
            )

    self._status.state = "idle"
    self._status.last_sync = datetime.now()