Adding Wearable Providers

This guide walks you through integrating a new fitness data provider (e.g., Strava, Samsung Health, Xiaomi, WHOOP) into the Open Wearables platform. The architecture uses the Strategy, Factory, and Template Method design patterns to make adding providers straightforward and consistent.

Architecture Overview

Each provider integration consists of three main components:

Strategy Class

Defines the provider’s capabilities, configuration, and initializes components.

OAuth Handler

Manages authentication flow for cloud-based providers (PULL pattern).

Workouts Handler

Fetches and normalizes workout data into the unified schema.

Data Flow Patterns

Providers support one or both patterns:

PULL

Fetch data from provider’s cloud API using OAuth tokens (Garmin, Strava, WHOOP).

PUSH

Receive data via webhooks or file uploads (Suunto, Polar, Apple Health).

Garmin supports both PULL and PUSH patterns, while Suunto and Polar are PUSH-only.

Prerequisites

Before starting, gather information about your provider:

Provider API Information

Base API URL (e.g., https://apis.garmin.com)
Authentication method (typically OAuth 2.0)
Available data endpoints (activities, workouts, health metrics)
Rate limits and pagination details
API documentation URL

OAuth Configuration (if applicable)

Authorization URL
Token exchange URL
Required scopes
PKCE requirement (yes/no)
Credentials transmission method (header vs. body)
Client ID and Secret
Redirect URL registration location

Data Format

Workout/activity data structure
Timestamp format (Unix, ISO 8601, etc.)
Available metrics (heart rate, distance, calories, cadence, etc.)
Workout type enumeration
Time series data format (if available)

Remember to provide an SVG icon for the new provider. Name it <lowercase_provider_name>.svg and place it in /backend/app/static/provider-icons/.

Step 1: Create Provider Directory

Create a new directory for your provider in backend/app/services/providers/:

backend/app/services/providers/whoop/
├── __init__.py
├── strategy.py     # Provider strategy implementation
├── oauth.py        # OAuth handler (skip if PUSH-only)
└── workouts.py     # Workouts data handler

Use lowercase for the provider name to maintain consistency with existing providers.

Step 2: Implement the Strategy Class

The strategy class is the entry point that defines the provider’s identity and initializes components. Create backend/app/services/providers/whoop/strategy.py:

from app.services.providers.base_strategy import BaseProviderStrategy
from app.services.providers.whoop.oauth import WhoopOAuth
from app.services.providers.whoop.workouts import WhoopWorkouts


class WhoopStrategy(BaseProviderStrategy):
    """WHOOP provider implementation.
    
    Supports:
    - OAuth 2.0 authentication
    - Workouts/activities
    - Recovery and sleep data
    """

    def __init__(self):
        super().__init__()
        
        # Initialize OAuth component (skip for PUSH-only providers)
        self.oauth = WhoopOAuth(
            user_repo=self.user_repo,
            connection_repo=self.connection_repo,
            provider_name=self.name,
            api_base_url=self.api_base_url,
        )
        
        # Initialize workouts component
        self.workouts = WhoopWorkouts(
            workout_repo=self.workout_repo,
            connection_repo=self.connection_repo,
            provider_name=self.name,
            api_base_url=self.api_base_url,
            oauth=self.oauth,  # Pass None for PUSH-only providers
        )

    @property
    def name(self) -> str:
        """Unique identifier for the provider (lowercase)."""
        return "whoop"

    @property
    def api_base_url(self) -> str:
        """Base URL for the provider's API."""
        return "https://api.prod.whoop.com"

    @property
    def display_name(self) -> str:
        """Display name shown in UI (optional override)."""
        return "WHOOP"

Key Properties

name

Unique lowercase identifier used in URLs and database.

api_base_url

Base URL for API client to construct requests.

display_name

Optional override for UI display (defaults to name.capitalize()).

BaseProviderStrategy initializes all required repositories, so you don’t need to handle database operations directly.

Step 3: Implement OAuth Handler

For cloud-based providers using OAuth 2.0, implement the OAuth handler. Create backend/app/services/providers/whoop/oauth.py:

import httpx
from app.config import settings
from app.schemas import (
    AuthenticationMethod,
    OAuthTokenResponse,
    ProviderCredentials,
    ProviderEndpoints,
)
from app.services.providers.templates.base_oauth import BaseOAuthTemplate


class WhoopOAuth(BaseOAuthTemplate):
    """WHOOP OAuth 2.0 implementation."""

    @property
    def endpoints(self) -> ProviderEndpoints:
        """OAuth endpoints for authorization and token exchange."""
        return ProviderEndpoints(
            authorize_url="https://api.prod.whoop.com/oauth/oauth2/auth",
            token_url="https://api.prod.whoop.com/oauth/oauth2/token",
        )

    @property
    def credentials(self) -> ProviderCredentials:
        """OAuth credentials from environment variables."""
        return ProviderCredentials(
            client_id=settings.whoop_client_id or "",
            client_secret=(
                settings.whoop_client_secret.get_secret_value()
                if settings.whoop_client_secret
                else ""
            ),
            redirect_uri=settings.whoop_redirect_uri,
            default_scope=settings.whoop_default_scope,
        )

    # OAuth configuration
    use_pkce: bool = False  # Set True if provider requires PKCE
    auth_method: AuthenticationMethod = AuthenticationMethod.BASIC_AUTH

    def _get_provider_user_info(
        self,
        token_response: OAuthTokenResponse,
        user_id: str,
    ) -> dict[str, str | None]:
        """Fetch user information from WHOOP API.
        
        Returns:
            Dict with provider_user_id and optional provider_user_email
        """
        headers = {"Authorization": f"Bearer {token_response.access_token}"}
        response = httpx.get(
            f"{self.api_base_url}/developer/v1/user/profile/basic",
            headers=headers,
        )
        response.raise_for_status()
        data = response.json()
        
        return {
            "provider_user_id": str(data["user_id"]),
            "provider_user_email": data.get("email"),
        }

Configuration Options

use_pkce

Set to True if provider requires PKCE (Proof Key for Code Exchange). Garmin enforces PKCE; Polar and Suunto don’t.

auth_method

BASIC_AUTH: Credentials in Authorization header (Polar, Suunto)
BODY: Credentials in request body (Garmin)

Add Environment Variables

Add OAuth credentials to backend/config/.env:

WHOOP_CLIENT_ID=your_client_id
WHOOP_CLIENT_SECRET=your_client_secret
WHOOP_REDIRECT_URI=http://localhost:8000/api/v1/oauth/whoop/callback
WHOOP_DEFAULT_SCOPE=read:profile read:workout

Update backend/app/config.py:

from pydantic import SecretStr
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # ... existing settings ...
    
    # WHOOP OAuth
    whoop_client_id: str | None = None
    whoop_client_secret: SecretStr | None = None
    whoop_redirect_uri: str = "http://localhost:8000/api/v1/oauth/whoop/callback"
    whoop_default_scope: str = "read:profile read:workout"

Step 4: Implement Workouts Handler

The workouts handler fetches and normalizes workout data into the unified schema. Create backend/app/services/providers/whoop/workouts.py:

from datetime import datetime
from typing import Any, Iterable
from uuid import UUID, uuid4

from app.database import DbSession
from app.schemas import EventRecordCreate, EventRecordDetailCreate, EventRecordMetrics
from app.services.providers.templates.base_workouts import BaseWorkoutsTemplate
from app.services import event_record_service
from app.constants.workout_types.whoop import get_unified_workout_type


class WhoopWorkouts(BaseWorkoutsTemplate):
    """WHOOP workouts implementation."""

    def _extract_dates(
        self,
        start_iso: str,
        end_iso: str,
    ) -> tuple[datetime, datetime]:
        """Extract start and end dates from ISO 8601 timestamps."""
        start_date = datetime.fromisoformat(start_iso.replace("Z", "+00:00"))
        end_date = datetime.fromisoformat(end_iso.replace("Z", "+00:00"))
        return start_date, end_date

    def _build_metrics(self, raw_workout: dict[str, Any]) -> EventRecordMetrics:
        """Build metrics from WHOOP workout data."""
        return {
            "heart_rate_avg": raw_workout.get("average_heart_rate"),
            "heart_rate_max": raw_workout.get("max_heart_rate"),
            "calories_total": raw_workout.get("kilojoules"),
            "distance_total": raw_workout.get("distance_meters"),
        }

    def _normalize_workout(
        self,
        raw_workout: dict[str, Any],
        user_id: UUID,
    ) -> tuple[EventRecordCreate, EventRecordDetailCreate]:
        """Normalize WHOOP workout to EventRecordCreate schema."""
        workout_id = uuid4()
        
        # Get unified workout type
        sport_id = raw_workout.get("sport_id", 0)
        workout_type = get_unified_workout_type(sport_id)
        
        # Extract dates
        start_date, end_date = self._extract_dates(
            raw_workout["start"],
            raw_workout["end"],
        )
        
        # Calculate duration in seconds
        duration_seconds = int(
            (end_date - start_date).total_seconds()
        )
        
        # Build metrics
        metrics = self._build_metrics(raw_workout)
        
        # Create workout record
        workout_create = EventRecordCreate(
            category="workout",
            type=workout_type.value,
            source_name="WHOOP",
            device_id=None,  # WHOOP doesn't expose device ID
            duration_seconds=duration_seconds,
            start_datetime=start_date,
            end_datetime=end_date,
            id=workout_id,
            provider_id=str(raw_workout["id"]),
            user_id=user_id,
        )
        
        # Create workout details
        workout_detail_create = EventRecordDetailCreate(
            record_id=workout_id,
            **metrics,
        )
        
        return workout_create, workout_detail_create

    def _build_bundles(
        self,
        raw: list[dict[str, Any]],
        user_id: UUID,
    ) -> Iterable[tuple[EventRecordCreate, EventRecordDetailCreate]]:
        """Build event record bundles for WHOOP workouts."""
        for raw_workout in raw:
            record, details = self._normalize_workout(raw_workout, user_id)
            yield record, details

    def load_data(
        self,
        db: DbSession,
        user_id: UUID,
        **kwargs: Any,
    ) -> bool:
        """Load workout data from WHOOP API.
        
        Args:
            db: Database session
            user_id: User ID to sync data for
            **kwargs: Additional parameters (e.g., date range)
            
        Returns:
            True if sync successful
        """
        # Fetch workouts from API
        response = self.get_workouts_from_api(db, user_id, **kwargs)
        workouts_data = response.get("records", [])
        
        # Normalize and save
        for record, details in self._build_bundles(workouts_data, user_id):
            event_record_service.create(db, record)
            event_record_service.create_detail(db, details)
        
        return True

Key Methods

_normalize_workout()

Most important! Converts provider’s data format to the unified schema.

_extract_dates()

Handles provider-specific timestamp formats (Unix, ISO 8601, custom strings).

_build_metrics()

Extracts statistics (heart rate, distance, calories) from workout data.

_build_bundles()

Optimizes database operations by bundling workout records.

load_data()

Main sync method that orchestrates fetching and saving data.

Use app/services/providers/api_client.py for making authenticated API requests with automatic retry logic.

Step 5: Create Workout Type Mapping

Create a mapping to convert provider-specific workout types to unified types. Create backend/app/constants/workout_types/whoop.py:

"""WHOOP sport ID to Open Wearables unified workout type mapping."""
from app.schemas.workout_types import WorkoutType

WHOOP_WORKOUT_TYPE_MAPPINGS: list[tuple[int, str, WorkoutType]] = [
    (0, "Running", WorkoutType.RUNNING),
    (1, "Cycling", WorkoutType.CYCLING),
    (16, "Baseball", WorkoutType.OTHER),
    (17, "Basketball", WorkoutType.BASKETBALL),
    (18, "Rowing", WorkoutType.ROWING),
    (43, "Functional Fitness", WorkoutType.FUNCTIONAL_TRAINING),
    (52, "CrossFit", WorkoutType.CROSS_TRAINING),
    (55, "Gymnastics", WorkoutType.GYMNASTICS),
    (71, "Hiking", WorkoutType.HIKING),
    (-1, "Activity", WorkoutType.OTHER),
]

WHOOP_ID_TO_UNIFIED: dict[int, WorkoutType] = {
    sport_id: unified_type
    for sport_id, _, unified_type in WHOOP_WORKOUT_TYPE_MAPPINGS
}

WHOOP_ID_TO_NAME: dict[int, str] = {
    sport_id: name
    for sport_id, name, _ in WHOOP_WORKOUT_TYPE_MAPPINGS
}


def get_unified_workout_type(whoop_sport_id: int) -> WorkoutType:
    """Get unified workout type for WHOOP sport ID."""
    return WHOOP_ID_TO_UNIFIED.get(whoop_sport_id, WorkoutType.OTHER)


def get_sport_name(whoop_sport_id: int) -> str:
    """Get the WHOOP sport name for a given ID."""
    return WHOOP_ID_TO_NAME.get(whoop_sport_id, "Unknown")

Review existing unified workout types before mapping. You may need to add new types to accommodate provider-specific activities.

Step 6: Register Provider in Factory

Add your provider to the factory for system-wide instantiation. Edit backend/app/services/providers/factory.py:

from app.schemas.oauth import ProviderName
from app.services.providers.apple.strategy import AppleStrategy
from app.services.providers.base_strategy import BaseProviderStrategy
from app.services.providers.garmin.strategy import GarminStrategy
from app.services.providers.polar.strategy import PolarStrategy
from app.services.providers.strava.strategy import StravaStrategy
from app.services.providers.suunto.strategy import SuuntoStrategy
from app.services.providers.whoop.strategy import WhoopStrategy


class ProviderFactory:
    """Factory for creating provider instances."""

    def get_provider(self, provider_name: str) -> BaseProviderStrategy:
        match provider_name:
            case ProviderName.APPLE.value:
                return AppleStrategy()
            case ProviderName.GARMIN.value:
                return GarminStrategy()
            case ProviderName.SUUNTO.value:
                return SuuntoStrategy()
            case ProviderName.POLAR.value:
                return PolarStrategy()
            case ProviderName.STRAVA.value:
                return StravaStrategy()
            case ProviderName.WHOOP.value:
                return WhoopStrategy()
            case _:
                raise ValueError(f"Unknown provider: {provider_name}")

Step 7: Add Provider to Schema Enums

Update the ProviderName enum to include your provider. Edit backend/app/schemas/oauth.py:

from enum import Enum


class ProviderName(str, Enum):
    """Supported fitness data providers."""
    
    APPLE = "apple"
    GARMIN = "garmin"
    POLAR = "polar"
    STRAVA = "strava"
    SUUNTO = "suunto"
    WHOOP = "whoop"

This enables:

Type validation in API endpoints
Auto-generated API documentation
Enum-based routing

Step 8: Test Your Integration

1. Test OAuth Flow

curl -X GET "http://localhost:8000/api/v1/oauth/whoop/authorize?user_id=YOUR_USER_ID"

Visit the authorization URL, complete OAuth flow, and verify the callback works.

2. Test Data Sync

curl -X POST "http://localhost:8000/api/v1/sync/whoop/users/YOUR_USER_ID/sync" \
  -H "X-API-Key: YOUR_API_KEY" \
  -H "Content-Type: application/json"

3. Verify Database

Check that workouts are saved correctly:

SELECT 
  id, 
  type, 
  start_datetime, 
  duration_seconds, 
  provider_id
FROM event_records
WHERE user_id = 'YOUR_USER_ID'
  AND source_name = 'WHOOP'
ORDER BY start_datetime DESC
LIMIT 10;

4. Monitor Logs

# Backend logs
docker compose logs -f app

# Or if running locally
tail -f logs/app.log

5. Run Tests

Create unit tests for your provider:

cd backend
uv run pytest tests/providers/test_whoop_strategy.py -v

Advanced: Webhook Support

For providers supporting webhooks (PUSH pattern), implement webhook handling:

Create Webhook Endpoint

Create backend/app/api/routes/v1/webhooks.py:

from fastapi import APIRouter, Request, HTTPException
from app.database import DbSession
from app.services.providers.factory import ProviderFactory

router = APIRouter()
factory = ProviderFactory()


@router.post("/webhooks/whoop")
async def whoop_webhook_handler(
    request: Request,
    db: DbSession,
):
    """Handle WHOOP webhook events."""
    payload = await request.json()
    
    # Verify webhook signature (implement based on provider docs)
    # ...
    
    if payload["type"] == "workout.created":
        user_connection = # ... fetch user connection
        
        strategy = factory.get_provider("whoop")
        workout_data = payload["data"]
        
        record, detail = strategy.workouts._normalize_workout(
            workout_data,
            user_connection.user_id,
        )
        
        # Save to database
        strategy.workouts._save_workout(db, record, detail)
    
    return {"status": "success"}

Troubleshooting

OAuth flow fails

Verify client_id, client_secret, and redirect_uri are correct
Check that redirect URI is registered with the provider
Ensure PKCE is enabled/disabled correctly based on provider requirements
Review provider’s OAuth documentation for specific requirements

Data not normalizing correctly

Add detailed logging in _normalize_workout() to inspect raw data:

import logging
logger = logging.getLogger(__name__)
logger.info(f"Raw workout data: {raw_workout}")

Compare against provider’s API documentation.

Duplicate workouts created

Implement duplicate detection based on provider_id:

existing = event_record_service.get_by_provider_id(
    db, provider_id=raw_workout["id"]
)
if existing:
    return  # Skip duplicate

Missing workout types

Add missing types to your mapping file
Use WorkoutType.OTHER as fallback
Log unmapped types for future updates:

if sport_id not in WHOOP_ID_TO_UNIFIED:
    logger.warning(f"Unmapped WHOOP sport ID: {sport_id}")

Summary Checklist

Use this checklist to ensure completion:

Congratulations! You’ve successfully integrated a new provider into Open Wearables.

Next Steps

Submit Pull Request

Follow the contributing guidelines to submit your integration.

Add 24/7 Data Handler

Extend your provider to sync continuous health data (sleep, heart rate, etc.).

Implement Webhooks

Add real-time data sync via webhooks for instant updates.

Write Tests

Create comprehensive tests for OAuth, normalization, and edge cases.

Getting Started

Core Features

Guides

Deployment

Development

​Adding Wearable Providers

​Architecture Overview

​Data Flow Patterns

PULL

PUSH

​Prerequisites

​Step 1: Create Provider Directory

​Step 2: Implement the Strategy Class

​Key Properties

name

api_base_url

display_name

​Step 3: Implement OAuth Handler

​Configuration Options

use_pkce

auth_method

​Add Environment Variables

​Step 4: Implement Workouts Handler

​Key Methods

​Step 5: Create Workout Type Mapping

​Step 6: Register Provider in Factory

​Step 7: Add Provider to Schema Enums

​Step 8: Test Your Integration

​1. Test OAuth Flow

​2. Test Data Sync

​3. Verify Database

​4. Monitor Logs

​5. Run Tests

​Advanced: Webhook Support

​Create Webhook Endpoint

​Troubleshooting

​Summary Checklist

​Next Steps

Submit Pull Request

Add 24/7 Data Handler

Implement Webhooks

Write Tests

Build docs developers (and LLMs) love

Adding Wearable Providers

Architecture Overview

Data Flow Patterns

Prerequisites

Step 1: Create Provider Directory

Step 2: Implement the Strategy Class

Key Properties

Step 3: Implement OAuth Handler

Configuration Options

Add Environment Variables

Step 4: Implement Workouts Handler

Key Methods

Step 5: Create Workout Type Mapping

Step 6: Register Provider in Factory

Step 7: Add Provider to Schema Enums

Step 8: Test Your Integration

1. Test OAuth Flow

2. Test Data Sync

3. Verify Database

4. Monitor Logs

5. Run Tests

Advanced: Webhook Support

Create Webhook Endpoint

Troubleshooting

Summary Checklist

Next Steps