Strava API reference

Strava linking, sync and bulk export import. See Integrations and Importing data.

Routes

import_bikes_from_strava_export

import_bikes_from_strava_export(request, file, token_user_id, db)

Create the bikes listed in an uploaded Strava bikes.csv.

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

@router.post("/import/bikes", status_code=201)
@core_rate_limit.limiter.limit(core_rate_limit.UPLOAD)
def import_bikes_from_strava_export(
    request: Request,
    file: UploadFile,
    token_user_id: Annotated[
        int,
        Depends(auth_dependencies.get_sub_from_access_token),
    ],
    db: Annotated[
        Session,
        Depends(core_database.get_db),
    ],
):
    """Create the bikes listed in an uploaded Strava ``bikes.csv``."""
    work_dir, path = _save_gear_csv(file)
    try:
        bikes_dict = strava_gear_utils.iterate_over_bikes_csv(path)
        bikes = (
            strava_gear_utils.transform_csv_bike_gear_to_schema_gear(bikes_dict, token_user_id) if bikes_dict else []
        )
        if bikes:
            gears_crud.create_multiple_gears(bikes, token_user_id, db)
        logger.info("Bike import complete.", extra=core_logger.context(console=True))
    finally:
        work_dir.cleanup()

import_shoes_from_strava_export

import_shoes_from_strava_export(request, file, token_user_id, db)

Create the shoes listed in an uploaded Strava shoes.csv.

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

@router.post("/import/shoes", status_code=201)
@core_rate_limit.limiter.limit(core_rate_limit.UPLOAD)
def import_shoes_from_strava_export(
    request: Request,
    file: UploadFile,
    token_user_id: Annotated[
        int,
        Depends(auth_dependencies.get_sub_from_access_token),
    ],
    db: Annotated[
        Session,
        Depends(core_database.get_db),
    ],
):
    """Create the shoes listed in an uploaded Strava ``shoes.csv``."""
    work_dir, path = _save_gear_csv(file)
    try:
        shoes_list = strava_gear_utils.iterate_over_shoes_csv(path)
        shoes = (
            strava_gear_utils.transform_csv_shoe_gear_to_schema_gear(shoes_list, token_user_id, db)
            if shoes_list
            else []
        )
        if shoes:
            gears_crud.create_multiple_gears(shoes, token_user_id, db)
        logger.info("Shoe import complete.", extra=core_logger.context(console=True))
    finally:
        work_dir.cleanup()

strava_retrieve_activities_days

strava_retrieve_activities_days(request, start_date, end_date, _validate_access_token, _check_scopes, token_user_id, db, storage)

Queue a Strava 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/strava/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 strava_retrieve_activities_days(
    request: Request,
    start_date: date,
    end_date: date,
    _validate_access_token: Annotated[
        Callable,
        Depends(auth_dependencies.validate_access_token),
    ],
    _check_scopes: Annotated[
        Callable,
        Security(auth_dependencies.check_scopes, scopes=["profile"]),
    ],
    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 Strava 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,
        strava_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"Strava activities queued for user {token_user_id}")
    return job

Activity sync

fetch_and_process_activities

fetch_and_process_activities(strava_client, start_date, end_date, user_id, user_integrations)

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

Runs on a worker thread. The Strava round trips happen with no database session open; each activity then opens a short session of its own for the duplicate check, the lookups and the store.

Parameters:

Name	Type	Description	Default
strava_client	Client	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
user_integrations	UsersIntegrationsRead	The user's integration settings.	required

Returns:

Type	Description
list[ActivityRead | None] | None	One entry per activity Strava returned (None for one already stored),
list[ActivityRead | None] | None	or None when Strava returned nothing or the user is rate limited.

Raises:

Type	Description
HTTPException	When Strava rejects the request or the user is missing.

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

def fetch_and_process_activities(
    strava_client: Client,
    start_date: datetime,
    end_date: datetime,
    user_id: int,
    user_integrations: user_integrations_schema.UsersIntegrationsRead,
) -> list[activity_schema.ActivityRead | None] | None:
    """Fetch a user's Strava activities in a window and store the new ones.

    Runs on a worker thread. The Strava round trips happen with no database
    session open; each activity then opens a short session of its own for the
    duplicate check, the lookups and the store.

    Args:
        strava_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.
        user_integrations: The user's integration settings.

    Returns:
        One entry per activity Strava returned (``None`` for one already stored),
        or ``None`` when Strava returned nothing or the user is rate limited.

    Raises:
        HTTPException: When Strava rejects the request or the user is missing.
    """
    # Check if we're currently rate-limited before making any Strava API calls
    if strava_utils.rate_limit_tracker.is_rate_limited():
        logger.warning(
            f"User {user_id}: Strava rate limit active, skipping activity fetch",
            extra=core_logger.context(console=True),
        )
        return None

    try:
        strava_activities = list(strava_client.get_activities(after=start_date, before=end_date))
    except AccessUnauthorized as auth_err:
        logger.error(f"User {user_id}: Authentication error with Strava: {auth_err!s}", exc_info=auth_err)
        raise HTTPException(
            status_code=(status.HTTP_424_FAILED_DEPENDENCY),
            detail=("Strava authentication failed. Please reconnect your Strava account."),
        ) from auth_err
    except HTTPException:
        raise
    except Exception as err:
        if strava_utils.is_strava_rate_limit_error(err):
            err_str = str(err).lower()
            is_long = "long" in err_str or ("daily" in err_str)
            strava_utils.rate_limit_tracker.mark_rate_limited(is_long_term=is_long)
            logger.warning(f"User {user_id}: Strava rate limit exceeded, skipping activity fetch")
            raise HTTPException(
                status_code=(status.HTTP_429_TOO_MANY_REQUESTS),
                detail=("Strava API rate limit exceeded. Please try again later."),
            ) from err

        logger.error(f"User {user_id}: Error fetching Strava activities: {err!s}", exc_info=err)
        raise HTTPException(
            status_code=(status.HTTP_424_FAILED_DEPENDENCY),
            detail=("Not able to fetch Strava activities"),
        ) from err

    if not strava_activities:
        logger.info(f"User {user_id}: No new Strava activities found after {start_date}")
        return None

    with SessionLocal() as db:
        user = users_crud.get_user_by_id(user_id, db)
        if user is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="User not found",
            )
        user_privacy_settings = users_privacy_settings_crud.get_user_privacy_settings_by_user_id(user.id, db)

    return [
        process_activity(
            activity,
            user_id,
            user_privacy_settings,
            strava_client,
            user_integrations,
        )
        for activity in strava_activities
    ]

get_user_strava_activities_by_dates

get_user_strava_activities_by_dates(start_date, end_date, user_id)

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

Refreshes the user's access token first when it is about to expire, so the scheduled per-user refresh also keeps tokens current and no separate pass over every user is needed.

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] | None	The activities Strava returned, or None when Strava is not linked or
list[ActivityRead | None] | None	nothing came back.

Raises:

Type	Description
HTTPException	When Strava rejects the request or the account is missing.

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

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

    Refreshes the user's access token first when it is about to expire, so the
    scheduled per-user refresh also keeps tokens current and no separate pass over
    every user is needed.

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

    Returns:
        The activities Strava returned, or ``None`` when Strava is not linked or
        nothing came back.

    Raises:
        HTTPException: When Strava rejects the request or the account is missing.
    """
    strava_utils.refresh_user_strava_token(user_id)

    with SessionLocal() as db:
        user_integrations = strava_utils.fetch_user_integrations_and_validate_token(user_id, db)

    if user_integrations is None:
        logger.info(f"User {user_id}: Strava not linked")
        return None

    logger.info(f"User {user_id}: Started Strava activities processing")
    strava_client = strava_utils.create_strava_client(user_integrations)
    processed = fetch_and_process_activities(strava_client, start_date, end_date, user_id, user_integrations)

    logger.info(f"User {user_id}: {len([a for a in processed or [] if a])} Strava activities processed")
    return processed

process_activity

process_activity(activity, user_id, user_privacy_settings, strava_client, user_integrations)

Parse one Strava activity and store it, unless it is already stored.

The Strava calls inside :func:parse_activity run with no session open. The duplicate check and the store each use a short session, and the store publishes activity.created on this worker thread, where its subscribers run inline without touching the request path.

Parameters:

Name	Type	Description	Default
activity	SummaryActivity	The Strava summary activity.	required
user_id	int	The user the activity belongs to.	required
user_privacy_settings	UsersPrivacySettingsRead	The user's privacy defaults.	required
strava_client	Client	A client authenticated as the user.	required
user_integrations	UsersIntegrationsRead	The user's integration settings.	required

Returns:

Type	Description
ActivityRead | None	The stored activity, or None when it was already stored.

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

def process_activity(
    activity,
    user_id: int,
    user_privacy_settings: users_privacy_settings_schema.UsersPrivacySettingsRead,
    strava_client: Client,
    user_integrations: user_integrations_schema.UsersIntegrationsRead,
) -> activity_schema.ActivityRead | None:
    """Parse one Strava activity and store it, unless it is already stored.

    The Strava calls inside :func:`parse_activity` run with no session open. The
    duplicate check and the store each use a short session, and the store
    publishes ``activity.created`` on this worker thread, where its subscribers
    run inline without touching the request path.

    Args:
        activity (stravalib.model.SummaryActivity): The Strava summary activity.
        user_id: The user the activity belongs to.
        user_privacy_settings: The user's privacy defaults.
        strava_client: A client authenticated as the user.
        user_integrations: The user's integration settings.

    Returns:
        The stored activity, or ``None`` when it was already stored.
    """
    with SessionLocal() as db:
        already_stored = strava_utils.fetch_and_validate_activity(activity.id, user_id, db) is not None
    if already_stored:
        return None

    logger.info(f"User {user_id}: Strava activity {activity.id} will be processed")

    parsed_activity = parse_activity(activity, user_id, user_privacy_settings, strava_client, user_integrations)

    with SessionLocal() as db:
        return save_activity_streams_laps(
            parsed_activity["activity_to_store"],
            parsed_activity["stream_data"],
            parsed_activity["laps"],
            db,
        )

save_activity_streams_laps

save_activity_streams_laps(activity, stream_data, laps, db)

Persist a Strava-parsed activity through the canonical ingestion seam.

Strava is a thin adapter: it builds the format-agnostic :class:~modules.activities.activity.schema.ParsedActivity from the parse result and delegates persistence + activity.created publication to :func:~modules.activities.activity.activity_integration.store_parsed_activity, instead of calling create_activity and the stream/lap CRUD directly. The notification, thumbnail, and HR-zone subscribers react to the published event.

Parameters:

Name	Type	Description	Default
activity	ActivityCore	The parsed Strava activity to persist.	required
stream_data	list	(is_set, stream_type, waypoints) tuples; only set streams are persisted.	required
laps	list[dict] | None	Parsed lap dicts, or None when the activity has no laps.	required
db	Session	Database session.	required

Returns:

Type	Description
ActivityRead	The created activity schema.

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

def save_activity_streams_laps(
    activity: activity_contracts.ActivityCore,
    stream_data: list,
    laps: list[dict] | None,
    db: Session,
) -> activity_schema.ActivityRead:
    """Persist a Strava-parsed activity through the canonical ingestion seam.

    Strava is a thin adapter: it builds the
    format-agnostic :class:`~modules.activities.activity.schema.ParsedActivity`
    from the parse result and delegates persistence + ``activity.created``
    publication to
    :func:`~modules.activities.activity.activity_integration.store_parsed_activity`,
    instead of calling ``create_activity`` and the stream/lap CRUD directly. The
    notification, thumbnail, and HR-zone subscribers react to the published event.

    Args:
        activity: The parsed Strava activity to persist.
        stream_data: ``(is_set, stream_type, waypoints)`` tuples; only set streams
            are persisted.
        laps: Parsed lap dicts, or ``None`` when the activity has no laps.
        db: Database session.

    Returns:
        The created activity schema.
    """
    parsed_streams = [
        activity_streams_contracts.ParsedStream(stream_type=stream_type, stream_waypoints=waypoints)
        for is_set, stream_type, waypoints in (stream_data or [])
        if is_set
    ]
    parsed = activity_contracts.ParsedActivity(
        activity=activity,
        components={"streams": parsed_streams, "laps": laps},
        source=activity_contracts.ImportSource(
            kind="strava",
            provider_activity_id=activity.strava_activity_id,
            dedup_key=(f"strava:{activity.strava_activity_id}" if activity.strava_activity_id is not None else None),
        ),
    )

    # Persist + publish ``activity.created`` via the canonical seam. The seam marks
    # duplicate start times hidden and forwards that to the notification subscriber.
    stored = activity_integration.store_parsed_activity(parsed, db)
    db.commit()
    return stored

Gear sync

iterate_over_bikes_csv

iterate_over_bikes_csv(bikes_file_path)

Parses a Strava bikes CSV file and returns its contents as a dictionary.

Constructs a dictionary where each key is the "Bike Name" from the CSV, and the value is a dictionary of the bike's attributes as provided in the CSV row.

Parameters:

Name	Type	Description	Default
bikes_file_path	str	Path of the uploaded bikes CSV.	required

Returns:

Name	Type	Description
dict	dict	A dictionary mapping bike names to their corresponding data from the CSV file.

Raises:

Type	Description
HTTPException	If the CSV file is missing, has invalid headers, or cannot be parsed.

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

def iterate_over_bikes_csv(bikes_file_path: str) -> dict:
    """
    Parses a Strava bikes CSV file and returns its contents as a dictionary.

    Constructs a dictionary where each key is the "Bike Name" from the CSV, and the value is a dictionary of the bike's attributes as provided in the CSV row.

    Args:
        bikes_file_path: Path of the uploaded bikes CSV.

    Returns:
        dict: A dictionary mapping bike names to their corresponding data from the CSV file.

    Raises:
        HTTPException: If the CSV file is missing, has invalid headers, or cannot be parsed.
    """
    bikes_file_name = "bikes.csv"

    # Get file and parse it
    bikes_dict = {}
    try:
        if os.path.isfile(bikes_file_path):
            logger.info(
                f"Parsing the uploaded {bikes_file_name}.",
                extra=core_logger.context(console=True),
            )
            for row in core_text_imports.read_bounded_csv(bikes_file_path):
                if ("Bike Name" not in row) or ("Bike Brand" not in row) or ("Bike Model" not in row):
                    logger.info(
                        f"Aborting bikes import: Proper headers not found in {bikes_file_name}.  File should have 'Bike Name', 'Bike Brand', and 'Bike Model'.",
                        extra=core_logger.context(console=True),
                    )
                    raise HTTPException(
                        status_code=status.HTTP_424_FAILED_DEPENDENCY,
                        detail="Invalid file. Proper headers not found in Strava bikes CSV file.",
                    )
                bikes_dict[row["Bike Name"]] = row
            logger.info(
                f"Strava bike gear csv file parsed and gear dictionary created. File was {len(bikes_dict)} rows long, ignoring header row.",
                extra=core_logger.context(console=True),
            )
            return bikes_dict
        logger.info(
            "No Strava bikes CSV file found for import.",
            extra=core_logger.context(console=True),
        )
        raise HTTPException(
            status_code=status.HTTP_424_FAILED_DEPENDENCY,
            detail="No Strava bikes CSV file found for import.",
        )
    except HTTPException as http_err:
        raise http_err
    except Exception as err:
        logger.error(
            f"Error attempting to open {bikes_file_path} file:  {err}", extra=core_logger.context(console=True)
        )
        raise HTTPException(
            status_code=status.HTTP_424_FAILED_DEPENDENCY,
            detail="Error parsing Strava bikes CSV file.",
        ) from err

iterate_over_shoes_csv

iterate_over_shoes_csv(shoes_file_path)

Parses a Strava shoes CSV file and returns its contents as a list. Why a list? Strava shoes files do not have anything that can be used as a key (unlike Strava bikes files, where nickname is required and unique)

Constructs a list where each item a dictionary of the shoe's attributes as provided in the CSV row.

Parameters:

Name	Type	Description	Default
shoes_file_path	str	Path of the uploaded shoes CSV.	required

Returns:

Name	Type	Description
list	list	A list of shoe data obtained from the CSV file.

Raises:

Type	Description
HTTPException	If the CSV file is missing, has invalid headers, or cannot be parsed.

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

def iterate_over_shoes_csv(shoes_file_path: str) -> list:
    """
    Parses a Strava shoes CSV file and returns its contents as a list.
    Why a list?  Strava shoes files do not have anything that can be used as a key (unlike Strava bikes files, where nickname is required and unique)

    Constructs a list where each item a dictionary of the shoe's attributes as provided in the CSV row.

    Args:
        shoes_file_path: Path of the uploaded shoes CSV.

    Returns:
        list: A list of shoe data obtained from the CSV file.

    Raises:
        HTTPException: If the CSV file is missing, has invalid headers, or cannot be parsed.
    """
    shoesfilename = "shoes.csv"

    # Get file and parse it
    shoes_list = []  # Using a list for shoes, as there is no unique value to use as a dictionary key like with bikes
    try:
        if os.path.isfile(shoes_file_path):
            logger.info(
                f"Parsing the uploaded {shoesfilename}.",
                extra=core_logger.context(console=True),
            )
            for row in core_text_imports.read_bounded_csv(shoes_file_path):
                if ("Shoe Name" not in row) or ("Shoe Brand" not in row) or ("Shoe Model" not in row):
                    logger.info(
                        f"Aborting shoes import: Proper headers not found in {shoesfilename}. File should have 'Shoe Name', 'Shoe Brand', and 'Shoe Model'.",
                        extra=core_logger.context(console=True),
                    )
                    raise HTTPException(
                        status_code=status.HTTP_424_FAILED_DEPENDENCY,
                        detail="Invalid file. Proper headers not found in Strava shoes CSV file.",
                    )
                shoes_list.append(row)
            logger.info(
                f"Strava {shoesfilename} file parsed and gear dictionary created. File was {len(shoes_list)} rows long, ignoring header row.",
                extra=core_logger.context(console=True),
            )
            return shoes_list
        logger.info(
            "No Strava shoes CSV file found for import.",
            extra=core_logger.context(console=True),
        )
        raise HTTPException(
            status_code=status.HTTP_424_FAILED_DEPENDENCY,
            detail="No Strava shoes CSV file found for import.",
        )
    except HTTPException as http_err:
        raise http_err
    except Exception as err:
        logger.error(
            f"Error attempting to open {shoes_file_path} file:  {err}", extra=core_logger.context(console=True)
        )
        raise HTTPException(
            status_code=status.HTTP_424_FAILED_DEPENDENCY,
            detail="Error parsing Strava shoes CSV file.",
        ) from err

match_gear_for_activity

match_gear_for_activity(activity, gears)

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

Source code in backend/app/modules/strava/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 Strava gear, else None."""
    if activity.strava_gear_id is None:
        return None
    for gear in gears:
        if activity.strava_gear_id == gear.strava_gear_id:
            return gear.id
    return None

transform_csv_bike_gear_to_schema_gear

transform_csv_bike_gear_to_schema_gear(bikes_dict, token_user_id)

Transforms a dictionary of bike data (parsed from CSV) into a list of Gear schema objects.

Parameters:

Name	Type	Description	Default
bikes_dict	dict	A dictionary where each key is a bike nickname and each value is a dictionary containing bike attributes such as "Bike Brand" and "Bike Model".	required
token_user_id	int	The user ID to associate with each Gear object.	required

Returns:

Type	Description
list[GearCreate]	list[gears_schema.GearCreate]: A list of Gear schema objects created from the input bike data.

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

def transform_csv_bike_gear_to_schema_gear(bikes_dict: dict, token_user_id: int) -> list[gears_schema.GearCreate]:
    """
    Transforms a dictionary of bike data (parsed from CSV) into a list of Gear schema objects.

    Args:
        bikes_dict (dict): A dictionary where each key is a bike nickname and each value is a dictionary
            containing bike attributes such as "Bike Brand" and "Bike Model".
        token_user_id (int): The user ID to associate with each Gear object.

    Returns:
        list[gears_schema.GearCreate]: A list of Gear schema objects created from the input bike data.
    """
    gears = []
    for bike in bikes_dict:
        new_gear = gears_schema.GearCreate(
            user_id=token_user_id,
            brand=bikes_dict[bike]["Bike Brand"].strip() if bikes_dict[bike]["Bike Brand"] else None,
            model=bikes_dict[bike]["Bike Model"].strip() if bikes_dict[bike]["Bike Model"] else None,
            nickname=bike.strip(),
            gear_type=gears_schema.GearType.BIKE,
            active=True,
            strava_gear_id=None,
        )
        gears.append(new_gear)
    return gears

transform_csv_shoe_gear_to_schema_gear

transform_csv_shoe_gear_to_schema_gear(shoes_list, token_user_id, db)

Transforms a list of shoe data (parsed from CSV) into a list of Gear schema objects. Renames any name-less shoes to a default name plus a unique number, as Strava allows name-less shoes, but Endurain does not.

Parameters:

Name	Type	Description	Default
shoes_list	list	A list where each row is a single shoe's data, parsed from the strava shoe csv	required
token_user_id	int	The user ID to associate with each Gear object.	required
db	session	Database session	required

Returns:

Type	Description
list[GearCreate]	list[gears_schema.GearCreate]: A list of Gear schema objects created from the input shoe data.

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

def transform_csv_shoe_gear_to_schema_gear(
    shoes_list: list, token_user_id: int, db: Session
) -> list[gears_schema.GearCreate]:
    """
    Transforms a list of shoe data (parsed from CSV) into a list of Gear schema objects.
    Renames any name-less shoes to a default name plus a unique number, as Strava allows name-less shoes, but Endurain does not.

    Args:
        shoes_list (list): A list where each row is a single shoe's data, parsed from the strava shoe csv
        token_user_id (int): The user ID to associate with each Gear object.
        db (session): Database session

    Returns:
        list[gears_schema.GearCreate]: A list of Gear schema objects created from the input shoe data.
    """
    gears = []
    newnumber = 1
    shoe_name = ""
    for shoerow in shoes_list:
        # 1 - Check for nameless shoes and add a novel name. Why?  Because Strava allows nameless shoes, but Endurain does not.
        if (
            shoerow["Shoe Name"] is None
            or shoerow["Shoe Name"] == ""
            or shoerow["Shoe Name"].replace("+", " ").strip() == ""
        ):
            # Shoe name is blank or parses to blank; assign a new, non-duplicated name.
            proposed_name_is_already_present = True
            while proposed_name_is_already_present:
                proposed_name = core_config.STRAVA_BULK_IMPORT_SHOES_UNNAMED_SHOE + str(newnumber)
                gear_check = gears_crud.get_gear_user_by_nickname(token_user_id, proposed_name, db)
                if gear_check is not None:
                    newnumber += 1
                else:
                    proposed_name_is_already_present = False
                    shoe_name = proposed_name
                    logger.info(
                        f"Shoe name was blank, it has been updated to: {proposed_name}",
                        extra=core_logger.context(console=True),
                    )
                    newnumber += 1  # Iterate the number so the next unnamed shoe does not duplicate this one.
        else:
            # CSV data has a name for the shoe, use CSV's data as the name.
            shoe_name = shoerow["Shoe Name"].strip()

        # 2 - Add (possibly renamed) gear item to the list of gear to be added.
        new_gear = gears_schema.GearCreate(
            user_id=token_user_id,
            brand=shoerow["Shoe Brand"].strip() if shoerow["Shoe Brand"] else None,
            model=shoerow["Shoe Model"].strip() if shoerow["Shoe Model"] else None,
            nickname=shoe_name,
            gear_type=gears_schema.GearType.SHOES,
            active=True,
            strava_gear_id=None,
        )
        gears.append(new_gear)
    return gears

Athlete

Bulk import source

The Strava-export specialisation of the bulk-import source, and its archive layout.

A Strava export is a folder of activity files plus an activities.csv manifest and a media directory. Three things follow from that, and all three are facts about Strava rather than about bulk import:

• the manifest supplies names, descriptions and gear the exported files omit,
• it lists a multi-activity .fit once per activity inside it, so importing every activity from every listing multiplies them (a 5-activity file becomes 25),
• the photos ship next to the files rather than inside them.

Activities knows nothing about Strava: this module registers a :class:BulkImportFormat that says where the activity files are, where the user drops them, and how to build the source that answers those three questions.

StravaBulkImportSource dataclass

Bases: BulkImportSource

A file from a Strava bulk export.

Attributes:

Name	Type	Description
strava_activities	dict | None	Parsed Strava activities.csv, keyed by filename.
gear_nickname_to_id	dict | None	Gear nickname -> internal gear id, used to resolve the gear named in the CSV.
read_member	ArchiveMemberReader | None	Reads a member of the archive by name, to reach the photos.

Source code in backend/app/modules/strava/bulk_import_source.py

@dataclass(frozen=True)
class StravaBulkImportSource(activity_ingestion_integration.BulkImportSource):
    """A file from a Strava bulk export.

    Attributes:
        strava_activities: Parsed Strava ``activities.csv``, keyed by filename.
        gear_nickname_to_id: Gear nickname -> internal gear id, used to resolve
            the gear named in the CSV.
        read_member: Reads a member of the archive by name, to reach the photos.
    """

    strava_activities: dict | None = None
    gear_nickname_to_id: dict | None = None
    read_member: activity_ingestion_integration.ArchiveMemberReader | None = None

    def metadata_for(self, file_base_name: str) -> dict:
        """Build the supplemental metadata from the matching ``activities.csv`` row.

        Args:
            file_base_name: The original (pre-decompression) base filename, which
                is the key into the ``activities.csv`` data.

        Returns:
            The metadata dict, empty when the import carries no start timestamp.
        """
        if not self.import_initiated_time:
            return {}
        if not self.strava_activities:
            return super().metadata_for(file_base_name)
        return strava_bulk_import_utils.build_metadata_dict(
            file_base_name,
            self.strava_activities,
            self.import_initiated_time,
            self.gear_nickname_to_id,
        )

    def should_import(
        self,
        activity: activity_contracts.ActivityCore,
        activity_metadata: dict,
        *,
        activities_in_file: int,
    ) -> bool:
        """Whether one activity parsed from a multi-activity file should be imported.

        The ``activities.csv`` row identifies which activity inside the file this
        listing refers to, by start time; the others are duplicates of listings
        handled elsewhere in the run.

        Args:
            activity: The parsed activity under consideration.
            activity_metadata: Metadata from :meth:`metadata_for`.
            activities_in_file: How many activities the file yielded.

        Returns:
            ``True`` to import, ``False`` to skip as a duplicate listing.
        """
        if activities_in_file <= 1 or not self.strava_activities:
            return True
        if activity_metadata.get("metadata_found_in_csv") is not True:
            return True
        if strava_bulk_import_utils.does_activity_start_time_match_the_data_in_strava_activities_csv(
            activity, activity_metadata
        ):
            return True
        logger.debug(
            "Bulk activity import of multi-activity .fit file: skipping likely duplicate import. "
            "Start time does not align with start time for this .fit file in the Strava activities.csv file.",
            extra=core_logger.context(console=True),
        )
        return False

    def import_side_artifacts(
        self,
        created_activities: list[activity_schema.ActivityRead],
        file_base_name: str,
        db: Session,
    ) -> None:
        """Attach the photos the export ships alongside the activity file.

        Even a multi-activity ``.fit`` is handled correctly by attaching to the
        last created activity — a Strava export's activity directory holds one
        imported activity per file.

        Args:
            created_activities: Activities persisted from the file, in order.
            file_base_name: The file's original base name (the ``activities.csv``
                lookup key).
            db: Database session.
        """
        if not self.strava_activities or not created_activities or self.read_member is None:
            return
        strava_bulk_import_utils.import_media_from_strava_bulk_export(
            self.strava_activities,
            created_activities[-1],
            file_base_name,
            self.read_member,
            db,
        )

import_side_artifacts

import_side_artifacts(created_activities, file_base_name, db)

Attach the photos the export ships alongside the activity file.

Even a multi-activity .fit is handled correctly by attaching to the last created activity — a Strava export's activity directory holds one imported activity per file.

Parameters:

Name	Type	Description	Default
created_activities	list[ActivityRead]	Activities persisted from the file, in order.	required
file_base_name	str	The file's original base name (the activities.csv lookup key).	required
db	Session	Database session.	required

Source code in backend/app/modules/strava/bulk_import_source.py

def import_side_artifacts(
    self,
    created_activities: list[activity_schema.ActivityRead],
    file_base_name: str,
    db: Session,
) -> None:
    """Attach the photos the export ships alongside the activity file.

    Even a multi-activity ``.fit`` is handled correctly by attaching to the
    last created activity — a Strava export's activity directory holds one
    imported activity per file.

    Args:
        created_activities: Activities persisted from the file, in order.
        file_base_name: The file's original base name (the ``activities.csv``
            lookup key).
        db: Database session.
    """
    if not self.strava_activities or not created_activities or self.read_member is None:
        return
    strava_bulk_import_utils.import_media_from_strava_bulk_export(
        self.strava_activities,
        created_activities[-1],
        file_base_name,
        self.read_member,
        db,
    )

metadata_for

metadata_for(file_base_name)

Build the supplemental metadata from the matching activities.csv row.

Parameters:

Name	Type	Description	Default
file_base_name	str	The original (pre-decompression) base filename, which is the key into the activities.csv data.	required

Returns:

Type	Description
dict	The metadata dict, empty when the import carries no start timestamp.

Source code in backend/app/modules/strava/bulk_import_source.py

def metadata_for(self, file_base_name: str) -> dict:
    """Build the supplemental metadata from the matching ``activities.csv`` row.

    Args:
        file_base_name: The original (pre-decompression) base filename, which
            is the key into the ``activities.csv`` data.

    Returns:
        The metadata dict, empty when the import carries no start timestamp.
    """
    if not self.import_initiated_time:
        return {}
    if not self.strava_activities:
        return super().metadata_for(file_base_name)
    return strava_bulk_import_utils.build_metadata_dict(
        file_base_name,
        self.strava_activities,
        self.import_initiated_time,
        self.gear_nickname_to_id,
    )

should_import

should_import(activity, activity_metadata, *, activities_in_file)

Whether one activity parsed from a multi-activity file should be imported.

The activities.csv row identifies which activity inside the file this listing refers to, by start time; the others are duplicates of listings handled elsewhere in the run.

Parameters:

Name	Type	Description	Default
activity	ActivityCore	The parsed activity under consideration.	required
activity_metadata	dict	Metadata from :meth:metadata_for.	required
activities_in_file	int	How many activities the file yielded.	required

Returns:

Type	Description
bool	True to import, False to skip as a duplicate listing.

Source code in backend/app/modules/strava/bulk_import_source.py

def should_import(
    self,
    activity: activity_contracts.ActivityCore,
    activity_metadata: dict,
    *,
    activities_in_file: int,
) -> bool:
    """Whether one activity parsed from a multi-activity file should be imported.

    The ``activities.csv`` row identifies which activity inside the file this
    listing refers to, by start time; the others are duplicates of listings
    handled elsewhere in the run.

    Args:
        activity: The parsed activity under consideration.
        activity_metadata: Metadata from :meth:`metadata_for`.
        activities_in_file: How many activities the file yielded.

    Returns:
        ``True`` to import, ``False`` to skip as a duplicate listing.
    """
    if activities_in_file <= 1 or not self.strava_activities:
        return True
    if activity_metadata.get("metadata_found_in_csv") is not True:
        return True
    if strava_bulk_import_utils.does_activity_start_time_match_the_data_in_strava_activities_csv(
        activity, activity_metadata
    ):
        return True
    logger.debug(
        "Bulk activity import of multi-activity .fit file: skipping likely duplicate import. "
        "Start time does not align with start time for this .fit file in the Strava activities.csv file.",
        extra=core_logger.context(console=True),
    )
    return False

register_bulk_import_format

register_bulk_import_format()

Register the Strava export layout with activity ingestion.

Source code in backend/app/modules/strava/bulk_import_source.py

def register_bulk_import_format() -> None:
    """Register the Strava export layout with activity ingestion."""
    activity_ingestion_integration.register_bulk_import_format(
        activity_ingestion_integration.BulkImportFormat(
            name=FORMAT_NAME,
            is_activity_member=_is_activity_member,
            is_export_member=_is_export_member,
            drop_directory=core_config.strava_bulk_import_dir_for,
            build_source=_build_source,
        )
    )

Bulk import

build_import_dictionary

build_import_dictionary(file_base_name, import_initiated_time, is_strava_bulk_import=False, strava_activities=None)

Creates the "import_info" dictionary that is added to all activities that are imported from files.

Functions both for Strava imports and generic bulk imports, depending on whether the is_strava_bulk_import variable is set or not.

Returns: Dictionary that contains the import_dict values for the activity (which is then added to the activity as a dictionary in the "import_info" field of the activity).

Source code in backend/app/modules/strava/bulk_import_utils.py

def build_import_dictionary(
    file_base_name: str,  # String with the base filename being processed (also key to strav_activities dictionary)
    import_initiated_time: str,  # String containing the time the Strava bulk import was initiated.
    is_strava_bulk_import: bool = False,  # Boolean to track if we are doing a Strava bulk import or not
    strava_activities: dict
    | None = None,  # dictionary with info for a Strava bulk import - format strava_activities["filename"]["column header from Strava activities spreadsheet"]
) -> dict:
    """
    Creates the "import_info" dictionary that is added to all activities that are imported from files.

    Functions both for Strava imports and generic bulk imports, depending on whether the is_strava_bulk_import variable is set or not.

    Returns: Dictionary that contains the import_dict values for the activity (which is then added to the activity as a dictionary in the "import_info" field of the activity).
    """
    import_dict: dict[str, Any] = {}
    if is_strava_bulk_import:
        import_dict = activity_ingestion_integration.build_import_record(import_initiated_time, "Strava bulk import")
        activity_id_value = strava_activities[file_base_name].get("Activity ID", "")
        if activity_id_value:
            try:
                import_dict["strava_activity_id"] = int(activity_id_value)
            except (TypeError, ValueError):
                logger.warning(
                    f"Bulk file import: Ignoring invalid Strava Activity ID for {file_base_name}.",
                    extra=core_logger.context(console=True),
                )
    else:
        import_dict = activity_ingestion_integration.build_import_record(import_initiated_time)
    return import_dict

build_metadata_dict

build_metadata_dict(file_base_name, strava_activities, import_initiated_time, users_existing_gear_nickname_to_id=None)

Creates a dictionary with metadata information pulled from a Strava activities file.

The field strava_activity_metadata["metadata_found_in_csv"] identifies whether there was an entry for this activity in the activities.csv file or not.

Returns: Dictionary that contains the gear name as a key to look up the gear ID.

Source code in backend/app/modules/strava/bulk_import_utils.py

def build_metadata_dict(
    file_base_name: str,  # String with the base filename being processed (also key to strava_activities dictionary)
    strava_activities: dict,  # dictionary with info for a Strava bulk import - format strava_activities["filename"]["column header from Strava activities spreadsheet"]
    import_initiated_time: str,  # String containing the time the Strava bulk import was initiated.
    users_existing_gear_nickname_to_id: dict
    | None = None,  # Dictionary containing gear nickname to ID, needed for Strava bulk import
) -> dict:
    """
    Creates a dictionary with metadata information pulled from a Strava activities file.

    The field strava_activity_metadata["metadata_found_in_csv"] identifies whether there was an entry for this activity in the activities.csv file or not.

    Returns: Dictionary that contains the gear name as a key to look up the gear ID.
    """
    strava_activity_metadata: dict[str, Any] = {}
    if isinstance(strava_activities, dict) and strava_activities.get(
        file_base_name
    ):  # We have information on the activity
        # Strava bulk import notes:
        #     Importing Strava activity id to the activity's "strava_activity_id" field results in Endurain thinking the activity is linked to Strava via the Strava active linking mechanism.
        #     Strava media will be worked on after the activity has been created, so it is not dealt with here.
        strava_activity_metadata["name"] = strava_activities[file_base_name]["Activity Name"]
        strava_activity_metadata["description"] = strava_activities[file_base_name]["Activity Description"]
        strava_activity_metadata["activity type"] = strava_activities[file_base_name]["Activity Type"]

        # Needed for checking for duplicates within multi-file .fit file Strava bulk imports
        strava_activity_metadata["activity date"] = strava_activities[file_base_name]["Activity Date"]

        # Gear work
        activity_gear = None
        activity_gear = strava_activities[file_base_name]["Activity Gear"]
        if activity_gear and activity_gear is not None:
            if activity_gear.replace("+", " ").strip() in users_existing_gear_nickname_to_id:
                # Gear names in Endurain have all +'s swapped to spaces, thus need to do this here as well.
                strava_activity_metadata["gear_id"] = users_existing_gear_nickname_to_id[activity_gear][0]
            else:
                strava_activity_metadata["gear_id"] = None
                logger.info(
                    f"Bulk file import: Gear for activity {file_base_name}, which activities.csv shows as {activity_gear}, was not found in the user's existing gear. Not adding gear to activity.",
                    extra=core_logger.context(console=True),
                )
        else:
            strava_activity_metadata["gear_id"] = None
        import_dict = build_import_dictionary(file_base_name, import_initiated_time, True, strava_activities)
        strava_activity_metadata["import_dict"] = import_dict
        strava_activity_metadata["metadata_found_in_csv"] = True  # We found metadata in the CSV!
        logger.info(
            f"Bulk file import: Strava activities.csv metadata extracted for activity {file_base_name}.",
            extra=core_logger.context(console=True),
        )
    else:
        # We are in a Strava import, but don't have data on the file.  Just do a basic metadata addition.
        import_dict = build_import_dictionary(file_base_name, import_initiated_time, False)
        strava_activity_metadata["metadata_found_in_csv"] = (
            False  # No metadata found in CSV, so don't try to add it in later.
        )
        strava_activity_metadata["import_dict"] = import_dict

        logger.info(
            f"Bulk file import: No data in Strava activities.csv file for activity {file_base_name}.",
            extra=core_logger.context(console=True),
        )

    return strava_activity_metadata

create_activity_media_from_strava_bulk_import

create_activity_media_from_strava_bulk_import(activity_id, media_strava_filename, data, db)

Attaches a media file from a Strava export to an activity that was just imported.

The bytes are validated as an image through the unified upload pipeline before they are stored and become servable; a rejected file is logged and skipped so one bad photo does not fail the activity.

Parameters:

Name	Type	Description	Default
activity_id	int	The imported activity.	required
media_strava_filename	str	The file's name in the export.	required
data	bytes	The file's bytes.	required
db	Session	Database session.	required

Source code in backend/app/modules/strava/bulk_import_utils.py

def create_activity_media_from_strava_bulk_import(
    activity_id: int,
    media_strava_filename: str,
    data: bytes,
    db: Session,
) -> None:
    """
    Attaches a media file from a Strava export to an activity that was just imported.

    The bytes are validated as an image through the unified upload pipeline before
    they are stored and become servable; a rejected file is logged and skipped so
    one bad photo does not fail the activity.

    Args:
        activity_id: The imported activity.
        media_strava_filename: The file's name in the export.
        data: The file's bytes.
        db: Database session.
    """
    try:
        with tempfile.TemporaryDirectory(prefix="strava-media-") as work_dir:
            path = os.path.join(work_dir, "media")
            with open(path, "wb") as handle:
                handle.write(data)
            try:
                file_uploads.validate_local_file_sync(
                    path, kind=file_uploads.UploadKind.IMAGE, filename=media_strava_filename
                )
            except core_exceptions.DomainError as err:
                logger.warning(
                    f"Bulk file import media import: Rejecting {media_strava_filename} - failed image "
                    f"validation: {err.detail}",
                    extra=core_logger.context(console=True),
                )
                return

        # The media module owns the storage key and area; this only supplies the
        # validated bytes, which keeps the photo reachable on object storage and
        # only through the token-gated route.
        activity_media_integration.attach_media_bytes(activity_id, media_strava_filename, data, db)
        logger.info(
            f"Bulk file import media import: Media file {media_strava_filename} has been imported to db.",
            extra=core_logger.context(console=True),
        )
    except Exception as err:
        logger.error(
            f"Bulk file import media import: Error during processing of {media_strava_filename}: {err}",
            extra=core_logger.context(console=True),
        )

create_gear_dictionary_for_bulk_import

create_gear_dictionary_for_bulk_import(user_id, db)

Creates a dictionary that links gear nicknames in Endurain to their Endurain gear IDs.

The dictionary includes both base Endurain gear nickname and the gear name smoosh ("{brand} {model} {name}") that Strava uses for its shoe gear listing in activities.csv

Parameters:

Name	Type	Description	Default
user_id	int	The importing user.	required
db	Session	Database session.	required

Returns:

Type	Description
dict[str, list[int]]	Dictionary that uses the gear name as a key to look up the gear ID; empty
dict[str, list[int]]	when the user has no gear, which only means no activity is linked to one.

Source code in backend/app/modules/strava/bulk_import_utils.py

def create_gear_dictionary_for_bulk_import(user_id: int, db: Session) -> dict[str, list[int]]:
    """
    Creates a dictionary that links gear nicknames in Endurain to their Endurain gear IDs.

    The dictionary includes both base Endurain gear nickname and the gear name smoosh ("{brand} {model} {name}") that Strava uses for its shoe gear listing in activities.csv

    Args:
        user_id: The importing user.
        db: Database session.

    Returns:
        Dictionary that uses the gear name as a key to look up the gear ID; empty
        when the user has no gear, which only means no activity is linked to one.
    """
    gear_by_name: dict[str, list[int]] = {}
    for item in gears_crud.get_gear_user(user_id, db) or []:
        gear_by_name[item.nickname] = [item.id]
        # Strava apparently exports shoe names as a smoosh of "{brand} {model} {name}", so adding that as a second key for each gear item
        strava_name_smoosh = " ".join(
            part
            for part in [
                (item.brand or "").strip(),
                (item.model or "").strip(),
                (item.nickname or "").strip(),
            ]
            if part
        )
        gear_by_name.setdefault(strava_name_smoosh, [item.id])
    return gear_by_name

does_activity_start_time_match_the_data_in_strava_activities_csv

does_activity_start_time_match_the_data_in_strava_activities_csv(activity, activity_metadata_dict)

Strava includes each multi-activity .fit file once for each activity inside the fit file. Thus, run without filtering, a 5-activity fit file is imported 5 times for 25 activities imported.

So, to find out which activity goes with which file, we check if start time of activity aligns with start time of the activity's strava activities.csv file data. If it does import.

Formatting of times in the two files

Activity start time - from Endurain activity parser: 2023-10-21T07:41:47 Activity start time - from Strava activities.csv: Oct 21, 2023, 8:13:28 AM

Returns: True if the start times match. False if they do not.

Source code in backend/app/modules/strava/bulk_import_utils.py

def does_activity_start_time_match_the_data_in_strava_activities_csv(
    activity: activity_contracts.ActivityCore,  # A parsed activity about to be added to Endurain
    activity_metadata_dict: dict,  # A dictionary containing parsed information from a Strava activities.csv file
) -> bool:
    """
    Strava includes each multi-activity .fit file once for each activity inside the fit file.  Thus, run without filtering, a 5-activity fit file is imported 5 times for 25 activities imported.

    So, to find out which activity goes with which file, we check if start time of activity aligns with start time of the activity's strava activities.csv file data.  If it does import.

    Formatting of times in the two files:
        Activity start time - from Endurain activity parser: 2023-10-21T07:41:47
        Activity start time - from Strava activities.csv: Oct 21, 2023, 8:13:28 AM

    Returns: True if the start times match.  False if they do not.
    """
    # ``ActivityCore`` normalizes start_time to an aware UTC datetime at
    # construction, but accepts a string — go through the shared coercion rather
    # than assuming either. Parsing this as a string unconditionally raised
    # TypeError once the contract started coercing, which failed the whole file
    # and sent every multi-activity .fit in a Strava export to the error dir.
    endurain_parsed_file_start_date = core_timezone.to_utc_aware(activity.start_time)
    if endurain_parsed_file_start_date is None:
        return False
    strava_csv_start_date = datetime.strptime(activity_metadata_dict["activity date"], "%b %d, %Y, %-I:%-M:%-S %p")
    # Ensure both are tz-aware (or both naive) for comparison
    if strava_csv_start_date.tzinfo is None:
        strava_csv_start_date = strava_csv_start_date.replace(tzinfo=UTC)
    return endurain_parsed_file_start_date == strava_csv_start_date

import_media_from_strava_bulk_export

import_media_from_strava_bulk_export(strava_activities, created_activity, file_base_name, read_member, db)

Import media files associated with a Strava bulk export activity.

Reads the pipe-separated media list from the Strava activities.csv entry for the given file, then attaches each referenced image found in the export's media folder.

Parameters:

Name	Type	Description	Default
strava_activities	dict	Strava bulk-import metadata dict keyed by filename, then by activities.csv column header.	required
created_activity	ActivityRead	The newly created activity object to which media will be attached.	required
file_base_name	str	Base filename of the activity file, used as the lookup key in strava_activities.	required
read_member	Callable[[str], bytes | None]	Reads a member of the archive being imported by name.	required
db	Session	SQLAlchemy database session.	required

Source code in backend/app/modules/strava/bulk_import_utils.py

def import_media_from_strava_bulk_export(
    strava_activities: dict,
    created_activity: activity_schema.ActivityRead,
    file_base_name: str,
    read_member: Callable[[str], bytes | None],
    db: Session,
) -> None:
    """
    Import media files associated with a Strava bulk export activity.

    Reads the pipe-separated media list from the Strava ``activities.csv`` entry
    for the given file, then attaches each referenced image found in the export's
    ``media`` folder.

    Args:
        strava_activities: Strava bulk-import metadata dict keyed by filename,
            then by activities.csv column header.
        created_activity: The newly created activity object to which media will
            be attached.
        file_base_name: Base filename of the activity file, used as the lookup key
            in strava_activities.
        read_member: Reads a member of the archive being imported by name.
        db: SQLAlchemy database session.
    """
    if created_activity.id is None:
        logger.warning(
            f"Bulk file import: cannot import media for {file_base_name} - created activity has no id",
            extra=core_logger.context(console=True),
        )
        return
    row = strava_activities.get(file_base_name)
    if not row:
        return
    media_string = (row.get("Media") or "").strip()
    if not media_string:
        logger.info(
            f"Bulk file import: Media import section - no media list in activities.csv for {file_base_name}",
            extra=core_logger.context(console=True),
        )
        return
    for media_item in media_string.split("|"):
        media_file_base_name = os.path.basename(media_item)
        if not media_file_base_name:
            continue
        data = read_member(f"{MEDIA_FOLDER}/{media_file_base_name}")
        if data is None:
            logger.warning(
                f"Bulk file import media import warning: Media file {media_file_base_name} is not in the export, "
                "skipping import of it",
                extra=core_logger.context(console=True),
            )
            continue
        create_activity_media_from_strava_bulk_import(created_activity.id, media_file_base_name, data, db)

parse_activities_manifest

parse_activities_manifest(manifest_path)

Parses a Strava activities.csv manifest.

Parameters:

Name	Type	Description	Default
manifest_path	str	Path of the manifest file.	required

Returns:

Type	Description
dict | None	Dictionary of the manifest's rows keyed by each activity's filename, or
dict | None	None when the file cannot be parsed or lacks the required headers.

Source code in backend/app/modules/strava/bulk_import_utils.py

def parse_activities_manifest(manifest_path: str) -> dict | None:
    """
    Parses a Strava ``activities.csv`` manifest.

    Args:
        manifest_path: Path of the manifest file.

    Returns:
        Dictionary of the manifest's rows keyed by each activity's filename, or
        ``None`` when the file cannot be parsed or lacks the required headers.
    """
    strava_activities_file_name = os.path.basename(manifest_path)
    try:
        strava_activities_dict = {}
        for row in core_text_imports.read_bounded_csv(manifest_path):
            # Check to see if file has headers that will be used during parsing of the file.
            missing_headers = _STRAVA_ACTIVITIES_HEADERS - row.keys()
            if missing_headers:
                logger.error(
                    "Aborting Strava bulk activities import: Proper headers not found in "
                    f"{strava_activities_file_name}. Missing: {', '.join(sorted(missing_headers))}.",
                    extra=core_logger.context(console=True),
                )
                return None
            _, strava_act_file_name = os.path.split(row["Filename"])  # strips path, returns filename with extension.
            strava_activities_dict[strava_act_file_name] = row
        logger.info(
            f"Strava bulk import: Strava activities csv file parsed, and it is {len(strava_activities_dict)} rows long",
            extra=core_logger.context(console=True),
        )
        return strava_activities_dict
    except HTTPException as http_err:
        # ``read_bounded_csv`` raises HTTP 413/424 for oversized files or stat/open
        # failures; surface the structured detail instead of a bare ``None``.
        logger.error(
            f"Strava activities CSV parsing aborted: {http_err.detail}", extra=core_logger.context(console=True)
        )
        return None
    except Exception as err:
        logger.error(
            f"Strava activities CSV parsing failed with error: {err}.", extra=core_logger.context(console=True)
        )
        return None

Utilities

StravaRateLimitTracker

Tracks Strava API rate limit state in the platform's shared state.

The limits are Strava's, per application, so every process and replica has to see the same answer: a process-local flag would let each worker keep hammering an API another one was already told to back off from. The flags are TTL keys in the platform StateProvider (in-process under the local profile, Redis when distributed), so they expire on their own.

Source code in backend/app/modules/strava/utils.py

class StravaRateLimitTracker:
    """Tracks Strava API rate limit state in the platform's shared state.

    The limits are Strava's, per application, so every process and replica has to
    see the same answer: a process-local flag would let each worker keep
    hammering an API another one was already told to back off from. The flags are
    TTL keys in the platform ``StateProvider`` (in-process under the ``local``
    profile, Redis when distributed), so they expire on their own.
    """

    # Short-term limits reset every 15 minutes
    SHORT_TERM_COOLDOWN_SECONDS = 15 * 60

    _SHORT_TERM_KEY = "strava:rate_limited:short"
    _LONG_TERM_KEY = "strava:rate_limited:long"

    @staticmethod
    def _state() -> platform_providers.StateProvider:
        return platform_runtime.get_active_platform().state

    def mark_rate_limited(self, is_long_term: bool = False) -> None:
        """
        Record that a rate limit was hit.

        Args:
            is_long_term: True for the daily limit (resets at UTC midnight),
                False for the 15-minute limit.
        """
        now = datetime.now(UTC)
        if is_long_term:
            tomorrow = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
            key, ttl = self._LONG_TERM_KEY, max(1, int((tomorrow - now).total_seconds()))
            logger.warning(f"Strava daily rate limit hit. Skipping until {tomorrow}")
        else:
            key, ttl = self._SHORT_TERM_KEY, self.SHORT_TERM_COOLDOWN_SECONDS
            logger.warning(f"Strava 15-min rate limit hit. Skipping for {ttl} seconds")
        try:
            self._state().set(key, b"1", ttl_seconds=ttl)
        except platform_providers.StateBackendUnavailableError as err:
            logger.error("Could not record the Strava rate limit", exc_info=err)

    def is_rate_limited(self) -> bool:
        """
        Check if Strava API calls should be skipped.

        Fails open when the shared state is unreachable: a missed back-off costs
        one rejected call, which marks the limit again, whereas failing closed
        would stop every sync for as long as the state store is down.

        Returns:
            True if currently rate-limited.
        """
        try:
            state = self._state()
            return state.get(self._LONG_TERM_KEY) is not None or state.get(self._SHORT_TERM_KEY) is not None
        except platform_providers.StateBackendUnavailableError as err:
            logger.warning("Could not read the Strava rate limit state; assuming not limited", exc_info=err)
            return False

is_rate_limited

is_rate_limited()

Check if Strava API calls should be skipped.

Fails open when the shared state is unreachable: a missed back-off costs one rejected call, which marks the limit again, whereas failing closed would stop every sync for as long as the state store is down.

Returns:

Type	Description
bool	True if currently rate-limited.

Source code in backend/app/modules/strava/utils.py

def is_rate_limited(self) -> bool:
    """
    Check if Strava API calls should be skipped.

    Fails open when the shared state is unreachable: a missed back-off costs
    one rejected call, which marks the limit again, whereas failing closed
    would stop every sync for as long as the state store is down.

    Returns:
        True if currently rate-limited.
    """
    try:
        state = self._state()
        return state.get(self._LONG_TERM_KEY) is not None or state.get(self._SHORT_TERM_KEY) is not None
    except platform_providers.StateBackendUnavailableError as err:
        logger.warning("Could not read the Strava rate limit state; assuming not limited", exc_info=err)
        return False

mark_rate_limited

mark_rate_limited(is_long_term=False)

Record that a rate limit was hit.

Parameters:

Name	Type	Description	Default
is_long_term	bool	True for the daily limit (resets at UTC midnight), False for the 15-minute limit.	False

Source code in backend/app/modules/strava/utils.py

def mark_rate_limited(self, is_long_term: bool = False) -> None:
    """
    Record that a rate limit was hit.

    Args:
        is_long_term: True for the daily limit (resets at UTC midnight),
            False for the 15-minute limit.
    """
    now = datetime.now(UTC)
    if is_long_term:
        tomorrow = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
        key, ttl = self._LONG_TERM_KEY, max(1, int((tomorrow - now).total_seconds()))
        logger.warning(f"Strava daily rate limit hit. Skipping until {tomorrow}")
    else:
        key, ttl = self._SHORT_TERM_KEY, self.SHORT_TERM_COOLDOWN_SECONDS
        logger.warning(f"Strava 15-min rate limit hit. Skipping for {ttl} seconds")
    try:
        self._state().set(key, b"1", ttl_seconds=ttl)
    except platform_providers.StateBackendUnavailableError as err:
        logger.error("Could not record the Strava rate limit", exc_info=err)

is_strava_not_found_error

is_strava_not_found_error(err)

Check if an exception is a Strava 404 Not Found error.

This occurs for activities that have no streams, such as those manually added via Apple Health and imported into Strava.

Parameters:

Name	Type	Description	Default
err	Exception	The exception to check.	required

Returns:

Type	Description
bool	True if the error is a 404 not-found response.

Source code in backend/app/modules/strava/utils.py

def is_strava_not_found_error(err: Exception) -> bool:
    """
    Check if an exception is a Strava 404 Not Found error.

    This occurs for activities that have no streams, such as
    those manually added via Apple Health and imported into Strava.

    Args:
        err: The exception to check.

    Returns:
        True if the error is a 404 not-found response.
    """
    if isinstance(err, StravaFault):
        response = getattr(err, "response", None)
        if response is not None and getattr(response, "status_code", 0) == 404:
            return True
    err_str = str(err).lower()
    return "not found" in err_str or "404" in err_str

is_strava_rate_limit_error

is_strava_rate_limit_error(err)

Check if an exception is a Strava rate limit error.

Parameters:

Name	Type	Description	Default
err	Exception	The exception to check.	required

Returns:

Type	Description
bool	True if the error is a 429 rate limit response.

Source code in backend/app/modules/strava/utils.py

def is_strava_rate_limit_error(err: Exception) -> bool:
    """
    Check if an exception is a Strava rate limit error.

    Args:
        err: The exception to check.

    Returns:
        True if the error is a 429 rate limit response.
    """
    if isinstance(err, StravaFault):
        response = getattr(err, "response", None)
        if response is not None and getattr(response, "status_code", 0) == 429:
            return True
    err_str = str(err).lower()
    return "rate limit" in err_str or "429" in err_str

refresh_user_strava_token

refresh_user_strava_token(user_id)

Refresh a user's Strava access token when it is about to expire.

Called at the start of the user's own sync job, so tokens are kept current without a separate pass over every user. The integration is read in one short session, the Strava call runs with none open, and the new tokens are written in another. A failure is logged and left to the sync that follows, which will fail visibly if the token really is unusable.

Parameters:

Name	Type	Description	Default
user_id	int	The user whose token to refresh.	required

Source code in backend/app/modules/strava/utils.py

def refresh_user_strava_token(user_id: int) -> None:
    """Refresh a user's Strava access token when it is about to expire.

    Called at the start of the user's own sync job, so tokens are kept current
    without a separate pass over every user. The integration is read in one short
    session, the Strava call runs with none open, and the new tokens are written in
    another. A failure is logged and left to the sync that follows, which will fail
    visibly if the token really is unusable.

    Args:
        user_id: The user whose token to refresh.
    """
    if rate_limit_tracker.is_rate_limited():
        logger.warning(f"User {user_id}: Strava rate limit active, skipping token refresh")
        return

    with SessionLocal() as db:
        user_integrations = user_integrations_crud.get_user_integrations_by_user_id(user_id, db)

    if user_integrations is None:
        logger.info(f"User {user_id}: User integrations not found. Will skip processing")
        return

    if not (
        user_integrations.strava_token is not None
        and user_integrations.strava_refresh_token is not None
        and user_integrations.strava_token_expires_at is not None
        and user_integrations.strava_client_id is not None
        and user_integrations.strava_client_secret is not None
    ):
        logger.info(f"User {user_id}: No Strava token found. Will skip processing")
        return

    refresh_time = user_integrations.strava_token_expires_at.replace(tzinfo=UTC) - timedelta(minutes=60)
    if datetime.now(UTC) <= refresh_time:
        return

    try:
        strava_client = create_strava_client(user_integrations)
        tokens = strava_client.refresh_access_token(
            client_id=core_cryptography.decrypt_token_fernet(user_integrations.strava_client_id),
            client_secret=core_cryptography.decrypt_token_fernet(user_integrations.strava_client_secret),
            refresh_token=core_cryptography.decrypt_token_fernet(user_integrations.strava_refresh_token),
        )
    except Exception as err:
        if is_strava_rate_limit_error(err):
            err_str = str(err).lower()
            is_long = "long" in err_str or "daily" in err_str
            rate_limit_tracker.mark_rate_limited(is_long_term=is_long)
            logger.warning(f"User {user_id}: Strava rate limit hit during token refresh")
            return
        logger.error(f"User {user_id}: Error refreshing the Strava token", exc_info=err)
        return

    with SessionLocal() as db:
        user_integrations_crud.link_strava_account(user_id, tokens, db)
    logger.info(f"User {user_id}: Strava tokens refreshed")

Schemas