Open Wearables Integration Guide

OPEN_WEARABLES_API_URL=http://localhost:8000
OPEN_WEARABLES_API_KEY=sk-your-api-key-here

curl http://localhost:8000/api/v1/users \
  -H "X-Open-Wearables-API-Key: YOUR_API_KEY"

curl -X POST http://localhost:8000/api/v1/applications \
  -H "Authorization: Bearer YOUR_DEVELOPER_JWT" \
  -H "Content-Type: application/json" \
  -d '{"name": "My iOS App"}'

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "app_id": "app_abc123",
  "name": "My iOS App",
  "created_at": "2025-01-15T10:30:00Z",
  "app_secret": "secret_xyz789..."  // Store securely! Only shown once.
}

curl -X POST http://localhost:8000/api/v1/users/{user_id}/token \
  -H "Content-Type: application/json" \
  -d '{
    "app_id": "app_abc123",
    "app_secret": "secret_xyz789..."
  }'

{
  "access_token": "eyJhbGciOiJIUzI1NiIs...",
  "token_type": "bearer"
}

{
  "id": "176be8de-8452-4eb7-a7ea-147fec925d9d",
  "email": "user@example.com",
  "first_name": "Ada",
  "last_name": "Lovelace",
  "created_at": "2025-01-15T10:30:00Z"
}

[
  {
    "provider": "garmin",
    "name": "Garmin",
    "has_cloud_api": true,
    "is_enabled": true,
    "icon_url": "/static/provider-icons/garmin.svg"
  },
  {
    "provider": "polar",
    "name": "Polar",
    "has_cloud_api": true,
    "is_enabled": true,
    "icon_url": "/static/provider-icons/polar.svg"
  }
]

{
  "authorization_url": "https://connect.garmin.com/oauthConfirm?oauth_token=...",
  "state": "abc123..."
}

curl "http://localhost:8000/api/v1/oauth/garmin/authorize?user_id=USER_ID&redirect_uri=https://yourapp.com/oauth/callback" \
  -H "X-Open-Wearables-API-Key: YOUR_API_KEY"

curl "http://localhost:8000/api/v1/users/176be8de-8452-4eb7-a7ea-147fec925d9d/connections" \
  -H "X-Open-Wearables-API-Key: YOUR_API_KEY"

[
  {
    "id": "a1b2c3d4-...",
    "user_id": "176be8de-8452-4eb7-a7ea-147fec925d9d",
    "provider": "garmin",
    "status": "active",
    "provider_user_id": "12345678",
    "created_at": "2025-01-15T10:30:00Z",
    "updated_at": "2025-01-15T10:30:00Z",
    "last_synced_at": null
  }
]

{
  "data": [ /* items */ ],
  "pagination": {
    "next_cursor": "eyJpZCI6...",
    "previous_cursor": null,
    "has_more": true,
    "total_count": 150
  },
  "metadata": {
    "resolution": null,
    "sample_count": 50,
    "start_time": "2025-01-01T00:00:00Z",
    "end_time": "2025-01-31T23:59:59Z"
  }
}

import httpx
from typing import Literal

class OpenWearablesClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.headers = {"X-Open-Wearables-API-Key": api_key}

    async def get_or_create_user(self, local_user) -> dict:
        """Create an Open Wearables user if we haven't already, otherwise reuse the stored ID.

        `local_user` is your own user row - it stores `open_wearables_user_id` once set
        so we never create a second Open Wearables user for the same person.
        """
        async with httpx.AsyncClient() as client:
            if local_user.open_wearables_user_id:
                resp = await client.get(
                    f"{self.base_url}/api/v1/users/{local_user.open_wearables_user_id}",
                    headers=self.headers,
                )
                if resp.status_code == 200:
                    return resp.json()

            resp = await client.post(
                f"{self.base_url}/api/v1/users",
                headers=self.headers,
                json={"email": local_user.email},
            )
            user = resp.json()
            local_user.open_wearables_user_id = user["id"]  # persist on your side
            return user

    async def get_auth_url(
        self,
        provider: Literal["garmin", "polar", "suunto"],
        user_id: str,
        redirect_uri: str | None = None,
    ) -> str:
        """Get OAuth authorization URL."""
        params = {"user_id": user_id}
        if redirect_uri:
            params["redirect_uri"] = redirect_uri

        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.base_url}/api/v1/oauth/{provider}/authorize",
                headers=self.headers,
                params=params,
            )
            return resp.json()["authorization_url"]

    async def get_connections(self, user_id: str) -> list[dict]:
        """Get user's connected providers."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.base_url}/api/v1/users/{user_id}/connections",
                headers=self.headers,
            )
            return resp.json()

    async def _paginated(self, path: str, params: dict) -> list[dict]:
        """Walk `next_cursor` until `has_more` is false and return all items."""
        items: list[dict] = []
        async with httpx.AsyncClient() as client:
            while True:
                resp = await client.get(f"{self.base_url}{path}", headers=self.headers, params=params)
                payload = resp.json()
                items.extend(payload.get("data", []))
                cursor = payload["pagination"].get("next_cursor")
                if not payload["pagination"].get("has_more") or not cursor:
                    return items
                params = {**params, "cursor": cursor}

    async def get_activity_summary(self, user_id: str, start_date: str, end_date: str) -> list[dict]:
        return await self._paginated(
            f"/api/v1/users/{user_id}/summaries/activity",
            {"start_date": start_date, "end_date": end_date},
        )

    async def get_sleep_summary(self, user_id: str, start_date: str, end_date: str) -> list[dict]:
        return await self._paginated(
            f"/api/v1/users/{user_id}/summaries/sleep",
            {"start_date": start_date, "end_date": end_date},
        )

    async def get_body_summary(self, user_id: str) -> dict | None:
        """BodySummary is a single object (not paginated). Returns None if no data."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{self.base_url}/api/v1/users/{user_id}/summaries/body",
                headers=self.headers,
            )
            return resp.json()

    async def get_timeseries(
        self,
        user_id: str,
        start_time: str,
        end_time: str,
        types: list[str],
        resolution: Literal["raw", "1min", "5min", "15min", "1hour"] = "raw",
    ) -> list[dict]:
        return await self._paginated(
            f"/api/v1/users/{user_id}/timeseries",
            {"start_time": start_time, "end_time": end_time, "types": types, "resolution": resolution},
        )

    async def get_workouts(self, user_id: str, start_date: str, end_date: str) -> list[dict]:
        return await self._paginated(
            f"/api/v1/users/{user_id}/events/workouts",
            {"start_date": start_date, "end_date": end_date},
        )

    async def get_health_scores(
        self,
        user_id: str,
        category: str | None = None,
        provider: str | None = None,
    ) -> list[dict]:
        """Health scores use offset/limit pagination, not cursors."""
        params = {k: v for k, v in {"category": category, "provider": provider}.items() if v}
        items: list[dict] = []
        offset = 0
        async with httpx.AsyncClient() as client:
            while True:
                resp = await client.get(
                    f"{self.base_url}/api/v1/users/{user_id}/health-scores",
                    headers=self.headers,
                    params={**params, "limit": 200, "offset": offset},
                )
                payload = resp.json()
                data = payload.get("data", [])
                items.extend(data)
                if not payload["pagination"].get("has_more") or not data:
                    return items
                offset += len(data)


# Usage example
async def main():
    client = OpenWearablesClient(
        base_url="http://localhost:8000",
        api_key="sk-your-api-key",
    )

    # 1. Create or get user (pass your own user row; see get_or_create_user above)
    user = await client.get_or_create_user(local_user)
    user_id = user["id"]

    # 2. Get OAuth URL for Garmin
    auth_url = await client.get_auth_url(
        provider="garmin",
        user_id=user_id,
        redirect_uri="https://myapp.com/callback",
    )
    print(f"Redirect user to: {auth_url}")

    # 3. After OAuth callback, confirm the connection is active.
    # Historical backfill is dispatched automatically - no explicit sync call needed.
    connections = await client.get_connections(user_id)
    garmin_connected = any(
        c["provider"] == "garmin" and c["status"] == "active"
        for c in connections
    )

    if garmin_connected:
        # 4. Fetch whatever you need.
        workouts = await client.get_workouts(user_id, "2025-01-01", "2025-01-31")
        activity = await client.get_activity_summary(user_id, "2025-01-01", "2025-01-31")
        scores = await client.get_health_scores(user_id, category="sleep")
        print(f"{len(workouts)} workouts, {len(activity)} activity days, {len(scores)} sleep scores")