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

• Schemas: MFARequest, MFASetupRequest, MFASetupResponse, MFAStatusResponse
• MFA Store: MFASecretStore
• Services: ExportService, ImportService
• Exceptions: ProfileOperationError, ExportError, ProfileImportError, and related exception classes
• Utils: setup_user_mfa, enable_user_mfa, disable_user_mfa, verify_user_mfa, is_mfa_enabled_for_user

ActivityLimitError

Bases: ProfileImportError

Raised when import contains too many activities.

Source code in backend/app/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/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/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/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/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/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/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
                batch_activities = self._get_activities_batch(offset, batch_size)

                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 "
                    f"(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}", "warning", exc=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 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}",
                "warning",
                exc=err,
            )
            return []

    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}",
                    "warning",
                    exc=err,
                )

            core_logger.print_to_log(
                f"Processed {component_key} batch {i//batch_size + 1} "
                f"({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}",
                    "warning",
                    exc=err,
                )

            core_logger.print_to_log(
                f"Processed {component_key} batch {i//batch_size + 1} "
                f"({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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/gears.json", [], self.counts
                )

            # 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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/gear_components.json", [], self.counts
                )

        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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/health_weight.json", [], self.counts
                )

            # 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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/health_targets.json", [], self.counts
                )

        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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/notifications.json", [], self.counts
                )
        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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/user_default_gear.json", [], self.counts
                )

            # 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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/user_goals.json", [], self.counts
                )

            # Collect and write user identity providers
            try:
                user_identity_providers = (
                    user_identity_providers_crud.get_user_identity_providers_by_user_id(
                        self.user_id, self.db
                    )
                )
                if user_identity_providers:
                    identity_providers_dict = [
                        profile_utils.sqlalchemy_obj_to_dict(uidp)
                        for uidp in user_identity_providers
                    ]
                    profile_utils.write_json_to_zip(
                        zipf,
                        "data/user_identity_providers.json",
                        identity_providers_dict,
                        self.counts,
                    )
                else:
                    profile_utils.write_json_to_zip(
                        zipf, "data/user_identity_providers.json", [], self.counts
                    )
            except Exception as err:
                core_logger.print_to_log(
                    f"Failed to collect user identity providers: {err}",
                    "warning",
                    exc=err,
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/user_identity_providers.json", [], self.counts
                )

            # 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}", "warning", exc=err
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/user_integrations.json", [], self.counts
                )

            # 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}",
                    "warning",
                    exc=err,
                )
                profile_utils.write_json_to_zip(
                    zipf, "data/user_privacy_settings.json", [], self.counts
                )

        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, IOError) as err:
                        core_logger.print_to_log(
                            f"Failed to add activity file {file}: {err}",
                            "warning",
                            exc=err,
                        )
                        continue
                    except Exception as err:
                        core_logger.print_to_log(
                            f"Unexpected error adding activity file {file}: {err}",
                            "warning",
                            exc=err,
                        )
                        continue

        except (OSError, IOError) 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.ACTIVITY_MEDIA_DIR):
                core_logger.print_to_log(
                    f"Media directory does not exist: {core_config.ACTIVITY_MEDIA_DIR}",
                    "warning",
                )
                return

            for root, _, files in os.walk(core_config.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.ACTIVITY_MEDIA_DIR
                                ),
                            )
                            zipf.write(file_path, arcname)
                            self.counts["media"] += 1

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

        except (OSError, IOError) 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 Exception 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 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, None, None]:
        """
        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
                except MemoryAllocationError as err:
                    raise err
                except Exception as err:
                    raise 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/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, IOError) as err:
                    core_logger.print_to_log(
                        f"Failed to add activity file {file}: {err}",
                        "warning",
                        exc=err,
                    )
                    continue
                except Exception as err:
                    core_logger.print_to_log(
                        f"Unexpected error adding activity file {file}: {err}",
                        "warning",
                        exc=err,
                    )
                    continue

    except (OSError, IOError) 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/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.ACTIVITY_MEDIA_DIR):
            core_logger.print_to_log(
                f"Media directory does not exist: {core_config.ACTIVITY_MEDIA_DIR}",
                "warning",
            )
            return

        for root, _, files in os.walk(core_config.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.ACTIVITY_MEDIA_DIR
                            ),
                        )
                        zipf.write(file_path, arcname)
                        self.counts["media"] += 1

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

    except (OSError, IOError) 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/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 Exception 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/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}", "warning", exc=err
            )
            profile_utils.write_json_to_zip(
                zipf, "data/gears.json", [], self.counts
            )

        # 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}", "warning", exc=err
            )
            profile_utils.write_json_to_zip(
                zipf, "data/gear_components.json", [], self.counts
            )

    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/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}", "warning", exc=err
            )
            profile_utils.write_json_to_zip(
                zipf, "data/health_weight.json", [], self.counts
            )

        # 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}", "warning", exc=err
            )
            profile_utils.write_json_to_zip(
                zipf, "data/health_targets.json", [], self.counts
            )

    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/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
            batch_activities = self._get_activities_batch(offset, batch_size)

            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 "
                f"(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}", "warning", exc=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 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/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}", "warning", exc=err
            )
            profile_utils.write_json_to_zip(
                zipf, "data/user_default_gear.json", [], self.counts
            )

        # 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}", "warning", exc=err
            )
            profile_utils.write_json_to_zip(
                zipf, "data/user_goals.json", [], self.counts
            )

        # Collect and write user identity providers
        try:
            user_identity_providers = (
                user_identity_providers_crud.get_user_identity_providers_by_user_id(
                    self.user_id, self.db
                )
            )
            if user_identity_providers:
                identity_providers_dict = [
                    profile_utils.sqlalchemy_obj_to_dict(uidp)
                    for uidp in user_identity_providers
                ]
                profile_utils.write_json_to_zip(
                    zipf,
                    "data/user_identity_providers.json",
                    identity_providers_dict,
                    self.counts,
                )
            else:
                profile_utils.write_json_to_zip(
                    zipf, "data/user_identity_providers.json", [], self.counts
                )
        except Exception as err:
            core_logger.print_to_log(
                f"Failed to collect user identity providers: {err}",
                "warning",
                exc=err,
            )
            profile_utils.write_json_to_zip(
                zipf, "data/user_identity_providers.json", [], self.counts
            )

        # 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}", "warning", exc=err
            )
            profile_utils.write_json_to_zip(
                zipf, "data/user_integrations.json", [], self.counts
            )

        # 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}",
                "warning",
                exc=err,
            )
            profile_utils.write_json_to_zip(
                zipf, "data/user_privacy_settings.json", [], self.counts
            )

    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
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/profile/export_service.py

def generate_export_archive(
    self, user_dict: dict[str, Any], timeout_seconds: int | None = 300
) -> Generator[bytes, None, None]:
    """
    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
            except MemoryAllocationError as err:
                raise err
            except Exception as err:
                raise 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/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/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/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/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/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 = {}
                activities_id_mapping = {}

                # 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_identity_providers_data = self._load_single_json(
                    zipf, "data/user_identity_providers.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_identity_providers_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: {str(e)}") from e
        except (OSError, IOError) as e:
            raise FileSystemError(f"File system error during import: {str(e)}") from e

        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:
                return []

            data = json.loads(zipf.read(filename))
            core_logger.print_to_log(
                f"Loaded {len(data) if isinstance(data, list) else 1} 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

    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 = {}

        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.Gear(**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["user_id"] = self.user_id
            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.GearComponents(
                **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_identity_providers_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_identity_providers_data: Identity providers data.
            user_integrations_data: Integration settings.
            user_privacy_settings_data: Privacy settings.
            gears_id_mapping: Mapping of old to new gear IDs.
        """
        if not user_data:
            core_logger.print_to_log("No user data to import", "info")
            return

        # 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}"

        user = users_schema.UsersRead(**user_profile)
        await users_crud.edit_user(self.user_id, user, self.db)
        self.counts["user"] += 1

        # Import user-related settings
        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_identity_providers(
            user_identity_providers_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 to update", "info")
            return

        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)
            if old_gear_id in gears_id_mapping:
                gear_data[field] = gears_id_mapping[old_gear_id]
            else:
                gear_data[field] = None

        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(f"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(f"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_identity_providers(
        self, user_identity_providers_data: list[Any]
    ) -> None:
        """
        Import user identity provider links.

        Args:
            user_identity_providers_data: Identity provider data.
        """
        if not user_identity_providers_data:
            core_logger.print_to_log(
                "No user identity providers data to import", "info"
            )
            return

        # Check if identity provider exists
        idps = identity_providers_crud.get_all_identity_providers(self.db)
        if not idps:
            core_logger.print_to_log(
                f"Skipping identity provider link: "
                f"there are no identity providers configured in the system.",
                "warning",
            )
            return

        for provider_data in user_identity_providers_data:
            if not identity_providers_crud.get_identity_provider(
                provider_data["idp_id"], self.db
            ):
                core_logger.print_to_log(
                    f"Skipping identity provider link for idp_id={provider_data['idp_id']}: "
                    f"identity provider not found in the system.",
                    "warning",
                )
                continue

            provider_data.pop("id", None)
            provider_data.pop("user_id", None)

            user_identity_providers_crud.create_user_identity_provider(
                self.user_id,
                provider_data["idp_id"],
                provider_data["idp_subject"],
                self.db,
            )
            self.counts["user_identity_providers"] += 1

        core_logger.print_to_log(
            f"Imported {self.counts['user_identity_providers']} user identity providers",
            "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(f"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.ActivitySets(**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 = old_path.split("/")[-1]
                    suffix = filename.split("_", 1)[1]
                    media_data["media_path"] = (
                        f"{core_config.ACTIVITY_MEDIA_DIR}/{new_activity_id}_{suffix}"
                    )

                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 = {}

        # 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}: "
                f"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(zipf.read(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} "
                        f"from {filename} for batch",
                        "debug",
                    )
            except json.JSONDecodeError as err:
                core_logger.print_to_log(
                    f"Failed to parse {filename}: {err}", "warning"
                )
            except Exception 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.Notification(**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(f"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(f"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 using file_uploads
                    file_bytes = zipf.read(file_path)
                    await file_uploads.save_file(
                        file_bytes, core_config.FILES_PROCESSED_DIR, new_file_name
                    )
                    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 using file_uploads
                        file_bytes = zipf.read(file_path)
                        await file_uploads.save_file(
                            file_bytes, core_config.ACTIVITY_MEDIA_DIR, new_file_name
                        )
                        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 using file_uploads
                file_bytes = zipf.read(file_path)
                await file_uploads.save_file(
                    file_bytes, core_config.USER_IMAGES_DIR, new_file_name
                )
                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/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 using file_uploads
                file_bytes = zipf.read(file_path)
                await file_uploads.save_file(
                    file_bytes, core_config.FILES_PROCESSED_DIR, new_file_name
                )
                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/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 using file_uploads
                    file_bytes = zipf.read(file_path)
                    await file_uploads.save_file(
                        file_bytes, core_config.ACTIVITY_MEDIA_DIR, new_file_name
                    )
                    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/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 using file_uploads
            file_bytes = zipf.read(file_path)
            await file_uploads.save_file(
                file_bytes, core_config.USER_IMAGES_DIR, new_file_name
            )
            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/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 = {}

    # 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}: "
            f"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/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.ActivitySets(**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 = old_path.split("/")[-1]
                suffix = filename.split("_", 1)[1]
                media_data["media_path"] = (
                    f"{core_config.ACTIVITY_MEDIA_DIR}/{new_activity_id}_{suffix}"
                )

            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/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["user_id"] = self.user_id
        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.GearComponents(
            **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/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 = {}

    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.Gear(**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/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(f"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(f"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_identity_providers_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_identity_providers_data	list[Any]	Identity providers 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/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_identity_providers_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_identity_providers_data: Identity providers data.
        user_integrations_data: Integration settings.
        user_privacy_settings_data: Privacy settings.
        gears_id_mapping: Mapping of old to new gear IDs.
    """
    if not user_data:
        core_logger.print_to_log("No user data to import", "info")
        return

    # 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}"

    user = users_schema.UsersRead(**user_profile)
    await users_crud.edit_user(self.user_id, user, self.db)
    self.counts["user"] += 1

    # Import user-related settings
    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_identity_providers(
        user_identity_providers_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/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 to update", "info")
        return

    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)
        if old_gear_id in gears_id_mapping:
            gear_data[field] = gears_id_mapping[old_gear_id]
        else:
            gear_data[field] = None

    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(f"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/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_identity_providers async

collect_and_import_user_identity_providers(user_identity_providers_data)

Import user identity provider links.

Parameters:

Name	Type	Description	Default
user_identity_providers_data	list[Any]	Identity provider data.	required

Source code in backend/app/profile/import_service.py

async def collect_and_import_user_identity_providers(
    self, user_identity_providers_data: list[Any]
) -> None:
    """
    Import user identity provider links.

    Args:
        user_identity_providers_data: Identity provider data.
    """
    if not user_identity_providers_data:
        core_logger.print_to_log(
            "No user identity providers data to import", "info"
        )
        return

    # Check if identity provider exists
    idps = identity_providers_crud.get_all_identity_providers(self.db)
    if not idps:
        core_logger.print_to_log(
            f"Skipping identity provider link: "
            f"there are no identity providers configured in the system.",
            "warning",
        )
        return

    for provider_data in user_identity_providers_data:
        if not identity_providers_crud.get_identity_provider(
            provider_data["idp_id"], self.db
        ):
            core_logger.print_to_log(
                f"Skipping identity provider link for idp_id={provider_data['idp_id']}: "
                f"identity provider not found in the system.",
                "warning",
            )
            continue

        provider_data.pop("id", None)
        provider_data.pop("user_id", None)

        user_identity_providers_crud.create_user_identity_provider(
            self.user_id,
            provider_data["idp_id"],
            provider_data["idp_subject"],
            self.db,
        )
        self.counts["user_identity_providers"] += 1

    core_logger.print_to_log(
        f"Imported {self.counts['user_identity_providers']} user identity providers",
        "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/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(f"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/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(f"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/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 = {}
            activities_id_mapping = {}

            # 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_identity_providers_data = self._load_single_json(
                zipf, "data/user_identity_providers.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_identity_providers_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: {str(e)}") from e
    except (OSError, IOError) as e:
        raise FileSystemError(f"File system error during import: {str(e)}") from e

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

ImportTimeoutError

Bases: ProfileImportError

Raised when import operation exceeds time limit.

Source code in backend/app/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/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/profile/exceptions.py

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

    pass

MFARequest

Bases: BaseModel

Request model for MFA verification.

Attributes:

Name	Type	Description
mfa_code	StrictStr	The MFA code to verify (6-digit TOTP or 9-char backup code).

Source code in backend/app/profile/schema.py

class MFARequest(BaseModel):
    """
    Request model for MFA verification.

    Attributes:
        mfa_code: The MFA code to verify (6-digit TOTP or
            9-char backup code).
    """

    mfa_code: StrictStr = Field(
        ...,
        min_length=6,
        max_length=9,
        description="MFA code (6-digit TOTP or XXXX-XXXX)",
    )

    model_config = ConfigDict(
        extra="forbid",
        validate_assignment=True,
    )

    @field_validator("mfa_code")
    @classmethod
    def validate_mfa_code_format(cls, v: str) -> str:
        """
        Validate MFA code format.

        Args:
            v: MFA code to validate.

        Returns:
            Validated MFA code.

        Raises:
            ValueError: If code format is invalid.
        """
        normalized = v.strip().upper()
        # 6-digit TOTP
        if len(normalized) == 6 and normalized.isdigit():
            return normalized
        # 9-char backup code (XXXX-XXXX)
        if len(normalized) == 9 and normalized[4] == "-":
            return normalized
        raise ValueError("MFA code must be 6-digit TOTP or XXXX-XXXX")

validate_mfa_code_format classmethod

validate_mfa_code_format(v)

Validate MFA code format.

Parameters:

Name	Type	Description	Default
v	str	MFA code to validate.	required

Returns:

Type	Description
str	Validated MFA code.

Raises:

Type	Description
ValueError	If code format is invalid.

Source code in backend/app/profile/schema.py

@field_validator("mfa_code")
@classmethod
def validate_mfa_code_format(cls, v: str) -> str:
    """
    Validate MFA code format.

    Args:
        v: MFA code to validate.

    Returns:
        Validated MFA code.

    Raises:
        ValueError: If code format is invalid.
    """
    normalized = v.strip().upper()
    # 6-digit TOTP
    if len(normalized) == 6 and normalized.isdigit():
        return normalized
    # 9-char backup code (XXXX-XXXX)
    if len(normalized) == 9 and normalized[4] == "-":
        return normalized
    raise ValueError("MFA code must be 6-digit TOTP or XXXX-XXXX")

MFASecretStore

Thread-safe storage for temporary MFA secrets with TTL.

Attributes:

Name	Type	Description
_store		Internal storage for encrypted secrets.
_lock		Thread synchronization lock.
_ttl_seconds		Time-to-live for stored secrets.
_cleanup_thread		Background thread for cleanup.

Source code in backend/app/profile/mfa_store.py

class MFASecretStore:
    """
    Thread-safe storage for temporary MFA secrets with TTL.

    Attributes:
        _store: Internal storage for encrypted secrets.
        _lock: Thread synchronization lock.
        _ttl_seconds: Time-to-live for stored secrets.
        _cleanup_thread: Background thread for cleanup.
    """

    def __init__(self, ttl_seconds: int = 300):
        """
        Initialize the MFA secret store.

        Args:
            ttl_seconds: Time-to-live for secrets in seconds.
        """
        self._store = (
            {}
        )  # Format: {user_id: {"encrypted_secret": str, "expires_at": float}}
        self._lock = threading.RLock()  # Reentrant lock for thread safety
        self._ttl_seconds = ttl_seconds
        self._cleanup_thread = None
        self._start_cleanup_thread()

    def _start_cleanup_thread(self):
        """
        Start the background cleanup thread if not running.
        """
        if self._cleanup_thread is None or not self._cleanup_thread.is_alive():
            self._cleanup_thread = threading.Thread(
                target=self._cleanup_expired_secrets,
                daemon=True,
                name="MFA-Secret-Cleanup",
            )
            self._cleanup_thread.start()

    def _cleanup_expired_secrets(self):
        """
        Continuously remove expired secrets from storage.
        """
        while True:
            try:
                current_time = time.time()
                with self._lock:
                    expired_users = [
                        user_id
                        for user_id, data in self._store.items()
                        if current_time > data["expires_at"]
                    ]
                    for user_id in expired_users:
                        del self._store[user_id]
                        core_logger.print_to_log(
                            f"Cleaned up expired MFA secret for user {user_id}", "debug"
                        )

                time.sleep(30)
            except Exception as err:
                core_logger.print_to_log(
                    f"Error in MFA secret cleanup thread: {err}", "error", exc=err
                )
                time.sleep(60)

    def add_secret(self, user_id: int, secret: str) -> None:
        """
        Store an encrypted MFA secret with expiration.

        Args:
            user_id: The user ID to associate with the secret.
            secret: The plaintext MFA secret to encrypt and store.

        Raises:
            ValueError: If encryption fails.
            Exception: If storage operation fails.
        """
        try:
            encrypted_secret = core_cryptography.encrypt_token_fernet(secret)
            if not encrypted_secret:
                raise ValueError("Failed to encrypt MFA secret")

            expires_at = time.time() + self._ttl_seconds

            with self._lock:
                self._store[user_id] = {
                    "encrypted_secret": encrypted_secret,
                    "expires_at": expires_at,
                }

            core_logger.print_to_log(
                f"Securely stored MFA secret for user {user_id} (expires in {self._ttl_seconds}s)",
                "debug",
            )
        except Exception as err:
            core_logger.print_to_log(
                f"Failed to add MFA secret for user {user_id}: {err}", "error", exc=err
            )
            raise

    def get_secret(self, user_id: int) -> str | None:
        """
        Retrieve and decrypt an MFA secret if not expired.

        Args:
            user_id: The user ID to retrieve the secret for.

        Returns:
            The decrypted MFA secret, or None if not found or
            expired.
        """
        try:
            with self._lock:
                if user_id not in self._store:
                    return None

                data = self._store[user_id]

                # Check if expired
                if time.time() > data["expires_at"]:
                    del self._store[user_id]
                    core_logger.print_to_log(
                        f"MFA secret expired for user {user_id}", "debug"
                    )
                    return None

                # Decrypt and return
                decrypted_secret = core_cryptography.decrypt_token_fernet(
                    data["encrypted_secret"]
                )
                return decrypted_secret

        except Exception as err:
            core_logger.print_to_log(
                f"Failed to get MFA secret for user {user_id}: {err}", "error", exc=err
            )
            return None

    def delete_secret(self, user_id: int) -> None:
        """
        Remove an MFA secret from storage.

        Args:
            user_id: The user ID whose secret to delete.
        """
        try:
            with self._lock:
                if user_id in self._store:
                    del self._store[user_id]
                    core_logger.print_to_log(
                        f"Securely deleted MFA secret for user {user_id}", "debug"
                    )
        except Exception as err:
            core_logger.print_to_log(
                f"Failed to delete MFA secret for user {user_id}: {err}",
                "error",
                exc=err,
            )

    def has_secret(self, user_id: int) -> bool:
        """
        Check if a non-expired secret exists for a user.

        Args:
            user_id: The user ID to check.

        Returns:
            True if a valid secret exists, False otherwise.
        """
        try:
            with self._lock:
                if user_id not in self._store:
                    return False

                # Check if expired
                if time.time() > self._store[user_id]["expires_at"]:
                    del self._store[user_id]
                    return False

                return True
        except Exception as err:
            core_logger.print_to_log(
                f"Failed to check MFA secret for user {user_id}: {err}",
                "error",
                exc=err,
            )
            return False

    def clear_all(self) -> None:
        """
        Remove all MFA secrets from storage.
        """
        try:
            with self._lock:
                cleared_count = len(self._store)
                self._store.clear()
                core_logger.print_to_log(
                    f"Cleared {cleared_count} MFA secrets from store", "info"
                )
        except Exception as err:
            core_logger.print_to_log(
                f"Failed to clear MFA secret store: {err}", "error", exc=err
            )

    def get_stats(self) -> dict:
        """
        Get statistics about the secret store.

        Returns:
            Dictionary with total, expired, and active secret
            counts.
        """
        try:
            with self._lock:
                current_time = time.time()
                total_count = len(self._store)
                expired_count = sum(
                    1
                    for data in self._store.values()
                    if current_time > data["expires_at"]
                )

                return {
                    "total_secrets": total_count,
                    "expired_secrets": expired_count,
                    "active_secrets": total_count - expired_count,
                    "ttl_seconds": self._ttl_seconds,
                }
        except Exception as err:
            core_logger.print_to_log(
                f"Failed to get MFA secret store stats: {err}", "error", exc=err
            )
            return {"error": str(err)}

    def __repr__(self) -> str:
        """
        Return a string representation of the store.

        Returns:
            String showing active secrets and TTL.
        """
        stats = self.get_stats()
        return f"MFASecretStore(active={stats.get('active_secrets', 0)}, ttl={self._ttl_seconds}s)"

__init__

__init__(ttl_seconds=300)

Initialize the MFA secret store.

Parameters:

Name	Type	Description	Default
ttl_seconds	int	Time-to-live for secrets in seconds.	300

Source code in backend/app/profile/mfa_store.py

def __init__(self, ttl_seconds: int = 300):
    """
    Initialize the MFA secret store.

    Args:
        ttl_seconds: Time-to-live for secrets in seconds.
    """
    self._store = (
        {}
    )  # Format: {user_id: {"encrypted_secret": str, "expires_at": float}}
    self._lock = threading.RLock()  # Reentrant lock for thread safety
    self._ttl_seconds = ttl_seconds
    self._cleanup_thread = None
    self._start_cleanup_thread()

__repr__

__repr__()

Return a string representation of the store.

Returns:

Type	Description
str	String showing active secrets and TTL.

Source code in backend/app/profile/mfa_store.py

def __repr__(self) -> str:
    """
    Return a string representation of the store.

    Returns:
        String showing active secrets and TTL.
    """
    stats = self.get_stats()
    return f"MFASecretStore(active={stats.get('active_secrets', 0)}, ttl={self._ttl_seconds}s)"

add_secret

add_secret(user_id, secret)

Store an encrypted MFA secret with expiration.

Parameters:

Name	Type	Description	Default
user_id	int	The user ID to associate with the secret.	required
secret	str	The plaintext MFA secret to encrypt and store.	required

Raises:

Type	Description
ValueError	If encryption fails.
Exception	If storage operation fails.

Source code in backend/app/profile/mfa_store.py

def add_secret(self, user_id: int, secret: str) -> None:
    """
    Store an encrypted MFA secret with expiration.

    Args:
        user_id: The user ID to associate with the secret.
        secret: The plaintext MFA secret to encrypt and store.

    Raises:
        ValueError: If encryption fails.
        Exception: If storage operation fails.
    """
    try:
        encrypted_secret = core_cryptography.encrypt_token_fernet(secret)
        if not encrypted_secret:
            raise ValueError("Failed to encrypt MFA secret")

        expires_at = time.time() + self._ttl_seconds

        with self._lock:
            self._store[user_id] = {
                "encrypted_secret": encrypted_secret,
                "expires_at": expires_at,
            }

        core_logger.print_to_log(
            f"Securely stored MFA secret for user {user_id} (expires in {self._ttl_seconds}s)",
            "debug",
        )
    except Exception as err:
        core_logger.print_to_log(
            f"Failed to add MFA secret for user {user_id}: {err}", "error", exc=err
        )
        raise

clear_all

clear_all()

Remove all MFA secrets from storage.

Source code in backend/app/profile/mfa_store.py

def clear_all(self) -> None:
    """
    Remove all MFA secrets from storage.
    """
    try:
        with self._lock:
            cleared_count = len(self._store)
            self._store.clear()
            core_logger.print_to_log(
                f"Cleared {cleared_count} MFA secrets from store", "info"
            )
    except Exception as err:
        core_logger.print_to_log(
            f"Failed to clear MFA secret store: {err}", "error", exc=err
        )

delete_secret

delete_secret(user_id)

Remove an MFA secret from storage.

Parameters:

Name	Type	Description	Default
user_id	int	The user ID whose secret to delete.	required

Source code in backend/app/profile/mfa_store.py

def delete_secret(self, user_id: int) -> None:
    """
    Remove an MFA secret from storage.

    Args:
        user_id: The user ID whose secret to delete.
    """
    try:
        with self._lock:
            if user_id in self._store:
                del self._store[user_id]
                core_logger.print_to_log(
                    f"Securely deleted MFA secret for user {user_id}", "debug"
                )
    except Exception as err:
        core_logger.print_to_log(
            f"Failed to delete MFA secret for user {user_id}: {err}",
            "error",
            exc=err,
        )

get_secret

get_secret(user_id)

Retrieve and decrypt an MFA secret if not expired.

Parameters:

Name	Type	Description	Default
user_id	int	The user ID to retrieve the secret for.	required

Returns:

Type	Description
str | None	The decrypted MFA secret, or None if not found or
str | None	expired.

Source code in backend/app/profile/mfa_store.py

def get_secret(self, user_id: int) -> str | None:
    """
    Retrieve and decrypt an MFA secret if not expired.

    Args:
        user_id: The user ID to retrieve the secret for.

    Returns:
        The decrypted MFA secret, or None if not found or
        expired.
    """
    try:
        with self._lock:
            if user_id not in self._store:
                return None

            data = self._store[user_id]

            # Check if expired
            if time.time() > data["expires_at"]:
                del self._store[user_id]
                core_logger.print_to_log(
                    f"MFA secret expired for user {user_id}", "debug"
                )
                return None

            # Decrypt and return
            decrypted_secret = core_cryptography.decrypt_token_fernet(
                data["encrypted_secret"]
            )
            return decrypted_secret

    except Exception as err:
        core_logger.print_to_log(
            f"Failed to get MFA secret for user {user_id}: {err}", "error", exc=err
        )
        return None

get_stats

get_stats()

Get statistics about the secret store.

Returns:

Type	Description
dict	Dictionary with total, expired, and active secret
dict	counts.

Source code in backend/app/profile/mfa_store.py

def get_stats(self) -> dict:
    """
    Get statistics about the secret store.

    Returns:
        Dictionary with total, expired, and active secret
        counts.
    """
    try:
        with self._lock:
            current_time = time.time()
            total_count = len(self._store)
            expired_count = sum(
                1
                for data in self._store.values()
                if current_time > data["expires_at"]
            )

            return {
                "total_secrets": total_count,
                "expired_secrets": expired_count,
                "active_secrets": total_count - expired_count,
                "ttl_seconds": self._ttl_seconds,
            }
    except Exception as err:
        core_logger.print_to_log(
            f"Failed to get MFA secret store stats: {err}", "error", exc=err
        )
        return {"error": str(err)}

has_secret

has_secret(user_id)

Check if a non-expired secret exists for a user.

Parameters:

Name	Type	Description	Default
user_id	int	The user ID to check.	required

Returns:

Type	Description
bool	True if a valid secret exists, False otherwise.

Source code in backend/app/profile/mfa_store.py

def has_secret(self, user_id: int) -> bool:
    """
    Check if a non-expired secret exists for a user.

    Args:
        user_id: The user ID to check.

    Returns:
        True if a valid secret exists, False otherwise.
    """
    try:
        with self._lock:
            if user_id not in self._store:
                return False

            # Check if expired
            if time.time() > self._store[user_id]["expires_at"]:
                del self._store[user_id]
                return False

            return True
    except Exception as err:
        core_logger.print_to_log(
            f"Failed to check MFA secret for user {user_id}: {err}",
            "error",
            exc=err,
        )
        return False

MFASetupRequest

Bases: BaseModel

Request model for MFA setup verification.

Attributes:

Name	Type	Description
mfa_code	StrictStr	The 6-digit MFA code to verify during setup.

Source code in backend/app/profile/schema.py

class MFASetupRequest(BaseModel):
    """
    Request model for MFA setup verification.

    Attributes:
        mfa_code: The 6-digit MFA code to verify during
            setup.
    """

    mfa_code: StrictStr = Field(
        ...,
        min_length=6,
        max_length=6,
        pattern=r"^\d{6}$",
        description="6-digit TOTP code",
    )

    model_config = ConfigDict(
        extra="forbid",
        validate_assignment=True,
    )

MFASetupResponse

Bases: BaseModel

Response model for MFA setup initialization.

Attributes:

Name	Type	Description
secret	StrictStr	The MFA secret key.
qr_code	StrictStr	Base64-encoded QR code image for setup.
app_name	StrictStr	Application name for MFA setup.

Source code in backend/app/profile/schema.py

class MFASetupResponse(BaseModel):
    """
    Response model for MFA setup initialization.

    Attributes:
        secret: The MFA secret key.
        qr_code: Base64-encoded QR code image for setup.
        app_name: Application name for MFA setup.
    """

    secret: StrictStr = Field(
        ...,
        description="TOTP secret key",
    )
    qr_code: StrictStr = Field(
        ...,
        description="Base64-encoded QR code image",
    )
    app_name: StrictStr = Field(
        default="Endurain",
        description="Application name for MFA",
    )

    model_config = ConfigDict(
        extra="forbid",
    )

MFAStatusResponse

Bases: BaseModel

Response model for MFA status.

Attributes:

Name	Type	Description
mfa_enabled	StrictBool	Whether MFA is enabled for the user.

Source code in backend/app/profile/schema.py

class MFAStatusResponse(BaseModel):
    """
    Response model for MFA status.

    Attributes:
        mfa_enabled: Whether MFA is enabled for the user.
    """

    mfa_enabled: StrictBool = Field(
        ...,
        description="Whether MFA is enabled",
    )

    model_config = ConfigDict(
        extra="forbid",
    )

MemoryAllocationError

Bases: ProfileOperationError

Raised when memory allocation fails.

Source code in backend/app/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/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/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/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/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/profile/exceptions.py

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

    pass

disable_user_mfa

disable_user_mfa(user_id, mfa_code, db)

Disable MFA for user after verification.

Parameters:

Name	Type	Description	Default
user_id	int	User ID to disable MFA for.	required
mfa_code	str	MFA code to verify.	required
db	Session	Database session.	required

Raises:

Type	Description
HTTPException	If user not found, MFA not enabled, code invalid, or decryption fails.

Source code in backend/app/profile/utils.py

def disable_user_mfa(user_id: int, mfa_code: str, db: Session) -> None:
    """
    Disable MFA for user after verification.

    Args:
        user_id: User ID to disable MFA for.
        mfa_code: MFA code to verify.
        db: Database session.

    Raises:
        HTTPException: If user not found, MFA not enabled, code
            invalid, or decryption fails.
    """
    user = users_crud.get_user_by_id(user_id, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
        )

    if not user.mfa_enabled:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="MFA is not enabled for this user",
        )

    # Decrypt the secret
    secret = core_cryptography.decrypt_token_fernet(user.mfa_secret)

    # Check if decryption was successful
    if not secret:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to decrypt MFA secret",
        )

    # Verify the MFA code
    if not verify_totp(secret, mfa_code):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid MFA code"
        )

    # Disable MFA for user
    users_crud.update_user_mfa(user_id, db)

    # Delete all backup codes for user
    mfa_backup_codes_crud.delete_user_backup_codes(user_id, db)

enable_user_mfa

enable_user_mfa(user_id, secret, mfa_code, password_hasher, db)

Enable MFA for user after verification.

Parameters:

Name	Type	Description	Default
user_id	int	User ID to enable MFA for.	required
secret	str	TOTP secret to verify.	required
mfa_code	str	MFA code to verify.	required
password_hasher	PasswordHasher	Password hasher instance for backup code generation.	required
db	Session	Database session.	required

Returns:

Type	Description
list[str]	List of generated backup codes.

Raises:

Type	Description
HTTPException	If user not found, MFA enabled, code invalid, or encryption fails.

Source code in backend/app/profile/utils.py

def enable_user_mfa(
    user_id: int,
    secret: str,
    mfa_code: str,
    password_hasher: auth_password_hasher.PasswordHasher,
    db: Session,
) -> list[str]:
    """
    Enable MFA for user after verification.

    Args:
        user_id: User ID to enable MFA for.
        secret: TOTP secret to verify.
        mfa_code: MFA code to verify.
        password_hasher: Password hasher instance for backup code generation.
        db: Database session.

    Returns:
        List of generated backup codes.

    Raises:
        HTTPException: If user not found, MFA enabled, code
            invalid, or encryption fails.
    """
    user = users_crud.get_user_by_id(user_id, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
        )

    if user.mfa_enabled:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="MFA is already enabled for this user",
        )

    # Verify the MFA code
    if not verify_totp(secret, mfa_code):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid MFA code"
        )

    # Encrypt the secret before storing
    encrypted_secret = core_cryptography.encrypt_token_fernet(secret)

    # Check if encryption was successful
    if not encrypted_secret:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Failed to encrypt MFA secret",
        )

    # Update user with MFA enabled and secret
    users_crud.update_user_mfa(user_id, db, encrypted_secret=encrypted_secret)

    backup_codes = mfa_backup_codes_crud.create_backup_codes(
        user_id, password_hasher, db
    )

    return backup_codes

is_mfa_enabled_for_user

is_mfa_enabled_for_user(user_id, db)

Check if MFA is enabled for user.

Parameters:

Name	Type	Description	Default
user_id	int	User ID to check.	required
db	Session	Database session.	required

Returns:

Type	Description
bool	True if MFA is enabled, False otherwise.

Source code in backend/app/profile/utils.py

def is_mfa_enabled_for_user(user_id: int, db: Session) -> bool:
    """
    Check if MFA is enabled for user.

    Args:
        user_id: User ID to check.
        db: Database session.

    Returns:
        True if MFA is enabled, False otherwise.
    """
    user = users_crud.get_user_by_id(user_id, db)
    if not user:
        return False
    return bool(user.mfa_enabled == 1 and user.mfa_secret is not None)

setup_user_mfa

setup_user_mfa(user_id, db)

Setup MFA for user.

Parameters:

Name	Type	Description	Default
user_id	int	User ID to setup MFA for.	required
db	Session	Database session.	required

Returns:

Type	Description
MFASetupResponse	MFA setup response with secret and QR code.

Raises:

Type	Description
HTTPException	If user not found or MFA enabled.

Source code in backend/app/profile/utils.py

def setup_user_mfa(user_id: int, db: Session) -> profile_schema.MFASetupResponse:
    """
    Setup MFA for user.

    Args:
        user_id: User ID to setup MFA for.
        db: Database session.

    Returns:
        MFA setup response with secret and QR code.

    Raises:
        HTTPException: If user not found or MFA enabled.
    """
    user = users_crud.get_user_by_id(user_id, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
        )

    if user.mfa_enabled:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="MFA is already enabled for this user",
        )

    # Generate new secret
    secret = generate_totp_secret()

    # Generate QR code
    qr_code = generate_qr_code(secret, user.username)

    return profile_schema.MFASetupResponse(
        secret=secret, qr_code=qr_code, app_name="Endurain"
    )

verify_user_mfa

verify_user_mfa(user_id, mfa_code, password_hasher, db)

Verify MFA code for user (TOTP or backup code).

Parameters:

Name	Type	Description	Default
user_id	int	User ID to verify MFA for.	required
mfa_code	str	MFA code to verify (6-digit TOTP or 8-character backup code).	required
password_hasher	PasswordHasher	Password hasher instance for backup code verification.	required
db	Session	Database session.	required

Returns:

Type	Description
bool	True if code is valid, False otherwise.

Raises:

Type	Description
HTTPException	If user not found.

Notes

• First tries TOTP verification (6 digits)
• If TOTP fails and code is 8 characters, tries backup code
• Backup codes are consumed on successful verification

Source code in backend/app/profile/utils.py

def verify_user_mfa(
    user_id: int,
    mfa_code: str,
    password_hasher: auth_password_hasher.PasswordHasher,
    db: Session,
) -> bool:
    """
    Verify MFA code for user (TOTP or backup code).

    Args:
        user_id: User ID to verify MFA for.
        mfa_code: MFA code to verify (6-digit TOTP or 8-character backup code).
        password_hasher: Password hasher instance for backup code verification.
        db: Database session.

    Returns:
        True if code is valid, False otherwise.

    Raises:
        HTTPException: If user not found.

    Notes:
        - First tries TOTP verification (6 digits)
        - If TOTP fails and code is 8 characters, tries backup code
        - Backup codes are consumed on successful verification
    """
    user = users_crud.get_user_by_id(user_id, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
        )

    if not user.mfa_enabled or not user.mfa_secret:
        return False

    # Normalize code (remove whitespaces in the beginning and end, uppercase)
    normalized_code = mfa_code.strip().upper()

    # Try TOTP first (6 digits)
    if len(normalized_code) == 6 and normalized_code.isdigit():
        try:
            secret = core_cryptography.decrypt_token_fernet(user.mfa_secret)
            if not secret:
                core_logger.print_to_log("Failed to decrypt MFA secret", "error")
                return False

            if verify_totp(secret, normalized_code):
                core_logger.print_to_log(
                    f"User {user_id} verified MFA with TOTP", "debug"
                )
                return True
        except (ValueError, pyotp.OTPError) as err:
            core_logger.print_to_log(
                f"Error in TOTP verification: {err}", "error", exc=err
            )
            return False
        except Exception as err:
            core_logger.print_to_log(
                f"Unknown error in TOTP verification: {err}", "error", exc=err
            )
            return False

    # Try backup code (9 alphanumeric characters with dash XXXX-XXXX)
    elif len(normalized_code) == 9 and normalized_code[4] == "-":
        try:
            if mfa_backup_codes_utils.verify_and_consume_backup_code(
                user_id, normalized_code, password_hasher, db
            ):
                return True
        except Exception as err:
            core_logger.print_to_log(
                f"Error in backup code verification: {err}", "error", exc=err
            )
            return False

    # Invalid format or code didn't match
    return False