API Reference

Profile module for user profile management and operations.

This module provides functionality for: - User profile retrieval and updates - MFA (Multi-Factor Authentication) setup and management - Profile data export and import (ZIP archives) - Identity provider linking and management - Session management

Exports

• Services: ExportService, ImportService
• Exceptions: ProfileOperationError, ExportError, ProfileImportError, and related exception classes

ActivityLimitError

Bases: ProfileImportError

Raised when import contains too many activities.

Source code in backend/app/users/users_profile/exceptions.py

class ActivityLimitError(ProfileImportError):
    """
    Raised when import contains too many activities.
    """

    pass

DataCollectionError

Bases: ExportError

Raised when data collection from database fails.

Source code in backend/app/users/users_profile/exceptions.py

class DataCollectionError(ExportError):
    """
    Raised when data collection from database fails.
    """

    pass

DataIntegrityError

Bases: ProfileImportError

Raised when imported data has integrity issues.

Source code in backend/app/users/users_profile/exceptions.py

class DataIntegrityError(ProfileImportError):
    """
    Raised when imported data has integrity issues.
    """

    pass

DatabaseConnectionError

Bases: ExportError

Raised when database connection fails during export.

Source code in backend/app/users/users_profile/exceptions.py

class DatabaseConnectionError(ExportError):
    """
    Raised when database connection fails during export.
    """

    pass

DiskSpaceError

Bases: ProfileImportError

Raised when insufficient disk space is available.

Source code in backend/app/users/users_profile/exceptions.py

class DiskSpaceError(ProfileImportError):
    """
    Raised when insufficient disk space is available.
    """

    pass

ExportError

Bases: ProfileOperationError

Base exception for export operations.

Source code in backend/app/users/users_profile/exceptions.py

class ExportError(ProfileOperationError):
    """
    Base exception for export operations.
    """

    pass

ExportService

Service for exporting user profile data to ZIP archive.

Attributes:

Name	Type	Description
user_id		ID of user to export data for.
db		Database session.
counts		Dictionary tracking exported item counts.
performance_config	ExportPerformanceConfig	Performance configuration.

Source code in backend/app/users/users_profile/export_service.py

class ExportService:
    """
    Service for exporting user profile data to ZIP archive.

    Attributes:
        user_id: ID of user to export data for.
        db: Database session.
        counts: Dictionary tracking exported item counts.
        performance_config: Performance configuration.
    """

    def __init__(
        self,
        user_id: int,
        db: Session,
        performance_config: ExportPerformanceConfig | None = None,
    ):
        self.user_id = user_id
        self.db = db
        self.counts = profile_utils.initialize_operation_counts(include_user_count=True)
        self.performance_config: ExportPerformanceConfig = (
            performance_config or ExportPerformanceConfig.get_auto_config()
        )

        core_logger.print_to_log(
            f"ExportService initialized with performance config: "
            f"batch_size={self.performance_config.batch_size}, "
            f"max_memory_mb={self.performance_config.max_memory_mb}, "
            f"compression_level={self.performance_config.compression_level}, "
            f"timeout_seconds={self.performance_config.timeout_seconds}",
            "info",
        )

    def collect_user_activities_data(self, zipf: zipfile.ZipFile) -> list[Any]:
        """
        Collect and write user activities to ZIP.

        Args:
            zipf: ZipFile instance to write to.

        Returns:
            List of collected activity objects.

        Raises:
            DatabaseConnectionError: If database error occurs.
            MemoryAllocationError: If memory limit exceeded.
            DataCollectionError: If collection fails.
        """
        try:
            profile_utils.check_memory_usage(
                "activity collection start",
                self.performance_config.max_memory_mb,
                self.performance_config.enable_memory_monitoring,
            )

            # Get activities in batches using pagination
            all_activities = []
            offset = 0
            batch_size = self.performance_config.batch_size

            core_logger.print_to_log(
                f"Starting batched activity collection with batch_size={batch_size}",
                "info",
            )

            while True:
                # Get a batch of activities
                try:
                    batch_activities = self._get_activities_batch(offset, batch_size)
                except (SQLAlchemyError, MemoryAllocationError):
                    # Persistent failures must propagate so the outer handlers can
                    # return an explicit error rather than silently skipping pages.
                    raise
                except Exception as err:
                    core_logger.print_to_log(
                        f"Failed to get activities batch (offset={offset}), skipping: {err}",
                        "warning",
                        exc=err,
                    )
                    offset += batch_size
                    continue

                if not batch_activities:
                    break

                all_activities.extend(batch_activities)
                offset += batch_size

                # Check memory usage after each batch
                profile_utils.check_memory_usage(
                    f"activity batch {offset // batch_size}",
                    self.performance_config.max_memory_mb,
                    self.performance_config.enable_memory_monitoring,
                )

                core_logger.print_to_log(
                    f"Collected {len(batch_activities)} activities in batch (total: {len(all_activities)})",
                    "info",
                )

            if not all_activities:
                core_logger.print_to_log(f"No activities found for user {self.user_id}", "info")
                # Write empty activities file
                profile_utils.write_json_to_zip(zipf, "data/activities.json", [], self.counts)
                return []

            # Write activities to ZIP immediately
            activities_dicts = [profile_utils.sqlalchemy_obj_to_dict(a) for a in all_activities]
            profile_utils.write_json_to_zip(zipf, "data/activities.json", activities_dicts, self.counts)

            core_logger.print_to_log(
                f"Written {len(activities_dicts)} activities to ZIP",
                "info",
            )

            # Filter out activities with None IDs and collect valid IDs
            activity_ids = [activity.id for activity in all_activities if activity.id is not None]

            if not activity_ids:
                core_logger.print_to_log(f"No valid activity IDs found for user {self.user_id}", "warning")
                return all_activities

            # Collect and write activity components progressively
            self._collect_and_write_activity_components(zipf, activity_ids, all_activities)

            # Exercise titles don't depend on activity IDs
            try:
                exercise_titles = activity_exercise_titles_crud.get_activity_exercise_titles(self.db)
                if exercise_titles:
                    exercise_titles_dicts = [profile_utils.sqlalchemy_obj_to_dict(e) for e in exercise_titles]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/activity_exercise_titles.json",
                        exercise_titles_dicts,
                        self.counts,
                    )
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect exercise titles: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect exercise titles: {err}") from err

        except SQLAlchemyError as err:
            core_logger.print_to_log(f"Database error collecting activities: {err}", "error", exc=err)
            raise DatabaseConnectionError(f"Failed to collect activity data: {err}") from err
        except MemoryAllocationError as err:
            core_logger.print_to_log(
                f"Memory limit exceeded while collecting activities: {err}. ",
                "error",
                exc=err,
            )
            raise err
        except DataCollectionError:
            raise
        except Exception as err:
            core_logger.print_to_log(f"Unexpected error collecting activities: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect activity data: {err}") from err

        return all_activities

    def _get_activities_batch(self, offset: int, limit: int) -> list[Any]:
        """
        Get batch of activities using pagination.

        Args:
            offset: Offset for pagination.
            limit: Number of items per batch.

        Returns:
            List of activity objects for the batch.
        """
        try:
            # Convert offset to page number (1-based indexing)
            page_number = (offset // limit) + 1

            # Use the existing pagination function from activities CRUD
            activities = activities_crud.get_user_activities_with_pagination(
                user_id=self.user_id,
                db=self.db,
                page_number=page_number,
                num_records=limit,
                # Sort by start_time descending (most recent first) for consistency
                sort_by="start_time",
                sort_order="desc",
                user_is_owner=True,
            )

            return activities or []

        except Exception as err:
            core_logger.print_to_log(
                f"Failed to get activities batch (offset={offset}, limit={limit}): {err}",
                "error",
                exc=err,
            )
            raise

    def _collect_and_write_activity_components(
        self,
        zipf: zipfile.ZipFile,
        activity_ids: list[int],
        user_activities: list[Any],
    ) -> None:
        """
        Collect and write activity components to ZIP.

        Args:
            zipf: ZipFile instance to write to.
            activity_ids: List of activity IDs to process.
            user_activities: List of activity objects.
        """
        # Process activity IDs in smaller batches to reduce memory usage
        batch_size = self.performance_config.batch_size // 2  # Smaller batches for components

        # Component definitions: (key, filename, crud_function, should_split)
        component_types = [
            (
                "laps",
                "data/activity_laps.json",
                activity_laps_crud.get_activities_laps,
                True,
            ),
            (
                "sets",
                "data/activity_sets.json",
                activity_sets_crud.get_activities_sets,
                True,
            ),
            (
                "streams",
                "data/activity_streams.json",
                activity_streams_crud.get_activities_streams,
                True,
            ),
            (
                "steps",
                "data/activity_workout_steps.json",
                activity_workout_steps_crud.get_activities_workout_steps,
                False,
            ),
            (
                "media",
                "data/activity_media.json",
                activity_media_crud.get_activities_media,
                False,
            ),
        ]

        for component_key, base_filename, crud_func, should_split in component_types:
            # For large splittable components, write in chunks during collection
            if should_split:
                self._collect_and_write_component_chunked(
                    zipf,
                    component_key,
                    base_filename,
                    crud_func,
                    activity_ids,
                    user_activities,
                    batch_size,
                )
            else:
                # For small components, collect all then write
                self._collect_and_write_component_simple(
                    zipf,
                    component_key,
                    base_filename,
                    crud_func,
                    activity_ids,
                    user_activities,
                    batch_size,
                )

    def _collect_and_write_component_chunked(
        self,
        zipf: zipfile.ZipFile,
        component_key: str,
        base_filename: str,
        crud_func,
        activity_ids: list[int],
        user_activities: list[Any],
        batch_size: int,
    ) -> None:
        """
        Collect and write large components in chunks.

        Args:
            zipf: ZipFile instance to write to.
            component_key: Component type identifier.
            base_filename: Base name for output files.
            crud_func: CRUD function to fetch data.
            activity_ids: List of activity IDs.
            user_activities: List of activity objects.
            batch_size: Number of items per batch.
        """
        chunk_buffer = []
        file_counter = 0
        max_items_per_file = batch_size
        total_items = 0

        # Collect component data in batches
        for i in range(0, len(activity_ids), batch_size):
            batch_ids = activity_ids[i : i + batch_size]
            batch_activities = user_activities[i : i + batch_size]

            profile_utils.check_memory_usage(
                f"{component_key} batch {i // batch_size + 1}",
                self.performance_config.max_memory_mb,
                self.performance_config.enable_memory_monitoring,
            )

            try:
                data = crud_func(batch_ids, self.user_id, self.db, batch_activities)
                if data:
                    # Convert to dicts and add to chunk buffer
                    batch_dicts = [profile_utils.sqlalchemy_obj_to_dict(item) for item in data]
                    chunk_buffer.extend(batch_dicts)
                    total_items += len(batch_dicts)

                    # Write chunks to ZIP when buffer reaches max size
                    while len(chunk_buffer) >= max_items_per_file:
                        chunk_to_write = chunk_buffer[:max_items_per_file]
                        chunk_buffer = chunk_buffer[max_items_per_file:]

                        # Generate filename for this chunk
                        base_name = base_filename.rsplit(".", 1)[0]
                        extension = base_filename.rsplit(".", 1)[1] if "." in base_filename else "json"
                        chunk_filename = f"{base_name}_{file_counter:03d}.{extension}"

                        profile_utils.write_json_to_zip(zipf, chunk_filename, chunk_to_write, self.counts)
                        file_counter += 1

                        core_logger.print_to_log(
                            f"Written chunk {file_counter} for {component_key} ({len(chunk_to_write)} items)",
                            "debug",
                        )

            except Exception as err:
                core_logger.print_to_log(
                    f"Failed to collect batch for {component_key}: {err}",
                    "error",
                    exc=err,
                )
                raise DataCollectionError(f"Failed to collect {component_key}: {err}") from err

            core_logger.print_to_log(
                f"Processed {component_key} batch {i // batch_size + 1} ({len(batch_ids)} activities)",
                "info",
            )

        # Write remaining data in buffer
        if chunk_buffer:
            if file_counter == 0:
                # Only one chunk, use original filename
                profile_utils.write_json_to_zip(zipf, base_filename, chunk_buffer, self.counts)
                core_logger.print_to_log(
                    f"Written {len(chunk_buffer)} {component_key} items to single file",
                    "info",
                )
            else:
                # Multiple chunks, write with numbered filename
                base_name = base_filename.rsplit(".", 1)[0]
                extension = base_filename.rsplit(".", 1)[1] if "." in base_filename else "json"
                chunk_filename = f"{base_name}_{file_counter:03d}.{extension}"

                profile_utils.write_json_to_zip(zipf, chunk_filename, chunk_buffer, self.counts)
                file_counter += 1
                core_logger.print_to_log(
                    f"Written final chunk for {component_key} ({len(chunk_buffer)} items)",
                    "debug",
                )

        if total_items == 0:
            # Write empty file for component type
            profile_utils.write_json_to_zip(zipf, base_filename, [], self.counts)
            core_logger.print_to_log(
                f"No {component_key} data found, written empty file",
                "info",
            )
        else:
            core_logger.print_to_log(
                f"Written total {total_items} {component_key} items to {file_counter} file(s)",
                "info",
            )

    def _collect_and_write_component_simple(
        self,
        zipf: zipfile.ZipFile,
        component_key: str,
        base_filename: str,
        crud_func,
        activity_ids: list[int],
        user_activities: list[Any],
        batch_size: int,
    ) -> None:
        """
        Collect and write small components in single file.

        Args:
            zipf: ZipFile instance to write to.
            component_key: Component type identifier.
            base_filename: Name for output file.
            crud_func: CRUD function to fetch data.
            activity_ids: List of activity IDs.
            user_activities: List of activity objects.
            batch_size: Number of items per batch.
        """
        all_component_data = []

        # Collect component data in batches
        for i in range(0, len(activity_ids), batch_size):
            batch_ids = activity_ids[i : i + batch_size]
            batch_activities = user_activities[i : i + batch_size]

            profile_utils.check_memory_usage(
                f"{component_key} batch {i // batch_size + 1}",
                self.performance_config.max_memory_mb,
                self.performance_config.enable_memory_monitoring,
            )

            try:
                data = crud_func(batch_ids, self.user_id, self.db, batch_activities)
                if data:
                    all_component_data.extend(data)
            except Exception as err:
                core_logger.print_to_log(
                    f"Failed to collect batch for {component_key}: {err}",
                    "error",
                    exc=err,
                )
                raise DataCollectionError(f"Failed to collect {component_key}: {err}") from err

            core_logger.print_to_log(
                f"Processed {component_key} batch {i // batch_size + 1} ({len(batch_ids)} activities)",
                "info",
            )

        # Write all component data to ZIP
        if all_component_data:
            component_dicts = [profile_utils.sqlalchemy_obj_to_dict(item) for item in all_component_data]
            profile_utils.write_json_to_zip(zipf, base_filename, component_dicts, self.counts)
            core_logger.print_to_log(
                f"Written {len(component_dicts)} {component_key} items to ZIP",
                "info",
            )
            # Clear from memory
            all_component_data.clear()
            component_dicts.clear()
        else:
            # Write empty file for component type
            profile_utils.write_json_to_zip(zipf, base_filename, [], self.counts)
            core_logger.print_to_log(
                f"No {component_key} data found, written empty file",
                "info",
            )

    def collect_gear_data(self, zipf: zipfile.ZipFile) -> None:
        """
        Collect and write gear data to ZIP.

        Args:
            zipf: ZipFile instance to write to.

        Raises:
            DatabaseConnectionError: If database error occurs.
        """
        try:
            # Collect and write gears
            try:
                gears = gear_crud.get_gear_user(self.user_id, self.db)
                if gears:
                    gears_dicts = [profile_utils.sqlalchemy_obj_to_dict(g) for g in gears]
                    profile_utils.write_json_to_zip(zipf, "data/gears.json", gears_dicts, self.counts)
                else:
                    profile_utils.write_json_to_zip(zipf, "data/gears.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect gears: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect gears: {err}") from err

            # Collect and write gear components
            try:
                gear_components = gear_components_crud.get_gear_components_user(self.user_id, self.db)
                if gear_components:
                    gear_components_dicts = [profile_utils.sqlalchemy_obj_to_dict(gc) for gc in gear_components]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/gear_components.json",
                        gear_components_dicts,
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(zipf, "data/gear_components.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect gear components: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect gear components: {err}") from err

        except SQLAlchemyError as err:
            core_logger.print_to_log(f"Database error collecting gear data: {err}", "error", exc=err)
            raise DatabaseConnectionError(f"Failed to collect gear data: {err}") from err

    def collect_health_weight(self, zipf: zipfile.ZipFile) -> None:
        """
        Collect and write health data to ZIP.

        Args:
            zipf: ZipFile instance to write to.

        Raises:
            DatabaseConnectionError: If database error occurs.
        """
        try:
            # Collect and write health data
            try:
                health_weight = health_weight_crud.get_all_health_weight_by_user_id(self.user_id, self.db)
                if health_weight:
                    health_weight_dicts = [profile_utils.sqlalchemy_obj_to_dict(hd) for hd in health_weight]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/health_weight.json",
                        health_weight_dicts,
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(zipf, "data/health_weight.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect health data: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect health data: {err}") from err

            # Collect and write health targets
            try:
                health_targets = health_targets_crud.get_health_targets_by_user_id(self.user_id, self.db)
                if health_targets:
                    # health_targets is a single object, not a list
                    health_targets_dict = profile_utils.sqlalchemy_obj_to_dict(health_targets)
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/health_targets.json",
                        [health_targets_dict],
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(zipf, "data/health_targets.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect health targets: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect health targets: {err}") from err

        except SQLAlchemyError as err:
            core_logger.print_to_log(f"Database error collecting health data: {err}", "error", exc=err)
            raise DatabaseConnectionError(f"Failed to collect health data: {err}") from err

    def collect_notifications_data(self, zipf: zipfile.ZipFile) -> None:
        try:
            try:
                notifications = notifications_crud.get_user_notifications(self.user_id, self.db)
                if notifications:
                    notifications_dicts = [profile_utils.sqlalchemy_obj_to_dict(n) for n in notifications]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/notifications.json",
                        notifications_dicts,
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(zipf, "data/notifications.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect user notifications: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect user notifications: {err}") from err
        except SQLAlchemyError as err:
            core_logger.print_to_log(f"Database error collecting user notifications: {err}", "error", exc=err)
            raise DatabaseConnectionError(f"Failed to collect user notifications: {err}") from err

    def collect_user_settings_data(self, zipf: zipfile.ZipFile) -> None:
        """
        Collect and write user settings to ZIP.

        Args:
            zipf: ZipFile instance to write to.

        Raises:
            DatabaseConnectionError: If database error occurs.
        """
        try:
            # Collect and write user default gear
            try:
                user_default_gear = user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)
                if user_default_gear:
                    default_gear_dict = [profile_utils.sqlalchemy_obj_to_dict(user_default_gear)]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/user_default_gear.json",
                        default_gear_dict,
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(zipf, "data/user_default_gear.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect user default gear: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect user default gear: {err}") from err

            # Collect and write user goals
            try:
                user_goals = user_goals_crud.get_user_goals_by_user_id(self.user_id, self.db)
                if user_goals:
                    user_goals_dicts = [profile_utils.sqlalchemy_obj_to_dict(ug) for ug in user_goals]
                    profile_utils.write_json_to_zip(zipf, "data/user_goals.json", user_goals_dicts, self.counts)
                else:
                    profile_utils.write_json_to_zip(zipf, "data/user_goals.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect user goals: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect user goals: {err}") from err

            # Collect and write user integrations
            try:
                user_integrations = user_integrations_crud.get_user_integrations_by_user_id(self.user_id, self.db)
                if user_integrations:
                    integrations_dict = [profile_utils.sqlalchemy_obj_to_dict(user_integrations)]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/user_integrations.json",
                        integrations_dict,
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(zipf, "data/user_integrations.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(f"Failed to collect user integrations: {err}", "error", exc=err)
                raise DataCollectionError(f"Failed to collect user integrations: {err}") from err

            # Collect and write user privacy settings
            try:
                user_privacy_settings = users_privacy_settings_crud.get_user_privacy_settings_by_user_id(
                    self.user_id, self.db
                )
                if user_privacy_settings:
                    privacy_dict = [profile_utils.sqlalchemy_obj_to_dict(user_privacy_settings)]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/user_privacy_settings.json",
                        privacy_dict,
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(zipf, "data/user_privacy_settings.json", [], self.counts)
            except Exception as err:
                core_logger.print_to_log(
                    f"Failed to collect user privacy settings: {err}",
                    "error",
                    exc=err,
                )
                raise DataCollectionError(f"Failed to collect user privacy settings: {err}") from err

        except SQLAlchemyError as err:
            core_logger.print_to_log(f"Database error collecting user settings: {err}", "error", exc=err)
            raise DatabaseConnectionError(f"Failed to collect user settings: {err}") from err

    def add_activity_files_to_zip(self, zipf: zipfile.ZipFile, user_activities: list[Any]):
        """
        Add activity files to ZIP archive.

        Args:
            zipf: ZipFile instance to write to.
            user_activities: List of activity objects.

        Raises:
            FileSystemError: If file system error occurs.
        """
        if not user_activities:
            return

        try:
            if not os.path.exists(core_config.FILES_PROCESSED_DIR):
                core_logger.print_to_log(
                    f"Files directory does not exist: {core_config.FILES_PROCESSED_DIR}",
                    "warning",
                )
                return

            for root, _, files in os.walk(core_config.FILES_PROCESSED_DIR):
                for file in files:
                    try:
                        file_id, _ = os.path.splitext(file)
                        if any(str(activity.id) == file_id for activity in user_activities):
                            file_path = os.path.join(root, file)

                            # Check if file exists and is readable
                            if not os.path.isfile(file_path):
                                core_logger.print_to_log(f"Activity file not found: {file_path}", "warning")
                                continue

                            arcname = os.path.join(
                                "activity_files",
                                os.path.relpath(file_path, core_config.FILES_PROCESSED_DIR),
                            )
                            zipf.write(file_path, arcname)
                            self.counts["activity_files"] += 1

                    except OSError as err:
                        core_logger.print_to_log(
                            f"Failed to add activity file {file}: {err}",
                            "warning",
                            exc=err,
                        )
                        continue
                    except MemoryAllocationError:
                        raise
                    except Exception as err:
                        core_logger.print_to_log(
                            f"Unexpected error adding activity file {file}: {err}",
                            "warning",
                            exc=err,
                        )
                        continue

        except OSError as err:
            core_logger.print_to_log(f"File system error accessing activity files: {err}", "error", exc=err)
            raise FileSystemError(f"Cannot access activity files directory: {err}") from err

    def add_activity_media_to_zip(self, zipf: zipfile.ZipFile, user_activities: list[Any]):
        """
        Add activity media files to ZIP archive.

        Args:
            zipf: ZipFile instance to write to.
            user_activities: List of activity objects.

        Raises:
            FileSystemError: If file system error occurs.
        """
        if not user_activities:
            return

        try:
            if not os.path.exists(core_config.settings.ACTIVITY_MEDIA_DIR):
                core_logger.print_to_log(
                    f"Media directory does not exist: {core_config.settings.ACTIVITY_MEDIA_DIR}",
                    "warning",
                )
                return

            for root, _, files in os.walk(core_config.settings.ACTIVITY_MEDIA_DIR):
                for file in files:
                    try:
                        file_id, _ = os.path.splitext(file)
                        file_activity_id = file_id.split("_")[0]

                        if any(str(activity.id) == file_activity_id for activity in user_activities):
                            file_path = os.path.join(root, file)

                            # Check if file exists and is readable
                            if not os.path.isfile(file_path):
                                core_logger.print_to_log(f"Media file not found: {file_path}", "warning")
                                continue

                            arcname = os.path.join(
                                "activity_media",
                                os.path.relpath(file_path, core_config.settings.ACTIVITY_MEDIA_DIR),
                            )
                            zipf.write(file_path, arcname)
                            self.counts["media"] += 1

                    except OSError as err:
                        core_logger.print_to_log(
                            f"Failed to add media file {file}: {err}",
                            "warning",
                            exc=err,
                        )
                        continue
                    except MemoryAllocationError:
                        raise
                    except Exception as err:
                        core_logger.print_to_log(
                            f"Unexpected error adding media file {file}: {err}",
                            "warning",
                            exc=err,
                        )
                        continue

        except OSError as err:
            core_logger.print_to_log(f"File system error accessing media files: {err}", "error", exc=err)
            raise FileSystemError(f"Cannot access media files directory: {err}") from err

    def add_user_images_to_zip(self, zipf: zipfile.ZipFile):
        """
        Add user image files to ZIP archive.

        Args:
            zipf: ZipFile instance to write to.

        Raises:
            FileSystemError: If file system error occurs.
        """
        try:
            if not os.path.exists(core_config.USER_IMAGES_DIR):
                core_logger.print_to_log(
                    f"User images directory does not exist: {core_config.USER_IMAGES_DIR}",
                    "warning",
                )
                return

            self._add_user_images_optimized(zipf, core_config.USER_IMAGES_DIR)

        except MemoryAllocationError:
            raise
        except OSError as err:
            core_logger.print_to_log(f"Error adding user images to ZIP: {err}", "error", exc=err)
            raise FileSystemError(f"Failed to add user images: {err}") from err

    def _add_user_images_optimized(self, zipf: zipfile.ZipFile, images_dir: str):
        """
        Recursively add user images from directory.

        Args:
            zipf: ZipFile instance to write to.
            images_dir: Directory path containing images.
        """
        try:
            with os.scandir(images_dir) as entries:
                for entry in entries:
                    if entry.is_file(follow_symlinks=False):
                        self._process_user_image_file(zipf, entry, images_dir)
                    elif entry.is_dir(follow_symlinks=False):
                        # Recursively process subdirectories
                        self._add_user_images_optimized(zipf, entry.path)
        except PermissionError as err:
            core_logger.print_to_log(f"Permission denied accessing {images_dir}: {err}", "warning")
        except OSError as err:
            core_logger.print_to_log(f"OS error accessing {images_dir}: {err}", "warning")

    def _process_user_image_file(self, zipf: zipfile.ZipFile, entry, images_dir: str):
        """
        Process and add single user image file to ZIP.

        Args:
            zipf: ZipFile instance to write to.
            entry: Directory entry for the image file.
            images_dir: Base images directory path.
        """
        try:
            file_id, _ = os.path.splitext(entry.name)
            if str(self.user_id) == file_id:
                # Get file size for monitoring
                file_size = entry.stat().st_size

                # Warn about large image files (>10MB)
                if file_size > 10 * 1024 * 1024:
                    core_logger.print_to_log(
                        f"Large image file: {entry.path} ({file_size / (1024 * 1024):.1f}MB)",
                        "warning",
                    )

                # Check memory usage before adding large files
                if file_size > 5 * 1024 * 1024:  # 5MB threshold for images
                    profile_utils.check_memory_usage(
                        f"before image {entry.name}",
                        self.performance_config.max_memory_mb,
                        self.performance_config.enable_memory_monitoring,
                    )

                arcname = os.path.join(
                    "user_images",
                    os.path.relpath(entry.path, images_dir),
                )
                zipf.write(entry.path, arcname)
                self.counts["user_images"] += 1

        except FileNotFoundError:
            core_logger.print_to_log(f"Image file not found: {entry.path}", "warning")
        except PermissionError:
            core_logger.print_to_log(f"Permission denied: {entry.path}", "warning")
        except OSError as err:
            core_logger.print_to_log(f"Error processing image {entry.path}: {err}", "warning")
        except MemoryAllocationError:
            raise
        except Exception as err:
            core_logger.print_to_log(f"Unexpected error with image {entry.path}: {err}", "warning", exc=err)

    def generate_export_archive(self, user_dict: dict[str, Any], timeout_seconds: int | None = 300) -> Generator[bytes]:
        """
        Generate and stream export archive as bytes.

        Args:
            user_dict: User data dictionary to export.
            timeout_seconds: Optional timeout in seconds.

        Yields:
            Chunks of ZIP archive as bytes.

        Raises:
            ExportTimeoutError: If operation times out.
            ZipCreationError: If ZIP creation fails.
            MemoryAllocationError: If memory limit exceeded.
            FileSystemError: If file system error occurs.
        """
        start_time = time.time()

        try:
            with tempfile.NamedTemporaryFile(delete=True) as tmp:
                try:
                    compression_level = self.performance_config.compression_level
                    core_logger.print_to_log(
                        f"Creating ZIP with compression level {compression_level}",
                        "info",
                    )

                    with zipfile.ZipFile(
                        tmp,
                        "w",
                        compression=zipfile.ZIP_DEFLATED,
                        compresslevel=compression_level,
                    ) as zipf:
                        core_logger.print_to_log(f"Starting export for user {self.user_id}", "info")

                        # Collect and write activities data progressively
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Collecting and writing activities data...", "info")
                        user_activities = self.collect_user_activities_data(zipf)

                        # Collect and write gear data progressively
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Collecting and writing gear data...", "info")
                        self.collect_gear_data(zipf)

                        # Collect and write health data progressively
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Collecting and writing health data...", "info")
                        self.collect_health_weight(zipf)

                        # Collect and write notifications progressively
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Collecting and writing notifications data...", "info")
                        self.collect_notifications_data(zipf)

                        # Collect and write settings data progressively
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Collecting and writing settings data...", "info")
                        self.collect_user_settings_data(zipf)

                        # Write user data
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Writing user data...", "info")
                        user_dict_list = [user_dict]
                        profile_utils.write_json_to_zip(zipf, "data/user.json", user_dict_list, self.counts)

                        # Add files to ZIP with timeout checks
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Adding activity files to archive...", "info")
                        self.add_activity_files_to_zip(zipf, user_activities)

                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Adding activity media to archive...", "info")
                        self.add_activity_media_to_zip(zipf, user_activities)

                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Adding user images to archive...", "info")
                        self.add_user_images_to_zip(zipf)

                        # Write counts file
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log("Writing counts file...", "info")
                        profile_utils.write_json_to_zip(zipf, "counts.json", [self.counts], self.counts)

                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        core_logger.print_to_log(
                            f"Export completed successfully. Counts: {self.counts}",
                            "info",
                        )

                except zipfile.BadZipFile as err:
                    core_logger.print_to_log(f"ZIP creation error: {err}", "error", exc=err)
                    raise ZipCreationError(f"Failed to create ZIP archive: {err}") from err
                except zipfile.LargeZipFile as err:
                    core_logger.print_to_log(f"ZIP file too large: {err}", "error", exc=err)
                    raise ZipCreationError(f"Export archive too large: {err}") from err

                # Ensure all data is written to disk before streaming
                # This is critical for proper ZIP file structure and MIME type detection
                tmp.flush()
                os.fsync(tmp.fileno())

                # Get file size for logging
                file_size = tmp.tell()
                core_logger.print_to_log(
                    f"ZIP archive created successfully: {file_size / (1024 * 1024):.2f}MB",
                    "info",
                )

                # Stream the file with error handling
                tmp.seek(0)
                chunk_count = 0
                while True:
                    try:
                        profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        chunk = tmp.read(8192)
                        if not chunk:
                            break
                        chunk_count += 1
                        yield chunk
                    except MemoryError as err:
                        core_logger.print_to_log(f"Memory error during streaming: {err}", "error", exc=err)
                        raise MemoryAllocationError(f"Insufficient memory to stream export: {err}") from err

                core_logger.print_to_log(
                    f"Successfully streamed {chunk_count} chunks for user {self.user_id}",
                    "info",
                )
        except MemoryAllocationError as err:
            raise err
        except OSError as err:
            core_logger.print_to_log(f"File system error during export: {err}", "error", exc=err)
            raise FileSystemError(f"File system error during export: {err}") from err
        except MemoryError as err:
            core_logger.print_to_log(f"Memory allocation error during export: {err}", "error", exc=err)
            raise MemoryAllocationError(f"Insufficient memory for export: {err}") from err

add_activity_files_to_zip

add_activity_files_to_zip(zipf, user_activities)

Add activity files to ZIP archive.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to write to.	required
user_activities	list[Any]	List of activity objects.	required

Raises:

Type	Description
FileSystemError	If file system error occurs.

Source code in backend/app/users/users_profile/export_service.py

def add_activity_files_to_zip(self, zipf: zipfile.ZipFile, user_activities: list[Any]):
    """
    Add activity files to ZIP archive.

    Args:
        zipf: ZipFile instance to write to.
        user_activities: List of activity objects.

    Raises:
        FileSystemError: If file system error occurs.
    """
    if not user_activities:
        return

    try:
        if not os.path.exists(core_config.FILES_PROCESSED_DIR):
            core_logger.print_to_log(
                f"Files directory does not exist: {core_config.FILES_PROCESSED_DIR}",
                "warning",
            )
            return

        for root, _, files in os.walk(core_config.FILES_PROCESSED_DIR):
            for file in files:
                try:
                    file_id, _ = os.path.splitext(file)
                    if any(str(activity.id) == file_id for activity in user_activities):
                        file_path = os.path.join(root, file)

                        # Check if file exists and is readable
                        if not os.path.isfile(file_path):
                            core_logger.print_to_log(f"Activity file not found: {file_path}", "warning")
                            continue

                        arcname = os.path.join(
                            "activity_files",
                            os.path.relpath(file_path, core_config.FILES_PROCESSED_DIR),
                        )
                        zipf.write(file_path, arcname)
                        self.counts["activity_files"] += 1

                except OSError as err:
                    core_logger.print_to_log(
                        f"Failed to add activity file {file}: {err}",
                        "warning",
                        exc=err,
                    )
                    continue
                except MemoryAllocationError:
                    raise
                except Exception as err:
                    core_logger.print_to_log(
                        f"Unexpected error adding activity file {file}: {err}",
                        "warning",
                        exc=err,
                    )
                    continue

    except OSError as err:
        core_logger.print_to_log(f"File system error accessing activity files: {err}", "error", exc=err)
        raise FileSystemError(f"Cannot access activity files directory: {err}") from err

add_activity_media_to_zip

add_activity_media_to_zip(zipf, user_activities)

Add activity media files to ZIP archive.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to write to.	required
user_activities	list[Any]	List of activity objects.	required

Raises:

Type	Description
FileSystemError	If file system error occurs.

Source code in backend/app/users/users_profile/export_service.py

def add_activity_media_to_zip(self, zipf: zipfile.ZipFile, user_activities: list[Any]):
    """
    Add activity media files to ZIP archive.

    Args:
        zipf: ZipFile instance to write to.
        user_activities: List of activity objects.

    Raises:
        FileSystemError: If file system error occurs.
    """
    if not user_activities:
        return

    try:
        if not os.path.exists(core_config.settings.ACTIVITY_MEDIA_DIR):
            core_logger.print_to_log(
                f"Media directory does not exist: {core_config.settings.ACTIVITY_MEDIA_DIR}",
                "warning",
            )
            return

        for root, _, files in os.walk(core_config.settings.ACTIVITY_MEDIA_DIR):
            for file in files:
                try:
                    file_id, _ = os.path.splitext(file)
                    file_activity_id = file_id.split("_")[0]

                    if any(str(activity.id) == file_activity_id for activity in user_activities):
                        file_path = os.path.join(root, file)

                        # Check if file exists and is readable
                        if not os.path.isfile(file_path):
                            core_logger.print_to_log(f"Media file not found: {file_path}", "warning")
                            continue

                        arcname = os.path.join(
                            "activity_media",
                            os.path.relpath(file_path, core_config.settings.ACTIVITY_MEDIA_DIR),
                        )
                        zipf.write(file_path, arcname)
                        self.counts["media"] += 1

                except OSError as err:
                    core_logger.print_to_log(
                        f"Failed to add media file {file}: {err}",
                        "warning",
                        exc=err,
                    )
                    continue
                except MemoryAllocationError:
                    raise
                except Exception as err:
                    core_logger.print_to_log(
                        f"Unexpected error adding media file {file}: {err}",
                        "warning",
                        exc=err,
                    )
                    continue

    except OSError as err:
        core_logger.print_to_log(f"File system error accessing media files: {err}", "error", exc=err)
        raise FileSystemError(f"Cannot access media files directory: {err}") from err

add_user_images_to_zip

add_user_images_to_zip(zipf)

Add user image files to ZIP archive.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to write to.	required

Raises:

Type	Description
FileSystemError	If file system error occurs.

Source code in backend/app/users/users_profile/export_service.py

def add_user_images_to_zip(self, zipf: zipfile.ZipFile):
    """
    Add user image files to ZIP archive.

    Args:
        zipf: ZipFile instance to write to.

    Raises:
        FileSystemError: If file system error occurs.
    """
    try:
        if not os.path.exists(core_config.USER_IMAGES_DIR):
            core_logger.print_to_log(
                f"User images directory does not exist: {core_config.USER_IMAGES_DIR}",
                "warning",
            )
            return

        self._add_user_images_optimized(zipf, core_config.USER_IMAGES_DIR)

    except MemoryAllocationError:
        raise
    except OSError as err:
        core_logger.print_to_log(f"Error adding user images to ZIP: {err}", "error", exc=err)
        raise FileSystemError(f"Failed to add user images: {err}") from err

collect_gear_data

collect_gear_data(zipf)

Collect and write gear data to ZIP.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to write to.	required

Raises:

Type	Description
DatabaseConnectionError	If database error occurs.

Source code in backend/app/users/users_profile/export_service.py

def collect_gear_data(self, zipf: zipfile.ZipFile) -> None:
    """
    Collect and write gear data to ZIP.

    Args:
        zipf: ZipFile instance to write to.

    Raises:
        DatabaseConnectionError: If database error occurs.
    """
    try:
        # Collect and write gears
        try:
            gears = gear_crud.get_gear_user(self.user_id, self.db)
            if gears:
                gears_dicts = [profile_utils.sqlalchemy_obj_to_dict(g) for g in gears]
                profile_utils.write_json_to_zip(zipf, "data/gears.json", gears_dicts, self.counts)
            else:
                profile_utils.write_json_to_zip(zipf, "data/gears.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect gears: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect gears: {err}") from err

        # Collect and write gear components
        try:
            gear_components = gear_components_crud.get_gear_components_user(self.user_id, self.db)
            if gear_components:
                gear_components_dicts = [profile_utils.sqlalchemy_obj_to_dict(gc) for gc in gear_components]
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/gear_components.json",
                    gear_components_dicts,
                    self.counts,
                )
            else:
                profile_utils.write_json_to_zip(zipf, "data/gear_components.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect gear components: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect gear components: {err}") from err

    except SQLAlchemyError as err:
        core_logger.print_to_log(f"Database error collecting gear data: {err}", "error", exc=err)
        raise DatabaseConnectionError(f"Failed to collect gear data: {err}") from err

collect_health_weight

collect_health_weight(zipf)

Collect and write health data to ZIP.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to write to.	required

Raises:

Type	Description
DatabaseConnectionError	If database error occurs.

Source code in backend/app/users/users_profile/export_service.py

def collect_health_weight(self, zipf: zipfile.ZipFile) -> None:
    """
    Collect and write health data to ZIP.

    Args:
        zipf: ZipFile instance to write to.

    Raises:
        DatabaseConnectionError: If database error occurs.
    """
    try:
        # Collect and write health data
        try:
            health_weight = health_weight_crud.get_all_health_weight_by_user_id(self.user_id, self.db)
            if health_weight:
                health_weight_dicts = [profile_utils.sqlalchemy_obj_to_dict(hd) for hd in health_weight]
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/health_weight.json",
                    health_weight_dicts,
                    self.counts,
                )
            else:
                profile_utils.write_json_to_zip(zipf, "data/health_weight.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect health data: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect health data: {err}") from err

        # Collect and write health targets
        try:
            health_targets = health_targets_crud.get_health_targets_by_user_id(self.user_id, self.db)
            if health_targets:
                # health_targets is a single object, not a list
                health_targets_dict = profile_utils.sqlalchemy_obj_to_dict(health_targets)
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/health_targets.json",
                    [health_targets_dict],
                    self.counts,
                )
            else:
                profile_utils.write_json_to_zip(zipf, "data/health_targets.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect health targets: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect health targets: {err}") from err

    except SQLAlchemyError as err:
        core_logger.print_to_log(f"Database error collecting health data: {err}", "error", exc=err)
        raise DatabaseConnectionError(f"Failed to collect health data: {err}") from err

collect_user_activities_data

collect_user_activities_data(zipf)

Collect and write user activities to ZIP.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to write to.	required

Returns:

Type	Description
list[Any]	List of collected activity objects.

Raises:

Type	Description
DatabaseConnectionError	If database error occurs.
MemoryAllocationError	If memory limit exceeded.
DataCollectionError	If collection fails.

Source code in backend/app/users/users_profile/export_service.py

def collect_user_activities_data(self, zipf: zipfile.ZipFile) -> list[Any]:
    """
    Collect and write user activities to ZIP.

    Args:
        zipf: ZipFile instance to write to.

    Returns:
        List of collected activity objects.

    Raises:
        DatabaseConnectionError: If database error occurs.
        MemoryAllocationError: If memory limit exceeded.
        DataCollectionError: If collection fails.
    """
    try:
        profile_utils.check_memory_usage(
            "activity collection start",
            self.performance_config.max_memory_mb,
            self.performance_config.enable_memory_monitoring,
        )

        # Get activities in batches using pagination
        all_activities = []
        offset = 0
        batch_size = self.performance_config.batch_size

        core_logger.print_to_log(
            f"Starting batched activity collection with batch_size={batch_size}",
            "info",
        )

        while True:
            # Get a batch of activities
            try:
                batch_activities = self._get_activities_batch(offset, batch_size)
            except (SQLAlchemyError, MemoryAllocationError):
                # Persistent failures must propagate so the outer handlers can
                # return an explicit error rather than silently skipping pages.
                raise
            except Exception as err:
                core_logger.print_to_log(
                    f"Failed to get activities batch (offset={offset}), skipping: {err}",
                    "warning",
                    exc=err,
                )
                offset += batch_size
                continue

            if not batch_activities:
                break

            all_activities.extend(batch_activities)
            offset += batch_size

            # Check memory usage after each batch
            profile_utils.check_memory_usage(
                f"activity batch {offset // batch_size}",
                self.performance_config.max_memory_mb,
                self.performance_config.enable_memory_monitoring,
            )

            core_logger.print_to_log(
                f"Collected {len(batch_activities)} activities in batch (total: {len(all_activities)})",
                "info",
            )

        if not all_activities:
            core_logger.print_to_log(f"No activities found for user {self.user_id}", "info")
            # Write empty activities file
            profile_utils.write_json_to_zip(zipf, "data/activities.json", [], self.counts)
            return []

        # Write activities to ZIP immediately
        activities_dicts = [profile_utils.sqlalchemy_obj_to_dict(a) for a in all_activities]
        profile_utils.write_json_to_zip(zipf, "data/activities.json", activities_dicts, self.counts)

        core_logger.print_to_log(
            f"Written {len(activities_dicts)} activities to ZIP",
            "info",
        )

        # Filter out activities with None IDs and collect valid IDs
        activity_ids = [activity.id for activity in all_activities if activity.id is not None]

        if not activity_ids:
            core_logger.print_to_log(f"No valid activity IDs found for user {self.user_id}", "warning")
            return all_activities

        # Collect and write activity components progressively
        self._collect_and_write_activity_components(zipf, activity_ids, all_activities)

        # Exercise titles don't depend on activity IDs
        try:
            exercise_titles = activity_exercise_titles_crud.get_activity_exercise_titles(self.db)
            if exercise_titles:
                exercise_titles_dicts = [profile_utils.sqlalchemy_obj_to_dict(e) for e in exercise_titles]
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/activity_exercise_titles.json",
                    exercise_titles_dicts,
                    self.counts,
                )
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect exercise titles: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect exercise titles: {err}") from err

    except SQLAlchemyError as err:
        core_logger.print_to_log(f"Database error collecting activities: {err}", "error", exc=err)
        raise DatabaseConnectionError(f"Failed to collect activity data: {err}") from err
    except MemoryAllocationError as err:
        core_logger.print_to_log(
            f"Memory limit exceeded while collecting activities: {err}. ",
            "error",
            exc=err,
        )
        raise err
    except DataCollectionError:
        raise
    except Exception as err:
        core_logger.print_to_log(f"Unexpected error collecting activities: {err}", "error", exc=err)
        raise DataCollectionError(f"Failed to collect activity data: {err}") from err

    return all_activities

collect_user_settings_data

collect_user_settings_data(zipf)

Collect and write user settings to ZIP.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to write to.	required

Raises:

Type	Description
DatabaseConnectionError	If database error occurs.

Source code in backend/app/users/users_profile/export_service.py

def collect_user_settings_data(self, zipf: zipfile.ZipFile) -> None:
    """
    Collect and write user settings to ZIP.

    Args:
        zipf: ZipFile instance to write to.

    Raises:
        DatabaseConnectionError: If database error occurs.
    """
    try:
        # Collect and write user default gear
        try:
            user_default_gear = user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)
            if user_default_gear:
                default_gear_dict = [profile_utils.sqlalchemy_obj_to_dict(user_default_gear)]
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/user_default_gear.json",
                    default_gear_dict,
                    self.counts,
                )
            else:
                profile_utils.write_json_to_zip(zipf, "data/user_default_gear.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect user default gear: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect user default gear: {err}") from err

        # Collect and write user goals
        try:
            user_goals = user_goals_crud.get_user_goals_by_user_id(self.user_id, self.db)
            if user_goals:
                user_goals_dicts = [profile_utils.sqlalchemy_obj_to_dict(ug) for ug in user_goals]
                profile_utils.write_json_to_zip(zipf, "data/user_goals.json", user_goals_dicts, self.counts)
            else:
                profile_utils.write_json_to_zip(zipf, "data/user_goals.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect user goals: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect user goals: {err}") from err

        # Collect and write user integrations
        try:
            user_integrations = user_integrations_crud.get_user_integrations_by_user_id(self.user_id, self.db)
            if user_integrations:
                integrations_dict = [profile_utils.sqlalchemy_obj_to_dict(user_integrations)]
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/user_integrations.json",
                    integrations_dict,
                    self.counts,
                )
            else:
                profile_utils.write_json_to_zip(zipf, "data/user_integrations.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(f"Failed to collect user integrations: {err}", "error", exc=err)
            raise DataCollectionError(f"Failed to collect user integrations: {err}") from err

        # Collect and write user privacy settings
        try:
            user_privacy_settings = users_privacy_settings_crud.get_user_privacy_settings_by_user_id(
                self.user_id, self.db
            )
            if user_privacy_settings:
                privacy_dict = [profile_utils.sqlalchemy_obj_to_dict(user_privacy_settings)]
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/user_privacy_settings.json",
                    privacy_dict,
                    self.counts,
                )
            else:
                profile_utils.write_json_to_zip(zipf, "data/user_privacy_settings.json", [], self.counts)
        except Exception as err:
            core_logger.print_to_log(
                f"Failed to collect user privacy settings: {err}",
                "error",
                exc=err,
            )
            raise DataCollectionError(f"Failed to collect user privacy settings: {err}") from err

    except SQLAlchemyError as err:
        core_logger.print_to_log(f"Database error collecting user settings: {err}", "error", exc=err)
        raise DatabaseConnectionError(f"Failed to collect user settings: {err}") from err

generate_export_archive

generate_export_archive(user_dict, timeout_seconds=300)

Generate and stream export archive as bytes.

Parameters:

Name	Type	Description	Default
user_dict	dict[str, Any]	User data dictionary to export.	required
timeout_seconds	int | None	Optional timeout in seconds.	300

Yields:

Type	Description
Generator[bytes]	Chunks of ZIP archive as bytes.

Raises:

Type	Description
ExportTimeoutError	If operation times out.
ZipCreationError	If ZIP creation fails.
MemoryAllocationError	If memory limit exceeded.
FileSystemError	If file system error occurs.

Source code in backend/app/users/users_profile/export_service.py

def generate_export_archive(self, user_dict: dict[str, Any], timeout_seconds: int | None = 300) -> Generator[bytes]:
    """
    Generate and stream export archive as bytes.

    Args:
        user_dict: User data dictionary to export.
        timeout_seconds: Optional timeout in seconds.

    Yields:
        Chunks of ZIP archive as bytes.

    Raises:
        ExportTimeoutError: If operation times out.
        ZipCreationError: If ZIP creation fails.
        MemoryAllocationError: If memory limit exceeded.
        FileSystemError: If file system error occurs.
    """
    start_time = time.time()

    try:
        with tempfile.NamedTemporaryFile(delete=True) as tmp:
            try:
                compression_level = self.performance_config.compression_level
                core_logger.print_to_log(
                    f"Creating ZIP with compression level {compression_level}",
                    "info",
                )

                with zipfile.ZipFile(
                    tmp,
                    "w",
                    compression=zipfile.ZIP_DEFLATED,
                    compresslevel=compression_level,
                ) as zipf:
                    core_logger.print_to_log(f"Starting export for user {self.user_id}", "info")

                    # Collect and write activities data progressively
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Collecting and writing activities data...", "info")
                    user_activities = self.collect_user_activities_data(zipf)

                    # Collect and write gear data progressively
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Collecting and writing gear data...", "info")
                    self.collect_gear_data(zipf)

                    # Collect and write health data progressively
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Collecting and writing health data...", "info")
                    self.collect_health_weight(zipf)

                    # Collect and write notifications progressively
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Collecting and writing notifications data...", "info")
                    self.collect_notifications_data(zipf)

                    # Collect and write settings data progressively
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Collecting and writing settings data...", "info")
                    self.collect_user_settings_data(zipf)

                    # Write user data
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Writing user data...", "info")
                    user_dict_list = [user_dict]
                    profile_utils.write_json_to_zip(zipf, "data/user.json", user_dict_list, self.counts)

                    # Add files to ZIP with timeout checks
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Adding activity files to archive...", "info")
                    self.add_activity_files_to_zip(zipf, user_activities)

                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Adding activity media to archive...", "info")
                    self.add_activity_media_to_zip(zipf, user_activities)

                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Adding user images to archive...", "info")
                    self.add_user_images_to_zip(zipf)

                    # Write counts file
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log("Writing counts file...", "info")
                    profile_utils.write_json_to_zip(zipf, "counts.json", [self.counts], self.counts)

                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    core_logger.print_to_log(
                        f"Export completed successfully. Counts: {self.counts}",
                        "info",
                    )

            except zipfile.BadZipFile as err:
                core_logger.print_to_log(f"ZIP creation error: {err}", "error", exc=err)
                raise ZipCreationError(f"Failed to create ZIP archive: {err}") from err
            except zipfile.LargeZipFile as err:
                core_logger.print_to_log(f"ZIP file too large: {err}", "error", exc=err)
                raise ZipCreationError(f"Export archive too large: {err}") from err

            # Ensure all data is written to disk before streaming
            # This is critical for proper ZIP file structure and MIME type detection
            tmp.flush()
            os.fsync(tmp.fileno())

            # Get file size for logging
            file_size = tmp.tell()
            core_logger.print_to_log(
                f"ZIP archive created successfully: {file_size / (1024 * 1024):.2f}MB",
                "info",
            )

            # Stream the file with error handling
            tmp.seek(0)
            chunk_count = 0
            while True:
                try:
                    profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    chunk = tmp.read(8192)
                    if not chunk:
                        break
                    chunk_count += 1
                    yield chunk
                except MemoryError as err:
                    core_logger.print_to_log(f"Memory error during streaming: {err}", "error", exc=err)
                    raise MemoryAllocationError(f"Insufficient memory to stream export: {err}") from err

            core_logger.print_to_log(
                f"Successfully streamed {chunk_count} chunks for user {self.user_id}",
                "info",
            )
    except MemoryAllocationError as err:
        raise err
    except OSError as err:
        core_logger.print_to_log(f"File system error during export: {err}", "error", exc=err)
        raise FileSystemError(f"File system error during export: {err}") from err
    except MemoryError as err:
        core_logger.print_to_log(f"Memory allocation error during export: {err}", "error", exc=err)
        raise MemoryAllocationError(f"Insufficient memory for export: {err}") from err

ExportTimeoutError

Bases: ExportError

Raised when export operation exceeds time limit.

Source code in backend/app/users/users_profile/exceptions.py

class ExportTimeoutError(ExportError):
    """
    Raised when export operation exceeds time limit.
    """

    pass

FileFormatError

Bases: ProfileImportError

Raised when imported file format is invalid or corrupted.

Source code in backend/app/users/users_profile/exceptions.py

class FileFormatError(ProfileImportError):
    """
    Raised when imported file format is invalid or corrupted.
    """

    pass

FileSizeError

Bases: ProfileImportError

Raised when imported file exceeds size limits.

Source code in backend/app/users/users_profile/exceptions.py

class FileSizeError(ProfileImportError):
    """
    Raised when imported file exceeds size limits.
    """

    pass

FileSystemError

Bases: ProfileOperationError

Raised when file system operations fail.

Source code in backend/app/users/users_profile/exceptions.py

class FileSystemError(ProfileOperationError):
    """
    Raised when file system operations fail.
    """

    pass

ImportService

Service for importing user profile data from ZIP archive.

Attributes:

Name	Type	Description
user_id		ID of user to import data for.
db		Database session.
websocket_manager		WebSocket manager for updates.
counts		Dictionary tracking imported item counts.
performance_config	ImportPerformanceConfig	Performance configuration.

Source code in backend/app/users/users_profile/import_service.py

class ImportService:
    """
    Service for importing user profile data from ZIP archive.

    Attributes:
        user_id: ID of user to import data for.
        db: Database session.
        websocket_manager: WebSocket manager for updates.
        counts: Dictionary tracking imported item counts.
        performance_config: Performance configuration.
    """

    def __init__(
        self,
        user_id: int,
        db: Session,
        websocket_manager: websocket_manager.WebSocketManager,
        performance_config: ImportPerformanceConfig | None = None,
    ):
        self.user_id = user_id
        self.db = db
        self.websocket_manager = websocket_manager
        self.counts = profile_utils.initialize_operation_counts(include_user_count=False)
        self.performance_config: ImportPerformanceConfig = (
            performance_config or ImportPerformanceConfig.get_auto_config()
        )

        core_logger.print_to_log(
            f"ImportService initialized with performance config: "
            f"batch_size={self.performance_config.batch_size}, "
            f"max_memory_mb={self.performance_config.max_memory_mb}, "
            f"max_file_size_mb={self.performance_config.max_file_size_mb}, "
            f"timeout_seconds={self.performance_config.timeout_seconds}",
            "info",
        )

    async def import_from_zip_data(self, zip_data: bytes) -> dict[str, Any]:
        """
        Import profile data from ZIP file bytes.

        Args:
            zip_data: ZIP file content as bytes.

        Returns:
            Dictionary with import results and counts.

        Raises:
            FileSizeError: If file exceeds size limit.
            FileFormatError: If ZIP format is invalid.
            FileSystemError: If file system error occurs.
            ImportTimeoutError: If operation times out.
        """
        start_time = time.time()
        timeout_seconds = self.performance_config.timeout_seconds

        # Check file size
        file_size_mb = len(zip_data) / (1024 * 1024)
        if file_size_mb > self.performance_config.max_file_size_mb:
            raise FileSizeError(
                f"ZIP file size ({file_size_mb:.1f}MB) exceeds maximum allowed "
                f"({self.performance_config.max_file_size_mb}MB)"
            )

        # Early memory check BEFORE loading any data
        profile_utils.check_memory_usage(
            "pre-import memory check",
            self.performance_config.max_memory_mb,
            self.performance_config.enable_memory_monitoring,
        )

        try:
            with zipfile.ZipFile(BytesIO(zip_data)) as zipf:
                file_list = set(zipf.namelist())

                # Create ID mappings for relationships
                gears_id_mapping: dict[int, int] = {}
                activities_id_mapping: dict[int, int] = {}

                # Import data in dependency order using streaming approach
                # Load and import gears
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
                gears_data = self._load_single_json(zipf, "data/gears.json")
                gears_id_mapping = await self.collect_and_import_gears_data(gears_data)
                del gears_data  # Explicit memory cleanup

                # Load and import gear components
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
                gear_components_data = self._load_single_json(zipf, "data/gear_components.json")
                await self.collect_and_import_gear_components_data(gear_components_data, gears_id_mapping)
                del gear_components_data

                # Load and import user data (includes user, default gear, integrations, goals, privacy)
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
                user_data = self._load_single_json(zipf, "data/user.json")
                user_default_gear_data = self._load_single_json(zipf, "data/user_default_gear.json")
                user_goals_data = self._load_single_json(zipf, "data/user_goals.json")
                user_integrations_data = self._load_single_json(zipf, "data/user_integrations.json")
                user_privacy_settings_data = self._load_single_json(zipf, "data/user_privacy_settings.json")

                await self.collect_and_import_user_data(
                    user_data,
                    user_default_gear_data,
                    user_goals_data,
                    user_integrations_data,
                    user_privacy_settings_data,
                    gears_id_mapping,
                )
                del user_data, user_default_gear_data, user_integrations_data
                del user_goals_data, user_privacy_settings_data

                # Load and import activities with their components
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")

                # Import activities and components using batched approach to avoid memory issues
                activities_id_mapping = await self.collect_and_import_activities_data_batched(
                    zipf,
                    file_list,
                    gears_id_mapping,
                    start_time,
                    timeout_seconds,
                )

                # Load and import notifications
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
                notifications_data = self._load_single_json(zipf, "data/notifications.json")

                await self.collect_and_import_notifications_data(notifications_data)
                del notifications_data

                # Load and import health data
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
                health_weight_data = self._load_single_json(zipf, "data/health_weight.json")
                health_targets_data = self._load_single_json(zipf, "data/health_targets.json")

                await self.collect_and_import_health_weight(health_weight_data, health_targets_data)
                del health_weight_data, health_targets_data

                # Import files and media
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
                await self.add_activity_files_from_zip(zipf, file_list, activities_id_mapping)
                await self.add_activity_media_from_zip(zipf, file_list, activities_id_mapping)
                await self.add_user_images_from_zip(zipf, file_list)

        except zipfile.BadZipFile as e:
            raise FileFormatError(f"Invalid ZIP file format: {e!s}") from e
        except OSError as e:
            raise FileSystemError(f"File system error during import: {e!s}") from e

        # Guard: if nothing was imported, fail visibly instead of
        # returning a misleading "success" with zero counts.
        if all(v == 0 for v in self.counts.values()):
            raise FileFormatError(
                "Import completed with zero items imported. The ZIP file contains no importable data."
            )

        return {"detail": "Import completed", "imported": self.counts}

    def _load_single_json(self, zipf: zipfile.ZipFile, filename: str, check_memory: bool = True) -> list[Any]:
        """
        Load and parse JSON file from ZIP archive.

        Args:
            zipf: ZipFile instance to read from.
            filename: Name of JSON file to load.
            check_memory: Whether to check memory usage.

        Returns:
            Parsed JSON data as list.

        Raises:
            JSONParseError: If JSON parsing fails.
        """
        try:
            file_list = set(zipf.namelist())
            if filename not in file_list:
                core_logger.print_to_log(
                    f"Expected data file {filename} not found in ZIP archive",
                    "warning",
                )
                return []

            data = json.loads(self._read_zip_entry(zipf, filename))

            if not isinstance(data, list):
                core_logger.print_to_log(
                    f"Expected list in {filename}, got {type(data).__name__}, returning empty list",
                    "warning",
                )
                return []

            core_logger.print_to_log(
                f"Loaded {len(data)} items from {filename}",
                "debug",
            )

            if check_memory:
                profile_utils.check_memory_usage(
                    f"loading {filename}",
                    self.performance_config.max_memory_mb,
                    self.performance_config.enable_memory_monitoring,
                )

            return data
        except json.JSONDecodeError as err:
            error_msg = f"Failed to parse JSON from {filename}: {err}"
            core_logger.print_to_log(error_msg, "error")
            raise JSONParseError(error_msg) from err

    def _read_zip_entry(
        self,
        zipf: zipfile.ZipFile,
        filename: str,
        *,
        max_bytes: int | None = None,
    ) -> bytes:
        """
        Read a ZIP entry after checking its declared size.

        Args:
            zipf: ZipFile instance to read from.
            filename: Name of the entry to read.
            max_bytes: Optional byte cap for this entry. Defaults
                to the import performance file-size cap.

        Returns:
            Entry bytes.

        Raises:
            FileSizeError: If the entry exceeds the allowed size.
        """
        limit = max_bytes if max_bytes is not None else self.performance_config.max_file_size_mb * 1024 * 1024
        info = zipf.getinfo(filename)
        # ``info.file_size`` comes from the central directory and a
        # malicious archive can lie. The declared-size check below
        # is a fast pre-filter; the streaming read that follows
        # enforces the same cap against the actual decompressed
        # bytes so a zip-bomb cannot exhaust memory.
        if info.file_size > limit:
            raise FileSizeError(
                f"ZIP entry {filename} size ({info.file_size} bytes) exceeds maximum allowed ({limit} bytes)"
            )
        chunks: list[bytes] = []
        bytes_read = 0
        chunk_size = 64 * 1024
        with zipf.open(info, "r") as src:
            while True:
                chunk = src.read(chunk_size)
                if not chunk:
                    break
                bytes_read += len(chunk)
                if bytes_read > limit:
                    raise FileSizeError(
                        f"ZIP entry {filename} decompressed size exceeds maximum allowed ({limit} bytes)"
                    )
                chunks.append(chunk)
        return b"".join(chunks)

    async def collect_and_import_gears_data(self, gears_data: list[Any]) -> dict[int, int]:
        """
        Import gear data and create ID mappings.

        Args:
            gears_data: List of gear data dictionaries.

        Returns:
            Dictionary mapping old gear IDs to new IDs.
        """
        gears_id_mapping: dict[int, int] = {}

        if not gears_data:
            core_logger.print_to_log("No gears data to import", "info")
            return gears_id_mapping

        for gear_data in gears_data:
            gear_data["user_id"] = self.user_id
            original_id = gear_data.get("id")
            gear_data.pop("id", None)

            gear = gear_schema.GearCreate(**gear_data)
            new_gear = gear_crud.create_gear(gear, self.user_id, self.db)
            gears_id_mapping[original_id] = new_gear.id
            self.counts["gears"] += 1

        core_logger.print_to_log(f"Imported {self.counts['gears']} gears", "info")
        return gears_id_mapping

    async def collect_and_import_gear_components_data(
        self, gear_components_data: list[Any], gears_id_mapping: dict[int, int]
    ) -> None:
        """
        Import gear components data with ID remapping.

        Args:
            gear_components_data: List of component dicts.
            gears_id_mapping: Mapping of old to new gear IDs.
        """
        if not gear_components_data:
            core_logger.print_to_log("No gear components data to import", "info")
            return

        for gear_component_data in gear_components_data:
            gear_component_data["gear_id"] = (
                gears_id_mapping.get(gear_component_data["gear_id"])
                if gear_component_data.get("gear_id") in gears_id_mapping
                else None
            )

            gear_component_data.pop("id", None)

            gear_component = gear_components_schema.GearComponentCreate(**gear_component_data)
            gear_components_crud.create_gear_component(gear_component, self.user_id, self.db)
            self.counts["gear_components"] += 1

        core_logger.print_to_log(f"Imported {self.counts['gear_components']} gear components", "info")

    async def collect_and_import_user_data(
        self,
        user_data: list[Any],
        user_default_gear_data: list[Any],
        user_goals_data: list[Any],
        user_integrations_data: list[Any],
        user_privacy_settings_data: list[Any],
        gears_id_mapping: dict[int, int],
    ) -> None:
        """
        Import user profile and related settings.

        Args:
            user_data: User profile data.
            user_default_gear_data: Default gear settings.
            user_goals_data: User goals data.
            user_integrations_data: Integration settings.
            user_privacy_settings_data: Privacy settings.
            gears_id_mapping: Mapping of old to new gear IDs.
        """
        if user_data:
            # Import user profile
            user_profile = user_data[0]
            user_profile["id"] = self.user_id

            # Handle photo path
            photo_path = user_profile.get("photo_path")
            if isinstance(photo_path, str) and photo_path.startswith("data/user_images/"):
                extension = photo_path.split(".")[-1]
                user_profile["photo_path"] = f"data/user_images/{self.user_id}.{extension}"

            # Strict allow-list before persistence: profile import is a
            # self-service operation and MUST NOT be a back door for
            # privilege escalation. Even if the source ZIP was crafted
            # to set ``access_type``, ``active``, ``mfa_enabled``,
            # ``mfa_secret``, ``email_verified``, or
            # ``pending_admin_approval``, those fields are dropped here
            # before reaching the database.
            sanitized = {
                key: value for key, value in user_profile.items() if key in users_crud.PROFILE_SELF_SERVICE_FIELDS
            }
            profile_payload = users_schema.ProfileUpdate.model_validate(sanitized)
            await users_crud.edit_profile_user(self.user_id, profile_payload, self.db)
            self.counts["user"] += 1
        else:
            core_logger.print_to_log("No user data to import, but continuing with sub-settings", "info")

        # Import user-related settings — always attempt, even if
        # user_data was empty, so that sub-setting data loaded from
        # the ZIP is not silently discarded.
        await self.collect_and_import_user_default_gear(user_default_gear_data, gears_id_mapping)
        await self.collect_and_import_user_goals(user_goals_data)
        await self.collect_and_import_user_integrations(user_integrations_data)
        await self.collect_and_import_user_privacy_settings(user_privacy_settings_data)

    async def collect_and_import_user_default_gear(
        self, user_default_gear_data: list[Any], gears_id_mapping: dict[int, int]
    ) -> None:
        """
        Import user default gear settings with ID remapping.

        Args:
            user_default_gear_data: Default gear data.
            gears_id_mapping: Mapping of old to new gear IDs.
        """
        if not user_default_gear_data:
            core_logger.print_to_log("No user default gear data to import", "info")
            return

        current_user_default_gear = user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)

        if current_user_default_gear is None:
            core_logger.print_to_log("No existing user default gear, creating new record", "info")
            user_default_gear_crud.create_user_default_gear(self.user_id, self.db)
            current_user_default_gear = user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)

        gear_data = user_default_gear_data[0]
        gear_data["id"] = current_user_default_gear.id
        gear_data["user_id"] = self.user_id

        # Map gear IDs
        gear_fields = [
            "run_gear_id",
            "trail_run_gear_id",
            "virtual_run_gear_id",
            "ride_gear_id",
            "gravel_ride_gear_id",
            "mtb_ride_gear_id",
            "virtual_ride_gear_id",
            "ows_gear_id",
            "walk_gear_id",
            "hike_gear_id",
            "tennis_gear_id",
            "alpine_ski_gear_id",
            "nordic_ski_gear_id",
            "snowboard_gear_id",
            "windsurf_gear_id",
        ]

        for field in gear_fields:
            old_gear_id = gear_data.get(field)
            gear_data[field] = gears_id_mapping.get(old_gear_id)

        user_default_gear = user_default_gear_schema.UsersDefaultGearUpdate(**gear_data)
        user_default_gear_crud.edit_user_default_gear(user_default_gear, self.user_id, self.db)
        core_logger.print_to_log("Imported user default gear", "info")
        self.counts["user_default_gear"] += 1

    async def collect_and_import_user_integrations(self, user_integrations_data: list[Any]) -> None:
        """
        Import user integration settings.

        Args:
            user_integrations_data: Integration data.
        """
        if not user_integrations_data:
            core_logger.print_to_log("No user integrations data to import", "info")
            return

        integrations_data = user_integrations_data[0]
        integrations_data.pop("id", None)
        integrations_data.pop("user_id", None)

        user_integrations = users_integrations_schema.UsersIntegrationsUpdate(**integrations_data)
        user_integrations_crud.edit_user_integrations(user_integrations, self.user_id, self.db)
        core_logger.print_to_log("Imported user integrations", "info")
        self.counts["user_integrations"] += 1

    async def collect_and_import_user_goals(self, user_goals_data: list[Any]) -> None:
        """
        Import user goals data.

        Args:
            user_goals_data: List of user goal dictionaries.
        """
        if not user_goals_data:
            core_logger.print_to_log("No user goals data to import", "info")
            return

        for goal_data in user_goals_data:
            goal_data.pop("id", None)
            goal_data.pop("user_id", None)

            goal = user_goals_schema.UsersGoalCreate(**goal_data)
            user_goals_crud.create_user_goal(self.user_id, goal, self.db)
            self.counts["user_goals"] += 1

        core_logger.print_to_log(f"Imported {self.counts['user_goals']} user goals", "info")

    async def collect_and_import_user_privacy_settings(self, user_privacy_settings_data: list[Any]) -> None:
        """
        Import user privacy settings.

        Args:
            user_privacy_settings_data: Privacy settings data.
        """
        if not user_privacy_settings_data:
            core_logger.print_to_log("No user privacy settings data to import", "info")
            return

        privacy_data = user_privacy_settings_data[0]
        privacy_data.pop("id", None)
        privacy_data.pop("user_id", None)

        user_privacy_settings = users_privacy_settings_schema.UsersPrivacySettingsUpdate(**privacy_data)
        users_privacy_settings_crud.edit_user_privacy_settings(self.user_id, user_privacy_settings, self.db)
        core_logger.print_to_log("Imported user privacy settings", "info")
        self.counts["user_privacy_settings"] += 1

    async def collect_and_import_activity_components(
        self,
        activity_laps_data: list[Any],
        activity_sets_data: list[Any],
        activity_streams_data: list[Any],
        activity_workout_steps_data: list[Any],
        activity_media_data: list[Any],
        activity_exercise_titles_data: list[Any],
        original_activity_id: int,
        new_activity_id: int,
    ) -> None:
        """
        Import all components for a single activity.

        Args:
            activity_laps_data: Laps data for all activities.
            activity_sets_data: Sets data for all activities.
            activity_streams_data: Streams data.
            activity_workout_steps_data: Workout steps data.
            activity_media_data: Media data for all activities.
            activity_exercise_titles_data: Exercise titles.
            original_activity_id: Old activity ID.
            new_activity_id: New activity ID.
        """
        # Import laps - filter for this activity
        if activity_laps_data:
            laps = []
            laps_for_activity = [lap for lap in activity_laps_data if lap.get("activity_id") == original_activity_id]
            for lap_data in laps_for_activity:
                lap_data.pop("id", None)
                lap_data["activity_id"] = new_activity_id
                laps.append(lap_data)

            if laps:
                activity_laps_crud.create_activity_laps(laps, new_activity_id, self.db)
                self.counts["activity_laps"] += len(laps)

        # Import sets - filter for this activity
        if activity_sets_data:
            sets = []
            sets_for_activity = [
                activity_set
                for activity_set in activity_sets_data
                if activity_set.get("activity_id") == original_activity_id
            ]
            for activity_set in sets_for_activity:
                activity_set.pop("id", None)
                activity_set["activity_id"] = new_activity_id
                set_activity = activity_sets_schema.ActivitySetsCreate(**activity_set)
                sets.append(set_activity)

            if sets:
                activity_sets_crud.create_activity_sets(sets, new_activity_id, self.db)
                self.counts["activity_sets"] += len(sets)

        # Import streams - filter for this activity
        if activity_streams_data:
            streams = []
            streams_for_activity = [
                stream for stream in activity_streams_data if stream.get("activity_id") == original_activity_id
            ]
            for stream_data in streams_for_activity:
                stream_data.pop("id", None)
                stream_data["activity_id"] = new_activity_id
                stream = activity_streams_schema.ActivityStreams(**stream_data)
                streams.append(stream)

            if streams:
                activity_streams_crud.create_activity_streams(streams, self.db)
                self.counts["activity_streams"] += len(streams)

        # Import workout steps
        if activity_workout_steps_data:
            steps = []
            steps_for_activity = [
                step for step in activity_workout_steps_data if step.get("activity_id") == original_activity_id
            ]
            for step_data in steps_for_activity:
                step_data.pop("id", None)
                step_data["activity_id"] = new_activity_id
                step = activity_workout_steps_schema.ActivityWorkoutSteps(**step_data)
                steps.append(step)

            if steps:
                activity_workout_steps_crud.create_activity_workout_steps(steps, new_activity_id, self.db)
                self.counts["activity_workout_steps"] += len(steps)

        # Import media
        if activity_media_data:
            media = []
            media_for_activity = [
                media_item
                for media_item in activity_media_data
                if media_item.get("activity_id") == original_activity_id
            ]
            for media_data in media_for_activity:
                media_data.pop("id", None)
                media_data["activity_id"] = new_activity_id

                # Update media path
                old_path = media_data.get("media_path", None)
                if old_path:
                    filename = os.path.basename(str(old_path).replace("\\", "/"))
                    if "_" not in filename:
                        core_logger.print_to_log(
                            f"Skipping activity media with invalid path: {old_path}",
                            "warning",
                        )
                        continue
                    suffix = filename.split("_", 1)[1]
                    new_file_name = f"{new_activity_id}_{suffix}"
                    try:
                        media_data["media_path"] = str(
                            file_uploads.resolve_storage_path(
                                core_config.settings.ACTIVITY_MEDIA_DIR,
                                new_file_name,
                            )
                        )
                    except HTTPException as err:
                        core_logger.print_to_log(
                            f"Skipping activity media with unsafe path {old_path}: {err.detail}",
                            "warning",
                        )
                        continue

                media_item = activity_media_schema.ActivityMedia(**media_data)
                media.append(media_item)

            if media:
                activity_media_crud.create_activity_medias(media, new_activity_id, self.db)
                self.counts["activity_media"] += len(media)

        # Import exercise titles
        if activity_exercise_titles_data:
            titles = []
            exercise_titles_for_activity = [
                title for title in activity_exercise_titles_data if title.get("activity_id") == original_activity_id
            ]
            for title_data in exercise_titles_for_activity:
                title_data.pop("id", None)
                title_data["activity_id"] = new_activity_id
                title = activity_exercise_titles_schema.ActivityExerciseTitles(**title_data)
                titles.append(title)

            if titles:
                activity_exercise_titles_crud.create_activity_exercise_titles(titles, self.db)
                self.counts["activity_exercise_titles"] += len(titles)

    async def collect_and_import_activities_data_batched(
        self,
        zipf: zipfile.ZipFile,
        file_list: set[str],
        gears_id_mapping: dict[int, int],
        start_time: float,
        timeout_seconds: int,
    ) -> dict[int, int]:
        """
        Import activities in batches to manage memory.

        Args:
            zipf: ZipFile instance to read from.
            file_list: Set of file paths in ZIP.
            gears_id_mapping: Mapping of old to new gear IDs.
            start_time: Import operation start time.
            timeout_seconds: Timeout limit in seconds.

        Returns:
            Dictionary mapping old activity IDs to new IDs.

        Raises:
            ActivityLimitError: If too many activities.
            ImportTimeoutError: If operation times out.
        """
        activities_id_mapping: dict[int, int] = {}

        # Load activities list
        activities_data = self._load_single_json(zipf, "data/activities.json")
        if not activities_data:
            core_logger.print_to_log("No activities data to import", "info")
            return activities_id_mapping

        # Check activity count limit
        if len(activities_data) > self.performance_config.max_activities:
            raise ActivityLimitError(
                f"Too many activities ({len(activities_data)}). "
                f"Maximum allowed: {self.performance_config.max_activities}"
            )

        # Load small component files that won't cause memory issues
        activity_workout_steps_data = self._load_single_json(
            zipf, "data/activity_workout_steps.json", check_memory=False
        )
        activity_media_data = self._load_single_json(zipf, "data/activity_media.json", check_memory=False)
        activity_exercise_titles_data = self._load_single_json(
            zipf, "data/activity_exercise_titles.json", check_memory=False
        )

        # Get list of split files for large components
        laps_files = self._get_split_files_list(file_list, "data/activity_laps")
        sets_files = self._get_split_files_list(file_list, "data/activity_sets")
        streams_files = self._get_split_files_list(file_list, "data/activity_streams")

        core_logger.print_to_log(
            f"Importing {len(activities_data)} activities with batched component loading",
            "info",
        )

        # Process activities in batches
        batch_size = self.performance_config.batch_size
        for batch_start in range(0, len(activities_data), batch_size):
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")

            batch_end = min(batch_start + batch_size, len(activities_data))
            activities_batch = activities_data[batch_start:batch_end]

            core_logger.print_to_log(
                f"Processing activities batch {batch_start // batch_size + 1}: activities {batch_start}-{batch_end}",
                "info",
            )

            # Load components for this batch only
            batch_laps = self._load_components_for_batch(zipf, laps_files, activities_batch, "laps")
            batch_sets = self._load_components_for_batch(zipf, sets_files, activities_batch, "sets")
            batch_streams = self._load_components_for_batch(zipf, streams_files, activities_batch, "streams")

            # Import activities in this batch
            for activity_data in activities_batch:
                activity_data["user_id"] = self.user_id
                activity_data["gear_id"] = (
                    gears_id_mapping.get(activity_data["gear_id"])
                    if activity_data.get("gear_id") in gears_id_mapping
                    else None
                )

                original_activity_id = activity_data.get("id")
                activity_data.pop("id", None)

                activity = activity_schema.Activity(**activity_data)
                new_activity = await activities_crud.create_activity(activity, self.websocket_manager, self.db, False)

                if original_activity_id is not None and new_activity.id is not None:
                    activities_id_mapping[original_activity_id] = new_activity.id

                    # Import activity components using batch-loaded data
                    await self.collect_and_import_activity_components(
                        batch_laps,
                        batch_sets,
                        batch_streams,
                        activity_workout_steps_data,
                        activity_media_data,
                        activity_exercise_titles_data,
                        original_activity_id,
                        new_activity.id,
                    )

                self.counts["activities"] += 1

            # Clear batch data from memory
            del batch_laps, batch_sets, batch_streams
            profile_utils.check_memory_usage(
                f"activities batch {batch_start // batch_size + 1}",
                self.performance_config.max_memory_mb,
                self.performance_config.enable_memory_monitoring,
            )

        core_logger.print_to_log(f"Imported {self.counts['activities']} activities", "info")
        return activities_id_mapping

    def _get_split_files_list(self, file_list: set[str], base_filename: str) -> list[str]:
        """
        Get list of split component files from ZIP.

        Args:
            file_list: Set of all file paths in ZIP.
            base_filename: Base filename without extension.

        Returns:
            Sorted list of matching file paths.
        """
        split_files = sorted([f for f in file_list if f.startswith(f"{base_filename}_") and f.endswith(".json")])
        if split_files:
            return split_files
        # Fall back to single file if no split files found
        single_file = f"{base_filename}.json"
        if single_file in file_list:
            return [single_file]
        return []

    def _load_components_for_batch(
        self,
        zipf: zipfile.ZipFile,
        component_files: list[str],
        activities_batch: list[Any],
        component_name: str,
    ) -> list[Any]:
        """
        Load components only for activities in current batch.

        Args:
            zipf: ZipFile instance to read from.
            component_files: List of component file paths.
            activities_batch: Activities in current batch.
            component_name: Name of component type.

        Returns:
            List of component data for batch activities.
        """
        if not component_files:
            return []

        # Get activity IDs in this batch
        batch_activity_ids = set(activity.get("id") for activity in activities_batch if activity.get("id") is not None)

        all_components = []

        # Load and filter components from each file
        for filename in component_files:
            try:
                components = json.loads(self._read_zip_entry(zipf, filename))
                # Only keep components for activities in this batch
                filtered = [comp for comp in components if comp.get("activity_id") in batch_activity_ids]
                all_components.extend(filtered)

                if filtered:
                    core_logger.print_to_log(
                        f"Loaded {len(filtered)}/{len(components)} {component_name} from {filename} for batch",
                        "debug",
                    )
            except json.JSONDecodeError as err:
                core_logger.print_to_log(f"Failed to parse {filename}: {err}", "warning")
            except (FileFormatError, FileSizeError, OSError) as err:
                core_logger.print_to_log(f"Error loading {filename}: {err}", "warning")

        return all_components

    async def collect_and_import_notifications_data(self, notifications_data: list[Any]) -> None:
        if not notifications_data:
            core_logger.print_to_log("No notifications data to import", "info")
            return

        for notification_data in notifications_data:
            notification_data["user_id"] = self.user_id
            notification_data.pop("id", None)

            notification = notifications_schema.NotificationCreate(**notification_data)
            notifications_crud.create_notification(notification, self.db)
            self.counts["notifications"] += 1

        core_logger.print_to_log(f"Imported {self.counts['notifications']} notifications", "info")

    async def collect_and_import_health_weight(
        self, health_weight_data: list[Any], health_targets_data: list[Any]
    ) -> None:
        """
        Import health data and targets.

        Args:
            health_weight_data: List of health data records.
            health_targets_data: List of health target records.
        """
        # Import health data
        if health_weight_data:
            for health_weight in health_weight_data:
                health_weight.pop("id", None)
                health_weight.pop("user_id", None)

                # Convert string numeric values to floats
                numeric_fields = [
                    "weight",
                    "bmi",
                    "body_fat",
                    "body_water",
                    "bone_mass",
                    "muscle_mass",
                    "visceral_fat",
                ]
                for field in numeric_fields:
                    if field in health_weight and isinstance(health_weight[field], str):
                        try:
                            health_weight[field] = float(health_weight[field])
                        except (ValueError, TypeError):
                            health_weight[field] = None

                # Convert integer fields
                int_fields = ["physique_rating", "metabolic_age"]
                for field in int_fields:
                    if field in health_weight and isinstance(health_weight[field], str):
                        try:
                            health_weight[field] = int(health_weight[field])
                        except (ValueError, TypeError):
                            health_weight[field] = None

                data = health_weight_schema.HealthWeightCreate(**health_weight)
                health_weight_crud.create_health_weight(self.user_id, data, self.db)
                self.counts["health_weight"] += 1
            core_logger.print_to_log(f"Imported {self.counts['health_weight']} health weight records", "info")
        else:
            core_logger.print_to_log("No health weight data to import", "debug")

        # Import health targets
        if health_targets_data:
            for target_data in health_targets_data:
                current_health_target = health_targets_crud.get_health_targets_by_user_id(self.user_id, self.db)

                # Convert string numeric values to floats/ints
                if isinstance(target_data.get("weight"), str):
                    try:
                        target_data["weight"] = float(target_data["weight"])
                    except (ValueError, TypeError):
                        target_data["weight"] = None

                int_fields = ["steps", "sleep"]
                for field in int_fields:
                    if isinstance(target_data.get(field), str):
                        try:
                            target_data[field] = int(target_data[field])
                        except (ValueError, TypeError):
                            target_data[field] = None

                target_data["user_id"] = self.user_id
                if current_health_target is not None:
                    target_data["id"] = current_health_target.id
                else:
                    target_data.pop("id", None)

                target = health_targets_schema.HealthTargetsUpdate(**target_data)
                health_targets_crud.edit_health_target(target, self.user_id, self.db)
                self.counts["health_targets"] += 1
            core_logger.print_to_log(f"Imported {self.counts['health_targets']} health targets", "info")
        else:
            core_logger.print_to_log("No health targets to import", "debug")

    async def add_activity_files_from_zip(
        self,
        zipf: zipfile.ZipFile,
        file_list: set,
        activities_id_mapping: dict[int, int],
    ) -> None:
        """
        Extract and import activity files from ZIP.

        Args:
            zipf: ZipFile instance to read from.
            file_list: Set of file paths in ZIP.
            activities_id_mapping: Mapping of old to new IDs.
        """
        profile_utils.check_memory_usage(
            "activity files import",
            self.performance_config.max_memory_mb,
            self.performance_config.enable_memory_monitoring,
        )

        for file_path in file_list:
            path = file_path.replace("\\", "/")

            # Import activity files
            if path.lower().endswith((".gpx", ".fit", ".tcx")) and path.startswith("activity_files/"):
                file_id_str = os.path.splitext(os.path.basename(path))[0]
                ext = os.path.splitext(path)[1]
                try:
                    file_id_int = int(file_id_str)
                    new_id = activities_id_mapping.get(file_id_int)

                    if new_id is None:
                        continue

                    new_file_name = f"{new_id}{ext}"

                    # Read bytes from ZIP and save through the
                    # unified validated-write pipeline so the entry
                    # is re-checked as an activity file.
                    activity_limit = file_uploads.file_validator.config.limits.max_activity_file_size
                    file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=activity_limit)
                    try:
                        await file_uploads.save_validated_bytes(
                            file_bytes,
                            kind=file_uploads.UploadKind.ACTIVITY,
                            upload_dir=core_config.FILES_PROCESSED_DIR,
                            filename=new_file_name,
                        )
                    except HTTPException as err:
                        core_logger.print_to_log(
                            f"Profile import dropped invalid activity file {new_file_name}: {err.detail}",
                            "warning",
                        )
                        continue
                    self.counts["activity_files"] += 1
                except ValueError:
                    # Skip files that don't have numeric activity IDs
                    continue

    async def add_activity_media_from_zip(
        self,
        zipf: zipfile.ZipFile,
        file_list: set,
        activities_id_mapping: dict[int, int],
    ) -> None:
        """
        Extract and import activity media from ZIP.

        Args:
            zipf: ZipFile instance to read from.
            file_list: Set of file paths in ZIP.
            activities_id_mapping: Mapping of old to new IDs.
        """
        profile_utils.check_memory_usage(
            "activity media import",
            self.performance_config.max_memory_mb,
            self.performance_config.enable_memory_monitoring,
        )

        for file_path in file_list:
            path = file_path.replace("\\", "/")

            # Import activity media
            if path.lower().endswith((".png", ".jpg", ".jpeg")) and path.startswith("activity_media/"):
                file_name = os.path.basename(path)
                base_name, ext = os.path.splitext(file_name)

                if "_" in base_name:
                    orig_id_str, suffix = base_name.split("_", 1)
                    try:
                        orig_id_int = int(orig_id_str)
                        new_id = activities_id_mapping.get(orig_id_int)

                        if new_id is None:
                            continue

                        new_file_name = f"{new_id}_{suffix}{ext}"

                        # Read bytes from ZIP and save through the
                        # unified validated-write pipeline so the
                        # entry is re-checked as an image.
                        image_limit = file_uploads.file_validator.config.limits.max_image_size
                        file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
                        try:
                            await file_uploads.save_validated_bytes(
                                file_bytes,
                                kind=file_uploads.UploadKind.IMAGE,
                                upload_dir=(core_config.settings.ACTIVITY_MEDIA_DIR),
                                filename=new_file_name,
                            )
                        except HTTPException as err:
                            core_logger.print_to_log(
                                f"Profile import dropped invalid activity media {new_file_name}: {err.detail}",
                                "warning",
                            )
                            continue
                        self.counts["media"] += 1
                    except ValueError:
                        # Skip files that don't have numeric activity IDs
                        continue

    async def add_user_images_from_zip(
        self,
        zipf: zipfile.ZipFile,
        file_list: set,
    ) -> None:
        """
        Extract and import user images from ZIP.

        Args:
            zipf: ZipFile instance to read from.
            file_list: Set of file paths in ZIP.
        """
        profile_utils.check_memory_usage(
            "user images import",
            self.performance_config.max_memory_mb,
            self.performance_config.enable_memory_monitoring,
        )

        for file_path in file_list:
            path = file_path.replace("\\", "/")

            # Import user images
            if path.lower().endswith((".png", ".jpg", ".jpeg")) and path.startswith("user_images/"):
                core_logger.print_to_log(f"Processing user image file: {file_path}", "debug")
                ext = os.path.splitext(path)[1]
                new_file_name = f"{self.user_id}{ext}"

                # Read bytes from ZIP and save through the unified
                # validated-write pipeline so the entry is re-checked
                # as an image.
                image_limit = file_uploads.file_validator.config.limits.max_image_size
                file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
                try:
                    await file_uploads.save_validated_bytes(
                        file_bytes,
                        kind=file_uploads.UploadKind.IMAGE,
                        upload_dir=core_config.USER_IMAGES_DIR,
                        filename=new_file_name,
                    )
                except HTTPException as err:
                    core_logger.print_to_log(
                        f"Profile import dropped invalid user image {new_file_name}: {err.detail}",
                        "warning",
                    )
                    continue
                self.counts["user_images"] += 1

add_activity_files_from_zip async

add_activity_files_from_zip(zipf, file_list, activities_id_mapping)

Extract and import activity files from ZIP.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to read from.	required
file_list	set	Set of file paths in ZIP.	required
activities_id_mapping	dict[int, int]	Mapping of old to new IDs.	required

Source code in backend/app/users/users_profile/import_service.py

async def add_activity_files_from_zip(
    self,
    zipf: zipfile.ZipFile,
    file_list: set,
    activities_id_mapping: dict[int, int],
) -> None:
    """
    Extract and import activity files from ZIP.

    Args:
        zipf: ZipFile instance to read from.
        file_list: Set of file paths in ZIP.
        activities_id_mapping: Mapping of old to new IDs.
    """
    profile_utils.check_memory_usage(
        "activity files import",
        self.performance_config.max_memory_mb,
        self.performance_config.enable_memory_monitoring,
    )

    for file_path in file_list:
        path = file_path.replace("\\", "/")

        # Import activity files
        if path.lower().endswith((".gpx", ".fit", ".tcx")) and path.startswith("activity_files/"):
            file_id_str = os.path.splitext(os.path.basename(path))[0]
            ext = os.path.splitext(path)[1]
            try:
                file_id_int = int(file_id_str)
                new_id = activities_id_mapping.get(file_id_int)

                if new_id is None:
                    continue

                new_file_name = f"{new_id}{ext}"

                # Read bytes from ZIP and save through the
                # unified validated-write pipeline so the entry
                # is re-checked as an activity file.
                activity_limit = file_uploads.file_validator.config.limits.max_activity_file_size
                file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=activity_limit)
                try:
                    await file_uploads.save_validated_bytes(
                        file_bytes,
                        kind=file_uploads.UploadKind.ACTIVITY,
                        upload_dir=core_config.FILES_PROCESSED_DIR,
                        filename=new_file_name,
                    )
                except HTTPException as err:
                    core_logger.print_to_log(
                        f"Profile import dropped invalid activity file {new_file_name}: {err.detail}",
                        "warning",
                    )
                    continue
                self.counts["activity_files"] += 1
            except ValueError:
                # Skip files that don't have numeric activity IDs
                continue

add_activity_media_from_zip async

add_activity_media_from_zip(zipf, file_list, activities_id_mapping)

Extract and import activity media from ZIP.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to read from.	required
file_list	set	Set of file paths in ZIP.	required
activities_id_mapping	dict[int, int]	Mapping of old to new IDs.	required

Source code in backend/app/users/users_profile/import_service.py

async def add_activity_media_from_zip(
    self,
    zipf: zipfile.ZipFile,
    file_list: set,
    activities_id_mapping: dict[int, int],
) -> None:
    """
    Extract and import activity media from ZIP.

    Args:
        zipf: ZipFile instance to read from.
        file_list: Set of file paths in ZIP.
        activities_id_mapping: Mapping of old to new IDs.
    """
    profile_utils.check_memory_usage(
        "activity media import",
        self.performance_config.max_memory_mb,
        self.performance_config.enable_memory_monitoring,
    )

    for file_path in file_list:
        path = file_path.replace("\\", "/")

        # Import activity media
        if path.lower().endswith((".png", ".jpg", ".jpeg")) and path.startswith("activity_media/"):
            file_name = os.path.basename(path)
            base_name, ext = os.path.splitext(file_name)

            if "_" in base_name:
                orig_id_str, suffix = base_name.split("_", 1)
                try:
                    orig_id_int = int(orig_id_str)
                    new_id = activities_id_mapping.get(orig_id_int)

                    if new_id is None:
                        continue

                    new_file_name = f"{new_id}_{suffix}{ext}"

                    # Read bytes from ZIP and save through the
                    # unified validated-write pipeline so the
                    # entry is re-checked as an image.
                    image_limit = file_uploads.file_validator.config.limits.max_image_size
                    file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
                    try:
                        await file_uploads.save_validated_bytes(
                            file_bytes,
                            kind=file_uploads.UploadKind.IMAGE,
                            upload_dir=(core_config.settings.ACTIVITY_MEDIA_DIR),
                            filename=new_file_name,
                        )
                    except HTTPException as err:
                        core_logger.print_to_log(
                            f"Profile import dropped invalid activity media {new_file_name}: {err.detail}",
                            "warning",
                        )
                        continue
                    self.counts["media"] += 1
                except ValueError:
                    # Skip files that don't have numeric activity IDs
                    continue

add_user_images_from_zip async

add_user_images_from_zip(zipf, file_list)

Extract and import user images from ZIP.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to read from.	required
file_list	set	Set of file paths in ZIP.	required

Source code in backend/app/users/users_profile/import_service.py

async def add_user_images_from_zip(
    self,
    zipf: zipfile.ZipFile,
    file_list: set,
) -> None:
    """
    Extract and import user images from ZIP.

    Args:
        zipf: ZipFile instance to read from.
        file_list: Set of file paths in ZIP.
    """
    profile_utils.check_memory_usage(
        "user images import",
        self.performance_config.max_memory_mb,
        self.performance_config.enable_memory_monitoring,
    )

    for file_path in file_list:
        path = file_path.replace("\\", "/")

        # Import user images
        if path.lower().endswith((".png", ".jpg", ".jpeg")) and path.startswith("user_images/"):
            core_logger.print_to_log(f"Processing user image file: {file_path}", "debug")
            ext = os.path.splitext(path)[1]
            new_file_name = f"{self.user_id}{ext}"

            # Read bytes from ZIP and save through the unified
            # validated-write pipeline so the entry is re-checked
            # as an image.
            image_limit = file_uploads.file_validator.config.limits.max_image_size
            file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
            try:
                await file_uploads.save_validated_bytes(
                    file_bytes,
                    kind=file_uploads.UploadKind.IMAGE,
                    upload_dir=core_config.USER_IMAGES_DIR,
                    filename=new_file_name,
                )
            except HTTPException as err:
                core_logger.print_to_log(
                    f"Profile import dropped invalid user image {new_file_name}: {err.detail}",
                    "warning",
                )
                continue
            self.counts["user_images"] += 1

collect_and_import_activities_data_batched async

collect_and_import_activities_data_batched(zipf, file_list, gears_id_mapping, start_time, timeout_seconds)

Import activities in batches to manage memory.

Parameters:

Name	Type	Description	Default
zipf	ZipFile	ZipFile instance to read from.	required
file_list	set[str]	Set of file paths in ZIP.	required
gears_id_mapping	dict[int, int]	Mapping of old to new gear IDs.	required
start_time	float	Import operation start time.	required
timeout_seconds	int	Timeout limit in seconds.	required

Returns:

Type	Description
dict[int, int]	Dictionary mapping old activity IDs to new IDs.

Raises:

Type	Description
ActivityLimitError	If too many activities.
ImportTimeoutError	If operation times out.

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_activities_data_batched(
    self,
    zipf: zipfile.ZipFile,
    file_list: set[str],
    gears_id_mapping: dict[int, int],
    start_time: float,
    timeout_seconds: int,
) -> dict[int, int]:
    """
    Import activities in batches to manage memory.

    Args:
        zipf: ZipFile instance to read from.
        file_list: Set of file paths in ZIP.
        gears_id_mapping: Mapping of old to new gear IDs.
        start_time: Import operation start time.
        timeout_seconds: Timeout limit in seconds.

    Returns:
        Dictionary mapping old activity IDs to new IDs.

    Raises:
        ActivityLimitError: If too many activities.
        ImportTimeoutError: If operation times out.
    """
    activities_id_mapping: dict[int, int] = {}

    # Load activities list
    activities_data = self._load_single_json(zipf, "data/activities.json")
    if not activities_data:
        core_logger.print_to_log("No activities data to import", "info")
        return activities_id_mapping

    # Check activity count limit
    if len(activities_data) > self.performance_config.max_activities:
        raise ActivityLimitError(
            f"Too many activities ({len(activities_data)}). "
            f"Maximum allowed: {self.performance_config.max_activities}"
        )

    # Load small component files that won't cause memory issues
    activity_workout_steps_data = self._load_single_json(
        zipf, "data/activity_workout_steps.json", check_memory=False
    )
    activity_media_data = self._load_single_json(zipf, "data/activity_media.json", check_memory=False)
    activity_exercise_titles_data = self._load_single_json(
        zipf, "data/activity_exercise_titles.json", check_memory=False
    )

    # Get list of split files for large components
    laps_files = self._get_split_files_list(file_list, "data/activity_laps")
    sets_files = self._get_split_files_list(file_list, "data/activity_sets")
    streams_files = self._get_split_files_list(file_list, "data/activity_streams")

    core_logger.print_to_log(
        f"Importing {len(activities_data)} activities with batched component loading",
        "info",
    )

    # Process activities in batches
    batch_size = self.performance_config.batch_size
    for batch_start in range(0, len(activities_data), batch_size):
        profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")

        batch_end = min(batch_start + batch_size, len(activities_data))
        activities_batch = activities_data[batch_start:batch_end]

        core_logger.print_to_log(
            f"Processing activities batch {batch_start // batch_size + 1}: activities {batch_start}-{batch_end}",
            "info",
        )

        # Load components for this batch only
        batch_laps = self._load_components_for_batch(zipf, laps_files, activities_batch, "laps")
        batch_sets = self._load_components_for_batch(zipf, sets_files, activities_batch, "sets")
        batch_streams = self._load_components_for_batch(zipf, streams_files, activities_batch, "streams")

        # Import activities in this batch
        for activity_data in activities_batch:
            activity_data["user_id"] = self.user_id
            activity_data["gear_id"] = (
                gears_id_mapping.get(activity_data["gear_id"])
                if activity_data.get("gear_id") in gears_id_mapping
                else None
            )

            original_activity_id = activity_data.get("id")
            activity_data.pop("id", None)

            activity = activity_schema.Activity(**activity_data)
            new_activity = await activities_crud.create_activity(activity, self.websocket_manager, self.db, False)

            if original_activity_id is not None and new_activity.id is not None:
                activities_id_mapping[original_activity_id] = new_activity.id

                # Import activity components using batch-loaded data
                await self.collect_and_import_activity_components(
                    batch_laps,
                    batch_sets,
                    batch_streams,
                    activity_workout_steps_data,
                    activity_media_data,
                    activity_exercise_titles_data,
                    original_activity_id,
                    new_activity.id,
                )

            self.counts["activities"] += 1

        # Clear batch data from memory
        del batch_laps, batch_sets, batch_streams
        profile_utils.check_memory_usage(
            f"activities batch {batch_start // batch_size + 1}",
            self.performance_config.max_memory_mb,
            self.performance_config.enable_memory_monitoring,
        )

    core_logger.print_to_log(f"Imported {self.counts['activities']} activities", "info")
    return activities_id_mapping

collect_and_import_activity_components async

collect_and_import_activity_components(activity_laps_data, activity_sets_data, activity_streams_data, activity_workout_steps_data, activity_media_data, activity_exercise_titles_data, original_activity_id, new_activity_id)

Import all components for a single activity.

Parameters:

Name	Type	Description	Default
activity_laps_data	list[Any]	Laps data for all activities.	required
activity_sets_data	list[Any]	Sets data for all activities.	required
activity_streams_data	list[Any]	Streams data.	required
activity_workout_steps_data	list[Any]	Workout steps data.	required
activity_media_data	list[Any]	Media data for all activities.	required
activity_exercise_titles_data	list[Any]	Exercise titles.	required
original_activity_id	int	Old activity ID.	required
new_activity_id	int	New activity ID.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_activity_components(
    self,
    activity_laps_data: list[Any],
    activity_sets_data: list[Any],
    activity_streams_data: list[Any],
    activity_workout_steps_data: list[Any],
    activity_media_data: list[Any],
    activity_exercise_titles_data: list[Any],
    original_activity_id: int,
    new_activity_id: int,
) -> None:
    """
    Import all components for a single activity.

    Args:
        activity_laps_data: Laps data for all activities.
        activity_sets_data: Sets data for all activities.
        activity_streams_data: Streams data.
        activity_workout_steps_data: Workout steps data.
        activity_media_data: Media data for all activities.
        activity_exercise_titles_data: Exercise titles.
        original_activity_id: Old activity ID.
        new_activity_id: New activity ID.
    """
    # Import laps - filter for this activity
    if activity_laps_data:
        laps = []
        laps_for_activity = [lap for lap in activity_laps_data if lap.get("activity_id") == original_activity_id]
        for lap_data in laps_for_activity:
            lap_data.pop("id", None)
            lap_data["activity_id"] = new_activity_id
            laps.append(lap_data)

        if laps:
            activity_laps_crud.create_activity_laps(laps, new_activity_id, self.db)
            self.counts["activity_laps"] += len(laps)

    # Import sets - filter for this activity
    if activity_sets_data:
        sets = []
        sets_for_activity = [
            activity_set
            for activity_set in activity_sets_data
            if activity_set.get("activity_id") == original_activity_id
        ]
        for activity_set in sets_for_activity:
            activity_set.pop("id", None)
            activity_set["activity_id"] = new_activity_id
            set_activity = activity_sets_schema.ActivitySetsCreate(**activity_set)
            sets.append(set_activity)

        if sets:
            activity_sets_crud.create_activity_sets(sets, new_activity_id, self.db)
            self.counts["activity_sets"] += len(sets)

    # Import streams - filter for this activity
    if activity_streams_data:
        streams = []
        streams_for_activity = [
            stream for stream in activity_streams_data if stream.get("activity_id") == original_activity_id
        ]
        for stream_data in streams_for_activity:
            stream_data.pop("id", None)
            stream_data["activity_id"] = new_activity_id
            stream = activity_streams_schema.ActivityStreams(**stream_data)
            streams.append(stream)

        if streams:
            activity_streams_crud.create_activity_streams(streams, self.db)
            self.counts["activity_streams"] += len(streams)

    # Import workout steps
    if activity_workout_steps_data:
        steps = []
        steps_for_activity = [
            step for step in activity_workout_steps_data if step.get("activity_id") == original_activity_id
        ]
        for step_data in steps_for_activity:
            step_data.pop("id", None)
            step_data["activity_id"] = new_activity_id
            step = activity_workout_steps_schema.ActivityWorkoutSteps(**step_data)
            steps.append(step)

        if steps:
            activity_workout_steps_crud.create_activity_workout_steps(steps, new_activity_id, self.db)
            self.counts["activity_workout_steps"] += len(steps)

    # Import media
    if activity_media_data:
        media = []
        media_for_activity = [
            media_item
            for media_item in activity_media_data
            if media_item.get("activity_id") == original_activity_id
        ]
        for media_data in media_for_activity:
            media_data.pop("id", None)
            media_data["activity_id"] = new_activity_id

            # Update media path
            old_path = media_data.get("media_path", None)
            if old_path:
                filename = os.path.basename(str(old_path).replace("\\", "/"))
                if "_" not in filename:
                    core_logger.print_to_log(
                        f"Skipping activity media with invalid path: {old_path}",
                        "warning",
                    )
                    continue
                suffix = filename.split("_", 1)[1]
                new_file_name = f"{new_activity_id}_{suffix}"
                try:
                    media_data["media_path"] = str(
                        file_uploads.resolve_storage_path(
                            core_config.settings.ACTIVITY_MEDIA_DIR,
                            new_file_name,
                        )
                    )
                except HTTPException as err:
                    core_logger.print_to_log(
                        f"Skipping activity media with unsafe path {old_path}: {err.detail}",
                        "warning",
                    )
                    continue

            media_item = activity_media_schema.ActivityMedia(**media_data)
            media.append(media_item)

        if media:
            activity_media_crud.create_activity_medias(media, new_activity_id, self.db)
            self.counts["activity_media"] += len(media)

    # Import exercise titles
    if activity_exercise_titles_data:
        titles = []
        exercise_titles_for_activity = [
            title for title in activity_exercise_titles_data if title.get("activity_id") == original_activity_id
        ]
        for title_data in exercise_titles_for_activity:
            title_data.pop("id", None)
            title_data["activity_id"] = new_activity_id
            title = activity_exercise_titles_schema.ActivityExerciseTitles(**title_data)
            titles.append(title)

        if titles:
            activity_exercise_titles_crud.create_activity_exercise_titles(titles, self.db)
            self.counts["activity_exercise_titles"] += len(titles)

collect_and_import_gear_components_data async

collect_and_import_gear_components_data(gear_components_data, gears_id_mapping)

Import gear components data with ID remapping.

Parameters:

Name	Type	Description	Default
gear_components_data	list[Any]	List of component dicts.	required
gears_id_mapping	dict[int, int]	Mapping of old to new gear IDs.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_gear_components_data(
    self, gear_components_data: list[Any], gears_id_mapping: dict[int, int]
) -> None:
    """
    Import gear components data with ID remapping.

    Args:
        gear_components_data: List of component dicts.
        gears_id_mapping: Mapping of old to new gear IDs.
    """
    if not gear_components_data:
        core_logger.print_to_log("No gear components data to import", "info")
        return

    for gear_component_data in gear_components_data:
        gear_component_data["gear_id"] = (
            gears_id_mapping.get(gear_component_data["gear_id"])
            if gear_component_data.get("gear_id") in gears_id_mapping
            else None
        )

        gear_component_data.pop("id", None)

        gear_component = gear_components_schema.GearComponentCreate(**gear_component_data)
        gear_components_crud.create_gear_component(gear_component, self.user_id, self.db)
        self.counts["gear_components"] += 1

    core_logger.print_to_log(f"Imported {self.counts['gear_components']} gear components", "info")

collect_and_import_gears_data async

collect_and_import_gears_data(gears_data)

Import gear data and create ID mappings.

Parameters:

Name	Type	Description	Default
gears_data	list[Any]	List of gear data dictionaries.	required

Returns:

Type	Description
dict[int, int]	Dictionary mapping old gear IDs to new IDs.

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_gears_data(self, gears_data: list[Any]) -> dict[int, int]:
    """
    Import gear data and create ID mappings.

    Args:
        gears_data: List of gear data dictionaries.

    Returns:
        Dictionary mapping old gear IDs to new IDs.
    """
    gears_id_mapping: dict[int, int] = {}

    if not gears_data:
        core_logger.print_to_log("No gears data to import", "info")
        return gears_id_mapping

    for gear_data in gears_data:
        gear_data["user_id"] = self.user_id
        original_id = gear_data.get("id")
        gear_data.pop("id", None)

        gear = gear_schema.GearCreate(**gear_data)
        new_gear = gear_crud.create_gear(gear, self.user_id, self.db)
        gears_id_mapping[original_id] = new_gear.id
        self.counts["gears"] += 1

    core_logger.print_to_log(f"Imported {self.counts['gears']} gears", "info")
    return gears_id_mapping

collect_and_import_health_weight async

collect_and_import_health_weight(health_weight_data, health_targets_data)

Import health data and targets.

Parameters:

Name	Type	Description	Default
health_weight_data	list[Any]	List of health data records.	required
health_targets_data	list[Any]	List of health target records.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_health_weight(
    self, health_weight_data: list[Any], health_targets_data: list[Any]
) -> None:
    """
    Import health data and targets.

    Args:
        health_weight_data: List of health data records.
        health_targets_data: List of health target records.
    """
    # Import health data
    if health_weight_data:
        for health_weight in health_weight_data:
            health_weight.pop("id", None)
            health_weight.pop("user_id", None)

            # Convert string numeric values to floats
            numeric_fields = [
                "weight",
                "bmi",
                "body_fat",
                "body_water",
                "bone_mass",
                "muscle_mass",
                "visceral_fat",
            ]
            for field in numeric_fields:
                if field in health_weight and isinstance(health_weight[field], str):
                    try:
                        health_weight[field] = float(health_weight[field])
                    except (ValueError, TypeError):
                        health_weight[field] = None

            # Convert integer fields
            int_fields = ["physique_rating", "metabolic_age"]
            for field in int_fields:
                if field in health_weight and isinstance(health_weight[field], str):
                    try:
                        health_weight[field] = int(health_weight[field])
                    except (ValueError, TypeError):
                        health_weight[field] = None

            data = health_weight_schema.HealthWeightCreate(**health_weight)
            health_weight_crud.create_health_weight(self.user_id, data, self.db)
            self.counts["health_weight"] += 1
        core_logger.print_to_log(f"Imported {self.counts['health_weight']} health weight records", "info")
    else:
        core_logger.print_to_log("No health weight data to import", "debug")

    # Import health targets
    if health_targets_data:
        for target_data in health_targets_data:
            current_health_target = health_targets_crud.get_health_targets_by_user_id(self.user_id, self.db)

            # Convert string numeric values to floats/ints
            if isinstance(target_data.get("weight"), str):
                try:
                    target_data["weight"] = float(target_data["weight"])
                except (ValueError, TypeError):
                    target_data["weight"] = None

            int_fields = ["steps", "sleep"]
            for field in int_fields:
                if isinstance(target_data.get(field), str):
                    try:
                        target_data[field] = int(target_data[field])
                    except (ValueError, TypeError):
                        target_data[field] = None

            target_data["user_id"] = self.user_id
            if current_health_target is not None:
                target_data["id"] = current_health_target.id
            else:
                target_data.pop("id", None)

            target = health_targets_schema.HealthTargetsUpdate(**target_data)
            health_targets_crud.edit_health_target(target, self.user_id, self.db)
            self.counts["health_targets"] += 1
        core_logger.print_to_log(f"Imported {self.counts['health_targets']} health targets", "info")
    else:
        core_logger.print_to_log("No health targets to import", "debug")

collect_and_import_user_data async

collect_and_import_user_data(user_data, user_default_gear_data, user_goals_data, user_integrations_data, user_privacy_settings_data, gears_id_mapping)

Import user profile and related settings.

Parameters:

Name	Type	Description	Default
user_data	list[Any]	User profile data.	required
user_default_gear_data	list[Any]	Default gear settings.	required
user_goals_data	list[Any]	User goals data.	required
user_integrations_data	list[Any]	Integration settings.	required
user_privacy_settings_data	list[Any]	Privacy settings.	required
gears_id_mapping	dict[int, int]	Mapping of old to new gear IDs.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_user_data(
    self,
    user_data: list[Any],
    user_default_gear_data: list[Any],
    user_goals_data: list[Any],
    user_integrations_data: list[Any],
    user_privacy_settings_data: list[Any],
    gears_id_mapping: dict[int, int],
) -> None:
    """
    Import user profile and related settings.

    Args:
        user_data: User profile data.
        user_default_gear_data: Default gear settings.
        user_goals_data: User goals data.
        user_integrations_data: Integration settings.
        user_privacy_settings_data: Privacy settings.
        gears_id_mapping: Mapping of old to new gear IDs.
    """
    if user_data:
        # Import user profile
        user_profile = user_data[0]
        user_profile["id"] = self.user_id

        # Handle photo path
        photo_path = user_profile.get("photo_path")
        if isinstance(photo_path, str) and photo_path.startswith("data/user_images/"):
            extension = photo_path.split(".")[-1]
            user_profile["photo_path"] = f"data/user_images/{self.user_id}.{extension}"

        # Strict allow-list before persistence: profile import is a
        # self-service operation and MUST NOT be a back door for
        # privilege escalation. Even if the source ZIP was crafted
        # to set ``access_type``, ``active``, ``mfa_enabled``,
        # ``mfa_secret``, ``email_verified``, or
        # ``pending_admin_approval``, those fields are dropped here
        # before reaching the database.
        sanitized = {
            key: value for key, value in user_profile.items() if key in users_crud.PROFILE_SELF_SERVICE_FIELDS
        }
        profile_payload = users_schema.ProfileUpdate.model_validate(sanitized)
        await users_crud.edit_profile_user(self.user_id, profile_payload, self.db)
        self.counts["user"] += 1
    else:
        core_logger.print_to_log("No user data to import, but continuing with sub-settings", "info")

    # Import user-related settings — always attempt, even if
    # user_data was empty, so that sub-setting data loaded from
    # the ZIP is not silently discarded.
    await self.collect_and_import_user_default_gear(user_default_gear_data, gears_id_mapping)
    await self.collect_and_import_user_goals(user_goals_data)
    await self.collect_and_import_user_integrations(user_integrations_data)
    await self.collect_and_import_user_privacy_settings(user_privacy_settings_data)

collect_and_import_user_default_gear async

collect_and_import_user_default_gear(user_default_gear_data, gears_id_mapping)

Import user default gear settings with ID remapping.

Parameters:

Name	Type	Description	Default
user_default_gear_data	list[Any]	Default gear data.	required
gears_id_mapping	dict[int, int]	Mapping of old to new gear IDs.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_user_default_gear(
    self, user_default_gear_data: list[Any], gears_id_mapping: dict[int, int]
) -> None:
    """
    Import user default gear settings with ID remapping.

    Args:
        user_default_gear_data: Default gear data.
        gears_id_mapping: Mapping of old to new gear IDs.
    """
    if not user_default_gear_data:
        core_logger.print_to_log("No user default gear data to import", "info")
        return

    current_user_default_gear = user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)

    if current_user_default_gear is None:
        core_logger.print_to_log("No existing user default gear, creating new record", "info")
        user_default_gear_crud.create_user_default_gear(self.user_id, self.db)
        current_user_default_gear = user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)

    gear_data = user_default_gear_data[0]
    gear_data["id"] = current_user_default_gear.id
    gear_data["user_id"] = self.user_id

    # Map gear IDs
    gear_fields = [
        "run_gear_id",
        "trail_run_gear_id",
        "virtual_run_gear_id",
        "ride_gear_id",
        "gravel_ride_gear_id",
        "mtb_ride_gear_id",
        "virtual_ride_gear_id",
        "ows_gear_id",
        "walk_gear_id",
        "hike_gear_id",
        "tennis_gear_id",
        "alpine_ski_gear_id",
        "nordic_ski_gear_id",
        "snowboard_gear_id",
        "windsurf_gear_id",
    ]

    for field in gear_fields:
        old_gear_id = gear_data.get(field)
        gear_data[field] = gears_id_mapping.get(old_gear_id)

    user_default_gear = user_default_gear_schema.UsersDefaultGearUpdate(**gear_data)
    user_default_gear_crud.edit_user_default_gear(user_default_gear, self.user_id, self.db)
    core_logger.print_to_log("Imported user default gear", "info")
    self.counts["user_default_gear"] += 1

collect_and_import_user_goals async

collect_and_import_user_goals(user_goals_data)

Import user goals data.

Parameters:

Name	Type	Description	Default
user_goals_data	list[Any]	List of user goal dictionaries.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_user_goals(self, user_goals_data: list[Any]) -> None:
    """
    Import user goals data.

    Args:
        user_goals_data: List of user goal dictionaries.
    """
    if not user_goals_data:
        core_logger.print_to_log("No user goals data to import", "info")
        return

    for goal_data in user_goals_data:
        goal_data.pop("id", None)
        goal_data.pop("user_id", None)

        goal = user_goals_schema.UsersGoalCreate(**goal_data)
        user_goals_crud.create_user_goal(self.user_id, goal, self.db)
        self.counts["user_goals"] += 1

    core_logger.print_to_log(f"Imported {self.counts['user_goals']} user goals", "info")

collect_and_import_user_integrations async

collect_and_import_user_integrations(user_integrations_data)

Import user integration settings.

Parameters:

Name	Type	Description	Default
user_integrations_data	list[Any]	Integration data.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_user_integrations(self, user_integrations_data: list[Any]) -> None:
    """
    Import user integration settings.

    Args:
        user_integrations_data: Integration data.
    """
    if not user_integrations_data:
        core_logger.print_to_log("No user integrations data to import", "info")
        return

    integrations_data = user_integrations_data[0]
    integrations_data.pop("id", None)
    integrations_data.pop("user_id", None)

    user_integrations = users_integrations_schema.UsersIntegrationsUpdate(**integrations_data)
    user_integrations_crud.edit_user_integrations(user_integrations, self.user_id, self.db)
    core_logger.print_to_log("Imported user integrations", "info")
    self.counts["user_integrations"] += 1

collect_and_import_user_privacy_settings async

collect_and_import_user_privacy_settings(user_privacy_settings_data)

Import user privacy settings.

Parameters:

Name	Type	Description	Default
user_privacy_settings_data	list[Any]	Privacy settings data.	required

Source code in backend/app/users/users_profile/import_service.py

async def collect_and_import_user_privacy_settings(self, user_privacy_settings_data: list[Any]) -> None:
    """
    Import user privacy settings.

    Args:
        user_privacy_settings_data: Privacy settings data.
    """
    if not user_privacy_settings_data:
        core_logger.print_to_log("No user privacy settings data to import", "info")
        return

    privacy_data = user_privacy_settings_data[0]
    privacy_data.pop("id", None)
    privacy_data.pop("user_id", None)

    user_privacy_settings = users_privacy_settings_schema.UsersPrivacySettingsUpdate(**privacy_data)
    users_privacy_settings_crud.edit_user_privacy_settings(self.user_id, user_privacy_settings, self.db)
    core_logger.print_to_log("Imported user privacy settings", "info")
    self.counts["user_privacy_settings"] += 1

import_from_zip_data async

import_from_zip_data(zip_data)

Import profile data from ZIP file bytes.

Parameters:

Name	Type	Description	Default
zip_data	bytes	ZIP file content as bytes.	required

Returns:

Type	Description
dict[str, Any]	Dictionary with import results and counts.

Raises:

Type	Description
FileSizeError	If file exceeds size limit.
FileFormatError	If ZIP format is invalid.
FileSystemError	If file system error occurs.
ImportTimeoutError	If operation times out.

Source code in backend/app/users/users_profile/import_service.py

async def import_from_zip_data(self, zip_data: bytes) -> dict[str, Any]:
    """
    Import profile data from ZIP file bytes.

    Args:
        zip_data: ZIP file content as bytes.

    Returns:
        Dictionary with import results and counts.

    Raises:
        FileSizeError: If file exceeds size limit.
        FileFormatError: If ZIP format is invalid.
        FileSystemError: If file system error occurs.
        ImportTimeoutError: If operation times out.
    """
    start_time = time.time()
    timeout_seconds = self.performance_config.timeout_seconds

    # Check file size
    file_size_mb = len(zip_data) / (1024 * 1024)
    if file_size_mb > self.performance_config.max_file_size_mb:
        raise FileSizeError(
            f"ZIP file size ({file_size_mb:.1f}MB) exceeds maximum allowed "
            f"({self.performance_config.max_file_size_mb}MB)"
        )

    # Early memory check BEFORE loading any data
    profile_utils.check_memory_usage(
        "pre-import memory check",
        self.performance_config.max_memory_mb,
        self.performance_config.enable_memory_monitoring,
    )

    try:
        with zipfile.ZipFile(BytesIO(zip_data)) as zipf:
            file_list = set(zipf.namelist())

            # Create ID mappings for relationships
            gears_id_mapping: dict[int, int] = {}
            activities_id_mapping: dict[int, int] = {}

            # Import data in dependency order using streaming approach
            # Load and import gears
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
            gears_data = self._load_single_json(zipf, "data/gears.json")
            gears_id_mapping = await self.collect_and_import_gears_data(gears_data)
            del gears_data  # Explicit memory cleanup

            # Load and import gear components
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
            gear_components_data = self._load_single_json(zipf, "data/gear_components.json")
            await self.collect_and_import_gear_components_data(gear_components_data, gears_id_mapping)
            del gear_components_data

            # Load and import user data (includes user, default gear, integrations, goals, privacy)
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
            user_data = self._load_single_json(zipf, "data/user.json")
            user_default_gear_data = self._load_single_json(zipf, "data/user_default_gear.json")
            user_goals_data = self._load_single_json(zipf, "data/user_goals.json")
            user_integrations_data = self._load_single_json(zipf, "data/user_integrations.json")
            user_privacy_settings_data = self._load_single_json(zipf, "data/user_privacy_settings.json")

            await self.collect_and_import_user_data(
                user_data,
                user_default_gear_data,
                user_goals_data,
                user_integrations_data,
                user_privacy_settings_data,
                gears_id_mapping,
            )
            del user_data, user_default_gear_data, user_integrations_data
            del user_goals_data, user_privacy_settings_data

            # Load and import activities with their components
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")

            # Import activities and components using batched approach to avoid memory issues
            activities_id_mapping = await self.collect_and_import_activities_data_batched(
                zipf,
                file_list,
                gears_id_mapping,
                start_time,
                timeout_seconds,
            )

            # Load and import notifications
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
            notifications_data = self._load_single_json(zipf, "data/notifications.json")

            await self.collect_and_import_notifications_data(notifications_data)
            del notifications_data

            # Load and import health data
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
            health_weight_data = self._load_single_json(zipf, "data/health_weight.json")
            health_targets_data = self._load_single_json(zipf, "data/health_targets.json")

            await self.collect_and_import_health_weight(health_weight_data, health_targets_data)
            del health_weight_data, health_targets_data

            # Import files and media
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
            await self.add_activity_files_from_zip(zipf, file_list, activities_id_mapping)
            await self.add_activity_media_from_zip(zipf, file_list, activities_id_mapping)
            await self.add_user_images_from_zip(zipf, file_list)

    except zipfile.BadZipFile as e:
        raise FileFormatError(f"Invalid ZIP file format: {e!s}") from e
    except OSError as e:
        raise FileSystemError(f"File system error during import: {e!s}") from e

    # Guard: if nothing was imported, fail visibly instead of
    # returning a misleading "success" with zero counts.
    if all(v == 0 for v in self.counts.values()):
        raise FileFormatError(
            "Import completed with zero items imported. The ZIP file contains no importable data."
        )

    return {"detail": "Import completed", "imported": self.counts}

ImportTimeoutError

Bases: ProfileImportError

Raised when import operation exceeds time limit.

Source code in backend/app/users/users_profile/exceptions.py

class ImportTimeoutError(ProfileImportError):
    """
    Raised when import operation exceeds time limit.
    """

    pass

ImportValidationError

Bases: ProfileImportError

Raised when import data fails validation checks.

Source code in backend/app/users/users_profile/exceptions.py

class ImportValidationError(ProfileImportError):
    """
    Raised when import data fails validation checks.
    """

    pass

JSONParseError

Bases: ProfileImportError

Raised when JSON data cannot be parsed.

Source code in backend/app/users/users_profile/exceptions.py

class JSONParseError(ProfileImportError):
    """
    Raised when JSON data cannot be parsed.
    """

    pass

MemoryAllocationError

Bases: ProfileOperationError

Raised when memory allocation fails.

Source code in backend/app/users/users_profile/exceptions.py

class MemoryAllocationError(ProfileOperationError):
    """
    Raised when memory allocation fails.
    """

    pass

ProfileImportError

Bases: ProfileOperationError

Base exception for import operations.

Source code in backend/app/users/users_profile/exceptions.py

class ProfileImportError(ProfileOperationError):
    """
    Base exception for import operations.
    """

    pass

ProfileOperationError

Bases: Exception

Base exception for all profile operations.

Source code in backend/app/users/users_profile/exceptions.py

class ProfileOperationError(Exception):
    """
    Base exception for all profile operations.
    """

    pass

SchemaValidationError

Bases: ProfileImportError

Raised when imported data doesn't match expected schema.

Source code in backend/app/users/users_profile/exceptions.py

class SchemaValidationError(ProfileImportError):
    """
    Raised when imported data doesn't match expected schema.
    """

    pass

ZipCreationError

Bases: ExportError

Raised when ZIP file creation or writing fails.

Source code in backend/app/users/users_profile/exceptions.py

class ZipCreationError(ExportError):
    """
    Raised when ZIP file creation or writing fails.
    """

    pass

ZipStructureError

Bases: ProfileImportError

Raised when ZIP file structure is invalid.

Source code in backend/app/users/users_profile/exceptions.py

class ZipStructureError(ProfileImportError):
    """
    Raised when ZIP file structure is invalid.
    """

    pass