Garmin Connect API reference

Garmin Connect linking and sync. See Integrations for how it behaves for users.

Routes

garminconnect_retrieve_activities_days

garminconnect_retrieve_activities_days(request, start_date, end_date, token_user_id, db, storage)

Queue a Garmin Connect sync of an explicit date range.

Returns 202 with a job handle; poll GET /activities/ingestion-jobs/{job_id} for the outcome. The range is fetched by a background job, never on this request or an API worker thread.

Source code in backend/app/modules/garmin/router.py

@router.get(
    "/activities",
    status_code=202,
    response_model=activity_ingestion_schema.ActivityIngestionJobRead,
)
@core_rate_limit.limiter.limit(core_rate_limit.PROVIDER_SYNC)
def garminconnect_retrieve_activities_days(
    request: Request,
    start_date: date,
    end_date: date,
    token_user_id: Annotated[
        int,
        Depends(auth_dependencies.get_sub_from_access_token),
    ],
    db: Annotated[Session, Depends(core_database.get_db)],
    storage: Annotated[
        platform_providers.StorageProvider,
        Depends(core_dependencies.get_storage),
    ],
) -> activity_ingestion_schema.ActivityIngestionJobRead:
    """Queue a Garmin Connect sync of an explicit date range.

    Returns ``202`` with a job handle; poll ``GET /activities/ingestion-jobs/{job_id}``
    for the outcome. The range is fetched by a background job, never on this
    request or an API worker thread.
    """
    if start_date > end_date:
        raise core_exceptions.InvalidInputError("start_date must not be after end_date")

    job = activity_ingestion_integration.request_provider_sync(
        token_user_id,
        garmin_provider_registry.PROVIDER_NAME,
        datetime.combine(start_date, datetime.min.time(), tzinfo=UTC),
        datetime.combine(end_date, datetime.max.time(), tzinfo=UTC),
        storage,
        db,
    )
    logger.info(f"Garmin Connect activities queued for user {token_user_id}")
    return job

Activity sync

fetch_and_process_activities_by_dates

fetch_and_process_activities_by_dates(garminconnect_client, start_date, end_date, user_id)

Fetch a user's Garmin Connect activities in a window and store the new ones.

Runs on a worker thread. Garmin round trips run with no session open; the duplicate check and the store each open a short one. One activity failing is logged and skipped, so it does not stop the rest of the window.

Parameters:

Name	Type	Description	Default
garminconnect_client	Garmin	A client authenticated as the user.	required
start_date	datetime	Start of the window.	required
end_date	datetime	End of the window.	required
user_id	int	The user the activities belong to.	required

Returns:

Type	Description
list[ActivityRead] | None	The stored activities, or None when there were none.

Source code in backend/app/modules/garmin/activity_utils.py

def fetch_and_process_activities_by_dates(
    garminconnect_client: garminconnect.Garmin,
    start_date: datetime,
    end_date: datetime,
    user_id: int,
) -> list[activity_schema.ActivityRead] | None:
    """Fetch a user's Garmin Connect activities in a window and store the new ones.

    Runs on a worker thread. Garmin round trips run with no session open; the
    duplicate check and the store each open a short one. One activity failing is
    logged and skipped, so it does not stop the rest of the window.

    Args:
        garminconnect_client: A client authenticated as the user.
        start_date: Start of the window.
        end_date: End of the window.
        user_id: The user the activities belong to.

    Returns:
        The stored activities, or ``None`` when there were none.
    """
    try:
        garmin_activities = garminconnect_client.get_activities_by_date(
            str(start_date.date()),
            str(end_date.date()),
        )
    except garminconnect.GarminConnectAuthenticationError as err:
        logger.error(
            f"Garmin Connect token expired for user {user_id}. Unlinking account. User must re-link: {err}",
            exc_info=err,
        )
        _unlink_expired_account(user_id)
        return None
    except Exception as err:
        logger.error(
            f"Error fetching activities for user {user_id} between {start_date.date()} and {end_date.date()}: {err}",
            exc_info=err,
        )
        return None

    if garmin_activities is None:
        logger.info(
            f"User {user_id}: No new Garmin Connect activities found between {start_date.date()} and {end_date.date()}",
            extra=core_logger.context(console=True),
        )
        return None

    parsed_activities: list[activity_schema.ActivityRead] = []
    for activity in garmin_activities:
        activity_id = activity["activityId"]

        with SessionLocal() as db:
            already_stored = activity_integration.get_activity_by_garminconnect_id(activity_id, user_id, db)
        if already_stored:
            logger.info(f"User {user_id}: Activity {activity_id} already stored in the database")
            continue

        logger.info(f"User {user_id}: Processing activity {activity_id}")
        try:
            parsed_activities.extend(_download_and_ingest_activity(garminconnect_client, activity, user_id))
        except Exception as err:
            logger.error(
                f"User {user_id}: Error processing Garmin Connect activity {activity_id}: {err}",
                exc_info=err,
            )

    return parsed_activities if parsed_activities else None

get_user_garminconnect_activities_by_dates

get_user_garminconnect_activities_by_dates(start_date, end_date, user_id)

Fetch and store one user's Garmin Connect activities recorded in a window.

Parameters:

Name	Type	Description	Default
start_date	datetime	Start of the window.	required
end_date	datetime	End of the window.	required
user_id	int	The user to sync.	required

Returns:

Type	Description
list[ActivityRead] | None	The stored activities, or None when Garmin Connect is not linked or
list[ActivityRead] | None	nothing new came back.

Source code in backend/app/modules/garmin/activity_utils.py

def get_user_garminconnect_activities_by_dates(
    start_date: datetime,
    end_date: datetime,
    user_id: int,
) -> list[activity_schema.ActivityRead] | None:
    """Fetch and store one user's Garmin Connect activities recorded in a window.

    Args:
        start_date: Start of the window.
        end_date: End of the window.
        user_id: The user to sync.

    Returns:
        The stored activities, or ``None`` when Garmin Connect is not linked or
        nothing new came back.
    """
    try:
        with SessionLocal() as db:
            garminconnect_client = get_user_garminconnect_client(user_id, db)

        if garminconnect_client is None:
            return None

        logger.info(
            f"User {user_id}: Started Garmin Connect activities processing for date range "
            f"{start_date.date()} to {end_date.date()}"
        )
        processed = fetch_and_process_activities_by_dates(garminconnect_client, start_date, end_date, user_id)
        logger.info(f"User {user_id}: {len(processed) if processed else 0} Garmin Connect activities processed")
        return processed
    except Exception as err:
        logger.error(f"Error in get_user_garminconnect_activities_by_dates: {err}", exc_info=err)
        return None

Gear sync

match_gear_for_activity

match_gear_for_activity(activity, gears)

Return the local gear ID matching this activity's Garmin gear, else None.

Source code in backend/app/modules/garmin/gear_utils.py

def match_gear_for_activity(
    activity: activity_schema.ActivityRead,
    gears: list[gears_schema.GearRead],
) -> int | None:
    """Return the local gear ID matching this activity's Garmin gear, else None."""
    if activity.garminconnect_gear_id is None:
        return None
    for gear in gears:
        if activity.garminconnect_gear_id == gear.garminconnect_gear_id:
            logger.info(f"Gear found: {gear.nickname}")
            return gear.id
    return None

resolve_synced_gear_id

resolve_synced_gear_id(user_id, garminconnect_gear, db)

Resolve a downloaded activity's Garmin gear to a local gear ID.

Resolved here, before the file reaches ingestion, so the ingestion seam never has to ask a provider module whether gear sync is on or which local gear a Garmin UUID maps to.

Parameters:

Name	Type	Description	Default
user_id	int	Owner user ID.	required
garminconnect_gear	list | None	Garmin gear metadata ([{"uuid": ...}, ...]) from the activity download, when the activity has any.	required
db	Session	Database session.	required

Returns:

Type	Description
int | None	The local gear ID, or None when gear sync is off, the account's token is
int | None	invalid, the activity has no gear, or no local gear matches.

Source code in backend/app/modules/garmin/gear_utils.py

def resolve_synced_gear_id(user_id: int, garminconnect_gear: list | None, db: Session) -> int | None:
    """
    Resolve a downloaded activity's Garmin gear to a local gear ID.

    Resolved here, before the file reaches ingestion, so the ingestion seam never
    has to ask a provider module whether gear sync is on or which local gear a
    Garmin UUID maps to.

    Args:
        user_id: Owner user ID.
        garminconnect_gear: Garmin gear metadata (``[{"uuid": ...}, ...]``) from
            the activity download, when the activity has any.
        db: Database session.

    Returns:
        The local gear ID, or None when gear sync is off, the account's token is
        invalid, the activity has no gear, or no local gear matches.
    """
    if not garminconnect_gear:
        return None
    user_integrations = garmin_utils.fetch_user_integrations_and_validate_token(user_id, db)
    if user_integrations is None or not user_integrations.garminconnect_sync_gear:
        return None
    gear = gears_crud.get_gear_by_garminconnect_id_from_user_id(garminconnect_gear[0]["uuid"], user_id, db)
    return gear.id if gear is not None else None

Health sync

fetch_and_process_sleep_by_dates

fetch_and_process_sleep_by_dates(garminconnect_client, start_date, end_date, user_id, db)

Fetch and process sleep data from Garmin Connect.

Parameters:

Name	Type	Description	Default
garminconnect_client	Garmin	Authenticated Garmin Connect client.	required
start_date	datetime	Start date for sleep data retrieval.	required
end_date	datetime	End date for sleep data retrieval.	required
user_id	int	ID of the user to process sleep data for.	required
db	Session	Database session.	required

Returns:

Type	Description
int	Number of sleep records processed.

Source code in backend/app/modules/garmin/health_utils.py

def fetch_and_process_sleep_by_dates(
    garminconnect_client: garminconnect.Garmin,
    start_date: datetime,
    end_date: datetime,
    user_id: int,
    db: Session,
) -> int:
    """
    Fetch and process sleep data from Garmin Connect.

    Args:
        garminconnect_client: Authenticated Garmin Connect client.
        start_date: Start date for sleep data retrieval.
        end_date: End date for sleep data retrieval.
        user_id: ID of the user to process sleep data for.
        db: Database session.

    Returns:
        Number of sleep records processed.
    """
    count_processed = 0
    current_date = start_date

    # Iterate through each date since get_sleep_data only supports
    # single date
    while current_date <= end_date:
        date_string = current_date.strftime("%Y-%m-%d")

        try:
            garmin_sleep = garminconnect_client.get_sleep_data(date_string)
        except garminconnect.GarminConnectAuthenticationError:
            raise
        except Exception as err:
            logger.error(f"Error fetching sleep data for user {user_id} on {date_string}: {err}", exc_info=err)
            current_date += timedelta(days=1)
            continue

        if garmin_sleep is None or "dailySleepDTO" not in garmin_sleep or not garmin_sleep["dailySleepDTO"]:
            logger.info(f"User {user_id}: No Garmin Connect sleep data found for {date_string}")
            current_date += timedelta(days=1)
            continue

        sleep_dto = garmin_sleep["dailySleepDTO"]

        # For a date without sleep Garmin still returns a DTO whose fields are all null
        if not sleep_dto.get("id") or not sleep_dto.get("sleepTimeSeconds"):
            logger.info(f"User {user_id}: No Garmin Connect sleep data found for {date_string}")
            current_date += timedelta(days=1)
            continue

        # Convert timestamps from milliseconds to datetime
        sleep_start_gmt = (
            datetime.fromtimestamp(
                sleep_dto["sleepStartTimestampGMT"] / 1000,
                tz=UTC,
            )
            if sleep_dto.get("sleepStartTimestampGMT")
            else None
        )
        sleep_end_gmt = (
            datetime.fromtimestamp(
                sleep_dto["sleepEndTimestampGMT"] / 1000,
                tz=UTC,
            )
            if sleep_dto.get("sleepEndTimestampGMT")
            else None
        )
        # NOTE: Garmin's ``*TimestampLocal`` epochs are already offset-shifted —
        # they encode the sleeper's WALL CLOCK, not an instant. Reading them with
        # ``tz=UTC`` therefore yields a datetime whose UTC components equal the
        # local wall clock, which is what every consumer of these two columns
        # wants (the edit form renders them straight into ``datetime-local``
        # inputs). The true instants live in the ``*_gmt`` pair above.
        #
        # This is the same "wall clock wearing an instant's clothes" shape that
        # was removed from activities: the columns are ``timestamptz`` but do not
        # hold a meaningful instant. Deriving them from the GMT pair plus the
        # recording offset (which is exactly ``local - gmt``) would remove the
        # ambiguity; until then, do not convert or compare these values.
        sleep_start_local = (
            datetime.fromtimestamp(
                sleep_dto["sleepStartTimestampLocal"] / 1000,
                tz=UTC,
            )
            if sleep_dto.get("sleepStartTimestampLocal")
            else None
        )
        sleep_end_local = (
            datetime.fromtimestamp(
                sleep_dto["sleepEndTimestampLocal"] / 1000,
                tz=UTC,
            )
            if sleep_dto.get("sleepEndTimestampLocal")
            else None
        )

        # Process sleep stages from sleepLevels array
        sleep_stages = []
        if garmin_sleep.get("sleepLevels"):
            for level in garmin_sleep["sleepLevels"]:
                activity_level = level.get("activityLevel")

                # Validate and convert activity_level to enum
                try:
                    # Map Garmin activity levels to sleep stage types
                    # 0=deep, 1=light, 2=REM, 3=awake
                    stage_type = health_sleep_schema.SleepStageType(activity_level)
                except (TypeError, ValueError):
                    # Skip unknown or missing levels
                    continue

                start_gmt_str = level.get("startGMT")
                end_gmt_str = level.get("endGMT")

                start_gmt = (
                    datetime.strptime(
                        start_gmt_str,
                        "%Y-%m-%dT%H:%M:%S.%f",
                    ).replace(tzinfo=UTC)
                    if start_gmt_str
                    else None
                )
                end_gmt = (
                    datetime.strptime(
                        end_gmt_str,
                        "%Y-%m-%dT%H:%M:%S.%f",
                    ).replace(tzinfo=UTC)
                    if end_gmt_str
                    else None
                )

                duration_seconds = None
                if start_gmt and end_gmt:
                    duration_seconds = int((end_gmt - start_gmt).total_seconds())

                sleep_stage = health_sleep_schema.HealthSleepStage(
                    stage_type=stage_type,
                    start_time_gmt=start_gmt,
                    end_time_gmt=end_gmt,
                    duration_seconds=duration_seconds,
                )
                sleep_stages.append(sleep_stage)

        # Extract sleep scores
        sleep_scores = sleep_dto.get("sleepScores", {})
        overall_score = sleep_scores.get("overall", {})
        total_duration_score = sleep_scores.get(
            "totalDuration",
            {},
        )
        awake_count_score = sleep_scores.get("awakeCount", {})
        deep_percentage_score = sleep_scores.get(
            "deepPercentage",
            {},
        )
        light_percentage_score = sleep_scores.get(
            "lightPercentage",
            {},
        )
        rem_percentage_score = sleep_scores.get(
            "remPercentage",
            {},
        )
        sleep_stress_score = sleep_scores.get("stress", {})

        health_sleep = health_sleep_schema.HealthSleepBase(
            date=sleep_dto["calendarDate"],
            sleep_start_time_gmt=sleep_start_gmt,
            sleep_end_time_gmt=sleep_end_gmt,
            sleep_start_time_local=sleep_start_local,
            sleep_end_time_local=sleep_end_local,
            total_sleep_seconds=sleep_dto.get("sleepTimeSeconds"),
            nap_time_seconds=sleep_dto.get("napTimeSeconds"),
            unmeasurable_sleep_seconds=sleep_dto.get("unmeasurableSleepSeconds"),
            deep_sleep_seconds=sleep_dto.get("deepSleepSeconds"),
            light_sleep_seconds=sleep_dto.get("lightSleepSeconds"),
            rem_sleep_seconds=sleep_dto.get("remSleepSeconds"),
            awake_sleep_seconds=sleep_dto.get("awakeSleepSeconds"),
            avg_heart_rate=(int(sleep_dto.get("avgHeartRate")) if sleep_dto.get("avgHeartRate") is not None else None),
            min_heart_rate=None,
            max_heart_rate=None,
            avg_spo2=(
                int(sleep_dto.get("averageSpO2Value")) if sleep_dto.get("averageSpO2Value") is not None else None
            ),
            lowest_spo2=sleep_dto.get("lowestSpO2Value"),
            highest_spo2=sleep_dto.get("highestSpO2Value"),
            avg_respiration=(
                int(sleep_dto.get("averageRespirationValue"))
                if sleep_dto.get("averageRespirationValue") is not None
                else None
            ),
            lowest_respiration=(
                int(sleep_dto.get("lowestRespirationValue"))
                if sleep_dto.get("lowestRespirationValue") is not None
                else None
            ),
            highest_respiration=(
                int(sleep_dto.get("highestRespirationValue"))
                if sleep_dto.get("highestRespirationValue") is not None
                else None
            ),
            avg_stress_level=(
                int(sleep_dto.get("avgSleepStress")) if sleep_dto.get("avgSleepStress") is not None else None
            ),
            awake_count=sleep_dto.get("awakeCount"),
            restless_moments_count=None,
            sleep_score_overall=overall_score.get("value"),
            sleep_score_duration=total_duration_score.get("qualifierKey"),
            sleep_score_quality=overall_score.get("qualifierKey"),
            garminconnect_sleep_id=str(sleep_dto.get("id")),
            sleep_stages=sleep_stages if sleep_stages else None,
            source=health_sleep_schema.Source.GARMIN,
            hrv_status=(
                health_sleep_schema.HRVStatus(garmin_sleep.get("hrvStatus"))
                if garmin_sleep.get("hrvStatus")
                and garmin_sleep.get("hrvStatus") in health_sleep_schema.HRVStatus._value2member_map_
                else None
            ),
            resting_heart_rate=garmin_sleep.get("restingHeartRate"),
            avg_skin_temp_deviation=garmin_sleep.get("avgSkinTempDeviationC"),
            awake_count_score=(
                health_sleep_schema.SleepScore(awake_count_score.get("qualifierKey"))
                if awake_count_score
                and awake_count_score.get("qualifierKey") in health_sleep_schema.SleepScore._value2member_map_
                else None
            ),
            rem_percentage_score=(
                health_sleep_schema.SleepScore(rem_percentage_score.get("qualifierKey"))
                if rem_percentage_score
                and rem_percentage_score.get("qualifierKey") in health_sleep_schema.SleepScore._value2member_map_
                else None
            ),
            deep_percentage_score=(
                health_sleep_schema.SleepScore(deep_percentage_score.get("qualifierKey"))
                if deep_percentage_score
                and deep_percentage_score.get("qualifierKey") in health_sleep_schema.SleepScore._value2member_map_
                else None
            ),
            light_percentage_score=(
                health_sleep_schema.SleepScore(light_percentage_score.get("qualifierKey"))
                if light_percentage_score
                and light_percentage_score.get("qualifierKey") in health_sleep_schema.SleepScore._value2member_map_
                else None
            ),
            avg_sleep_stress=(
                int(sleep_dto.get("avgSleepStress")) if sleep_dto.get("avgSleepStress") is not None else None
            ),
            sleep_stress_score=(
                health_sleep_schema.SleepScore(sleep_stress_score.get("qualifierKey"))
                if sleep_stress_score
                and sleep_stress_score.get("qualifierKey") in health_sleep_schema.SleepScore._value2member_map_
                else None
            ),
        )

        health_sleep_db = health_sleep_crud.get_health_sleep_by_date_and_user_id(user_id, str(health_sleep.date), db)

        if health_sleep_db:
            # Never overwrite a row written by another source; rows created before
            # ``source`` was stored are recognised by their Garmin sleep id
            if (
                health_sleep_db.source != health_sleep_schema.Source.GARMIN
                and not health_sleep_db.garminconnect_sleep_id
            ):
                logger.info(f"User {user_id}: Keeping existing non-Garmin sleep data for date {health_sleep.date}")
                current_date += timedelta(days=1)
                continue
            # Convert to update schema with the existing ID
            health_sleep_update = health_sleep_schema.HealthSleepUpdate(
                id=health_sleep_db.id, user_id=user_id, **health_sleep.model_dump()
            )
            # Updates the health_sleep in the database
            health_sleep_crud.edit_health_sleep(user_id, health_sleep_update, db)
            logger.info(f"User {user_id}: Sleep data edited for date {health_sleep.date}")
        else:
            # Convert to update schema with the existing ID
            health_sleep_create = health_sleep_schema.HealthSleepCreate(**health_sleep.model_dump())
            # Creates the health_sleep in the database
            health_sleep_crud.create_health_sleep(user_id, health_sleep_create, db)
            logger.info(f"User {user_id}: Sleep data created for date {health_sleep.date}")

        count_processed += 1
        current_date += timedelta(days=1)

    return count_processed

Utilities

Schemas