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()
|