Skip to content

Users API reference

Users

User module for user account management and authentication.

This module provides comprehensive user management including account creation, profile updates, authentication, MFA support, email verification, and admin approval workflows.

Exports
  • CRUD: get_all_users, get_users_number, get_users_with_pagination, get_user_by_username, get_user_by_email, get_user_by_id, get_users_admin, create_user, create_signup_user, edit_user, approve_user, verify_user_email, update_user_photo, delete_user
  • Schemas: UsersBase, Users, UsersRead, UsersMe, UsersSignup, UsersCreate, UsersEditPassword, UsersListResponse
  • Enums: Gender, Language, WeekDay, UserAccessType
  • Utils: get_user_by_id_or_404, get_admin_users_or_404, check_user_is_active, create_user_default_data, save_user_image_file, delete_user_photo_filesystem

Gender

Bases: Enum

User gender enumeration.

Attributes:

Name Type Description
MALE

Male gender.
FEMALE

Female gender.
UNSPECIFIED

Unspecified or undisclosed gender.
Source code in backend/app/modules/users/users/schema.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
class Gender(Enum):
    """
    User gender enumeration.

    Attributes:
        MALE: Male gender.
        FEMALE: Female gender.
        UNSPECIFIED: Unspecified or undisclosed gender.
    """

    MALE = "male"
    FEMALE = "female"
    UNSPECIFIED = "unspecified"

Language

Bases: Enum

Supported application languages.

Attributes:

Name Type Description
CATALAN

Catalan (ca).
CHINESE_SIMPLIFIED

Simplified Chinese (zh-Hans).
CHINESE_TRADITIONAL

Traditional Chinese (zh-Hant).
GERMAN

German (de).
FRENCH

French (fr).
GALICIAN

Galician (gl).
ITALIAN

Italian (it).
DUTCH

Dutch (nl).
PORTUGUESE

Portuguese (pt-PT).
SLOVENIAN

Slovenian (sl).
SWEDISH

Swedish (sv).
SPANISH

Spanish (es).
ENGLISH

English (en).
POLISH

Polish (pl).
TURKISH

Turkish (tr).
UKRAINIAN

Ukrainian (uk).
ROMANIAN

Romanian (ro).
NORWEGIAN

Norwegian Bokmål (nb).
DANISH

Danish (da).
FINNISH

Finnish (fi).
CZECH

Czech (cs).
GREEK

Greek (el).
HUNGARIAN

Hungarian (hu).
BULGARIAN

Bulgarian (bg).
CROATIAN

Croatian (hr).
SERBIAN

Serbian (sr).
SLOVAK

Slovak (sk).
LITHUANIAN

Lithuanian (lt).
LATVIAN

Latvian (lv).
ESTONIAN

Estonian (et).
Source code in backend/app/modules/users/users/schema.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class Language(Enum):
    """
    Supported application languages.

    Attributes:
        CATALAN: Catalan (ca).
        CHINESE_SIMPLIFIED: Simplified Chinese (zh-Hans).
        CHINESE_TRADITIONAL: Traditional Chinese (zh-Hant).
        GERMAN: German (de).
        FRENCH: French (fr).
        GALICIAN: Galician (gl).
        ITALIAN: Italian (it).
        DUTCH: Dutch (nl).
        PORTUGUESE: Portuguese (pt-PT).
        SLOVENIAN: Slovenian (sl).
        SWEDISH: Swedish (sv).
        SPANISH: Spanish (es).
        ENGLISH: English (en).
        POLISH: Polish (pl).
        TURKISH: Turkish (tr).
        UKRAINIAN: Ukrainian (uk).
        ROMANIAN: Romanian (ro).
        NORWEGIAN: Norwegian Bokmål (nb).
        DANISH: Danish (da).
        FINNISH: Finnish (fi).
        CZECH: Czech (cs).
        GREEK: Greek (el).
        HUNGARIAN: Hungarian (hu).
        BULGARIAN: Bulgarian (bg).
        CROATIAN: Croatian (hr).
        SERBIAN: Serbian (sr).
        SLOVAK: Slovak (sk).
        LITHUANIAN: Lithuanian (lt).
        LATVIAN: Latvian (lv).
        ESTONIAN: Estonian (et).
    """

    CATALAN = "ca"
    CHINESE_SIMPLIFIED = "zh-Hans"
    CHINESE_TRADITIONAL = "zh-Hant"
    GERMAN = "de"
    FRENCH = "fr"
    GALICIAN = "gl"
    ITALIAN = "it"
    DUTCH = "nl"
    PORTUGUESE = "pt-PT"
    SLOVENIAN = "sl"
    SWEDISH = "sv"
    SPANISH = "es"
    ENGLISH = "en"
    POLISH = "pl"
    TURKISH = "tr"
    UKRAINIAN = "uk"
    ROMANIAN = "ro"
    NORWEGIAN = "nb"
    DANISH = "da"
    FINNISH = "fi"
    CZECH = "cs"
    GREEK = "el"
    HUNGARIAN = "hu"
    BULGARIAN = "bg"
    CROATIAN = "hr"
    SERBIAN = "sr"
    SLOVAK = "sk"
    LITHUANIAN = "lt"
    LATVIAN = "lv"
    ESTONIAN = "et"

UserAccessType

Bases: Enum

User access level enumeration.

Attributes:

Name Type Description
REGULAR

Standard user access.
ADMIN

Administrative access.
Source code in backend/app/modules/users/users/schema.py 
128
129
130
131
132
133
134
135
136
137
138
class UserAccessType(Enum):
    """
    User access level enumeration.

    Attributes:
        REGULAR: Standard user access.
        ADMIN: Administrative access.
    """

    REGULAR = "regular"
    ADMIN = "admin"

Users

Bases: UsersBase

Complete users schema with administrative fields.

Note

mfa_secret is intentionally NOT part of any API-facing schema. It lives only on the SQLAlchemy model and is accessed directly by MFA verification utilities. Including it in a Pydantic schema would risk leaking the encrypted seed through responses, exports, logs, or future model_dump callers.

Attributes:

Name Type Description
access_type UserAccessType

User access level.
photo_path StrictStr | None

Servable URL of the user's photo, resolved from the stored storage key. Read-only: writes are ignored.
active StrictBool

Whether the user is active.
mfa_enabled StrictBool

Whether MFA is enabled.
email_verified StrictBool

Whether email is verified.
pending_admin_approval StrictBool

Whether pending admin approval.
Source code in backend/app/modules/users/users/schema.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
class Users(UsersBase):
    """
    Complete users schema with administrative fields.

    Note:
        ``mfa_secret`` is intentionally NOT part of any API-facing
        schema. It lives only on the SQLAlchemy model and is
        accessed directly by MFA verification utilities. Including
        it in a Pydantic schema would risk leaking the encrypted
        seed through responses, exports, logs, or future
        ``model_dump`` callers.

    Attributes:
        access_type: User access level.
        photo_path: Servable URL of the user's photo, resolved from
            the stored storage key. Read-only: writes are ignored.
        active: Whether the user is active.
        mfa_enabled: Whether MFA is enabled.
        email_verified: Whether email is verified.
        pending_admin_approval: Whether pending admin approval.
    """

    access_type: UserAccessType = Field(
        ...,
        description="User access level",
    )
    photo_path: StrictStr | None = Field(
        default=None,
        max_length=500,
        description="Servable URL of the user's photo (read-only)",
    )
    active: StrictBool = Field(
        ...,
        description="Whether the user is active",
    )
    mfa_enabled: StrictBool = Field(
        default=False,
        description="Whether MFA is enabled",
    )
    email_verified: StrictBool = Field(
        default=False,
        description="Whether email is verified",
    )
    pending_admin_approval: StrictBool = Field(
        default=False,
        description="Whether pending admin approval",
    )

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

UsersBase

Bases: BaseModel

Base users schema with common fields.

Attributes:

Name Type Description
name StrictStr

User's full name (1-250 chars).
username StrictStr

Unique username (1-250 chars, alphanumeric and dots).
email EmailStr

User's email address (max 250 chars).
city StrictStr | None

User's city (max 250 chars).
birthdate date | None

User's birthdate.
preferred_language Language

Preferred language.
gender Gender

User's gender.
units Units

User units (metric, imperial).
height StrictInt | None

User's height in centimeters (1-300).
max_heart_rate StrictInt | None

Maximum heart rate in bpm (30-250).
first_day_of_week WeekDay

First day of the week.
currency Currency

User currency (euro, dollar, pound).
timezone StrictStr | None

IANA timezone the athlete lives in.
Source code in backend/app/modules/users/users/schema.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
class UsersBase(BaseModel):
    """
    Base users schema with common fields.

    Attributes:
        name: User's full name (1-250 chars).
        username: Unique username (1-250 chars, alphanumeric
            and dots).
        email: User's email address (max 250 chars).
        city: User's city (max 250 chars).
        birthdate: User's birthdate.
        preferred_language: Preferred language.
        gender: User's gender.
        units: User units (metric, imperial).
        height: User's height in centimeters (1-300).
        max_heart_rate: Maximum heart rate in bpm (30-250).
        first_day_of_week: First day of the week.
        currency: User currency (euro, dollar, pound).
        timezone: IANA timezone the athlete lives in.
    """

    name: StrictStr = Field(
        ...,
        min_length=1,
        max_length=250,
        description="User's full name",
    )
    username: StrictStr = Field(
        ...,
        min_length=1,
        max_length=250,
        pattern=r"^[a-zA-Z0-9._-]+$",
        description="Unique username (alphanumeric, dots, hyphen, underscore)",
    )
    email: EmailStr = Field(
        ...,
        max_length=250,
        description="User's email address",
    )
    city: StrictStr | None = Field(
        default=None,
        max_length=250,
        description="User's city",
    )
    birthdate: datetime_date | None = Field(
        default=None,
        description="User's birthdate",
    )
    preferred_language: Language = Field(
        default=Language.ENGLISH,
        description="Preferred language",
    )
    gender: Gender = Field(
        default=Gender.UNSPECIFIED,
        description="User's gender",
    )
    units: server_settings_schema.Units = Field(
        default=server_settings_schema.Units.METRIC,
        description="User units (metric, imperial)",
    )
    height: StrictInt | None = Field(
        default=None,
        ge=1,
        le=300,
        description="Height in centimeters",
    )
    max_heart_rate: StrictInt | None = Field(
        default=None,
        ge=30,
        le=250,
        description="Maximum heart rate in bpm",
    )
    first_day_of_week: WeekDay = Field(
        default=WeekDay.MONDAY,
        description="First day of the week",
    )
    currency: server_settings_schema.Currency = Field(
        default=server_settings_schema.Currency.EURO,
        description="User currency (euro, dollar, pound)",
    )
    timezone: StrictStr | None = Field(
        default=None,
        max_length=250,
        description=(
            "IANA timezone the athlete lives in. Used as the fallback for activities "
            "imported without a GPS track, which would otherwise inherit the server's "
            "timezone. Null means not set."
        ),
    )

    model_config = ConfigDict(use_enum_values=True)

    @field_validator("timezone")
    @classmethod
    def _validate_timezone(cls, value: str | None) -> str | None:
        """Reject anything that is not a known IANA timezone name.

        The value is stored and later handed to ``ZoneInfo`` on the backend and
        ``Intl.DateTimeFormat`` in the browser, so an unknown name would surface
        as a runtime error far from the setting that caused it.

        Args:
            value: Candidate timezone name, or None.

        Returns:
            The validated timezone name, or None.

        Raises:
            ValueError: When the name is not a known IANA timezone.
        """
        if value is None:
            return None
        if value not in available_timezones():
            raise ValueError(f"Unknown IANA timezone: {value}")
        return value

    @field_validator("birthdate", mode="before")
    @classmethod
    def validate_birthdate(cls, value: datetime_date | str | None) -> str | None:
        """
        Convert birthdate to ISO format string.

        Args:
            value: Birthdate as date object, string, or None.

        Returns:
            ISO format date string or None.
        """
        if value is None:
            return None
        if isinstance(value, datetime_date):
            return value.isoformat()
        return value

validate_birthdate classmethod 

validate_birthdate(value)

Convert birthdate to ISO format string.

Parameters:

Name Type Description Default
value date | str | None

Birthdate as date object, string, or None.

 required

Returns:

Type Description
str | None

ISO format date string or None.
Source code in backend/app/modules/users/users/schema.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
@field_validator("birthdate", mode="before")
@classmethod
def validate_birthdate(cls, value: datetime_date | str | None) -> str | None:
    """
    Convert birthdate to ISO format string.

    Args:
        value: Birthdate as date object, string, or None.

    Returns:
        ISO format date string or None.
    """
    if value is None:
        return None
    if isinstance(value, datetime_date):
        return value.isoformat()
    return value

UsersCreate

Bases: Users

Users schema for admin user creation.

Attributes:

Name Type Description
password StrictStr

User's password (min 8 chars).
Source code in backend/app/modules/users/users/schema.py 
442
443
444
445
446
447
448
449
450
451
452
453
454
455
class UsersCreate(Users):
    """
    Users schema for admin user creation.

    Attributes:
        password: User's password (min 8 chars).
    """

    password: StrictStr = Field(
        ...,
        min_length=8,
        max_length=250,
        description="User's password",
    )

UsersEditPassword

Bases: BaseModel

Schema for password update operations (self-service).

Requires the caller to prove possession of the current password — and an MFA code when MFA is enabled — before the new password is accepted. This prevents a stolen in-memory access token from being parlayed into permanent account takeover.

Attributes:

Name Type Description
current_password StrictStr

Caller's existing password.
password StrictStr

New password (min 8 chars).
mfa_code StrictStr | None

TOTP or backup code, required when MFA is enabled on the account.
Source code in backend/app/modules/users/users/schema.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
class UsersEditPassword(BaseModel):
    """
    Schema for password update operations (self-service).

    Requires the caller to prove possession of the current
    password — and an MFA code when MFA is enabled — before
    the new password is accepted. This prevents a stolen
    in-memory access token from being parlayed into permanent
    account takeover.

    Attributes:
        current_password: Caller's existing password.
        password: New password (min 8 chars).
        mfa_code: TOTP or backup code, required when MFA is
            enabled on the account.
    """

    current_password: StrictStr = Field(
        ...,
        min_length=1,
        max_length=250,
        description="Current password (step-up verification)",
    )
    password: StrictStr = Field(
        ...,
        min_length=8,
        max_length=250,
        description="New password",
    )
    mfa_code: StrictStr | None = Field(
        default=None,
        max_length=32,
        description="TOTP or backup code, required when MFA is enabled",
    )
    revoke_other_sessions: StrictBool = Field(
        default=False,
        description=(
            "When true, revoke all of the user's other sessions "
            "(keeping the current one) after the password change. "
            "Use this to evict an attacker when changing a password "
            "because of a suspected compromise."
        ),
    )

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

UsersListResponse

Bases: BaseModel

Response model for paginated user listing.

Attributes:

Name Type Description
total StrictInt

Total number of user records.
num_records StrictInt | None

Number of records in this response.
page_number StrictInt | None

Current page number.
records list[UsersRead]

List of user records.
Source code in backend/app/modules/users/users/schema.py 
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
class UsersListResponse(BaseModel):
    """
    Response model for paginated user listing.

    Attributes:
        total: Total number of user records.
        num_records: Number of records in this response.
        page_number: Current page number.
        records: List of user records.
    """

    total: StrictInt = Field(
        ...,
        ge=0,
        description="Total number of user records",
    )
    num_records: StrictInt | None = Field(
        default=None,
        ge=0,
        description="Number of records in this response",
    )
    page_number: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Current page number",
    )
    records: list[UsersRead] = Field(
        ...,
        description="List of user records",
    )

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

UsersMe

Bases: UsersRead

Extended users schema for current user profile.

Includes privacy settings and integration status.

Attributes:

Name Type Description
is_strava_linked StrictInt | None

Strava integration status.
is_garminconnect_linked StrictInt | None

Garmin Connect status.
default_activity_visibility StrictStr | None

Default visibility level.
hide_activity_start_time StrictBool | None

Hide start time setting.
hide_activity_location StrictBool | None

Hide location setting.
hide_activity_map StrictBool | None

Hide map setting.
hide_activity_hr StrictBool | None

Hide heart rate setting.
hide_activity_power StrictBool | None

Hide power setting.
hide_activity_cadence StrictBool | None

Hide cadence setting.
hide_activity_elevation StrictBool | None

Hide elevation setting.
hide_activity_speed StrictBool | None

Hide speed setting.
hide_activity_pace StrictBool | None

Hide pace setting.
hide_activity_laps StrictBool | None

Hide laps setting.
hide_activity_workout_sets_steps StrictBool | None

Hide workout sets/steps.
hide_activity_gear StrictBool | None

Hide gear setting.
has_local_password StrictBool | None

Whether the account has a local password (False for SSO-only accounts).
Source code in backend/app/modules/users/users/schema.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
class UsersMe(UsersRead):
    """
    Extended users schema for current user profile.

    Includes privacy settings and integration status.

    Attributes:
        is_strava_linked: Strava integration status.
        is_garminconnect_linked: Garmin Connect status.
        default_activity_visibility: Default visibility level.
        hide_activity_start_time: Hide start time setting.
        hide_activity_location: Hide location setting.
        hide_activity_map: Hide map setting.
        hide_activity_hr: Hide heart rate setting.
        hide_activity_power: Hide power setting.
        hide_activity_cadence: Hide cadence setting.
        hide_activity_elevation: Hide elevation setting.
        hide_activity_speed: Hide speed setting.
        hide_activity_pace: Hide pace setting.
        hide_activity_laps: Hide laps setting.
        hide_activity_workout_sets_steps: Hide workout
            sets/steps.
        hide_activity_gear: Hide gear setting.
        has_local_password: Whether the account has a local
            password (False for SSO-only accounts).
    """

    is_strava_linked: StrictInt | None = Field(default=None, description="Whether Strava is linked")
    is_garminconnect_linked: StrictInt | None = Field(default=None, description="Whether Garmin Connect is linked")
    default_activity_visibility: StrictStr | None = Field(default=None, description="Default activity visibility")
    hide_activity_start_time: StrictBool | None = Field(default=None, description="Hide activity start time")
    hide_activity_location: StrictBool | None = Field(default=None, description="Hide activity location")
    hide_activity_map: StrictBool | None = Field(default=None, description="Hide activity map")
    hide_activity_hr: StrictBool | None = Field(default=None, description="Hide activity heart rate")
    hide_activity_power: StrictBool | None = Field(default=None, description="Hide activity power")
    hide_activity_cadence: StrictBool | None = Field(default=None, description="Hide activity cadence")
    hide_activity_elevation: StrictBool | None = Field(default=None, description="Hide activity elevation")
    hide_activity_speed: StrictBool | None = Field(default=None, description="Hide activity speed")
    hide_activity_pace: StrictBool | None = Field(default=None, description="Hide activity pace")
    hide_activity_laps: StrictBool | None = Field(default=None, description="Hide activity laps")
    hide_activity_workout_sets_steps: StrictBool | None = Field(
        default=None, description="Hide activity workout sets and steps"
    )
    hide_activity_gear: StrictBool | None = Field(default=None, description="Hide activity gear")
    has_local_password: StrictBool | None = Field(
        default=None,
        description=(
            "Whether the account has a local password set. False"
            " indicates an SSO-only account, in which case"
            " step-up flows must skip the password factor. The"
            " raw password hash is never exposed; only this"
            " derived boolean is returned."
        ),
    )

UsersRead

Bases: Users

Users schema for read operations.

Attributes:

Name Type Description
id StrictInt

User ID.
external_auth_count StrictInt

Number of external auth providers.
Source code in backend/app/modules/users/users/schema.py 
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
class UsersRead(Users):
    """
    Users schema for read operations.

    Attributes:
        id: User ID.
        external_auth_count: Number of external auth providers.
    """

    id: StrictInt = Field(
        ...,
        ge=1,
        description="User ID",
    )
    external_auth_count: StrictInt = Field(
        default=0,
        ge=0,
        description="Number of external auth providers linked",
    )

UsersSignup

Bases: UsersBase

Users schema for signup operations.

Attributes:

Name Type Description
password StrictStr

User's password (min 8 chars).
Source code in backend/app/modules/users/users/schema.py 
426
427
428
429
430
431
432
433
434
435
436
437
438
439
class UsersSignup(UsersBase):
    """
    Users schema for signup operations.

    Attributes:
        password: User's password (min 8 chars).
    """

    password: StrictStr = Field(
        ...,
        min_length=8,
        max_length=250,
        description="User's password",
    )

WeekDay

Bases: Enum

Days of the week enumeration.

Attributes:

Name Type Description
SUNDAY

Sunday.
MONDAY

Monday.
TUESDAY

Tuesday.
WEDNESDAY

Wednesday.
THURSDAY

Thursday.
FRIDAY

Friday.
SATURDAY

Saturday.
Source code in backend/app/modules/users/users/schema.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
class WeekDay(Enum):
    """
    Days of the week enumeration.

    Attributes:
        SUNDAY: Sunday.
        MONDAY: Monday.
        TUESDAY: Tuesday.
        WEDNESDAY: Wednesday.
        THURSDAY: Thursday.
        FRIDAY: Friday.
        SATURDAY: Saturday.
    """

    SUNDAY = "sunday"
    MONDAY = "monday"
    TUESDAY = "tuesday"
    WEDNESDAY = "wednesday"
    THURSDAY = "thursday"
    FRIDAY = "friday"
    SATURDAY = "saturday"

approve_user 

approve_user(user_id, db)

Approve a user by marking them as active.

Parameters:

Name Type Description Default
user_id int

ID of user to approve.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 if user not found.
HTTPException

400 if user email not verified.
HTTPException

500 if database error occurs.
Source code in backend/app/modules/users/users/crud.py 
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
@core_decorators.handle_db_errors
def approve_user(user_id: int, db: Session) -> None:
    """
    Approve a user by marking them as active.

    Args:
        user_id: ID of user to approve.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 400 if user email not verified.
        HTTPException: 500 if database error occurs.
    """
    # Get the user from the database
    db_users = _get_user_model_by_id_or_404(user_id, db)

    if not db_users.email_verified:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="User email is not verified",
        )

    db_users.pending_admin_approval = False
    db_users.active = True

    # Commit the transaction
    db.commit()
    db.refresh(db_users)

check_user_is_active 

check_user_is_active(user)

Check if user is active and raise 403 if inactive.

Parameters:

Name Type Description Default
user UsersRead

User object to check (UsersRead schema).

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

403 if user is not active.
Source code in backend/app/modules/users/users/utils.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def check_user_is_active(
    user: users_schema.UsersRead,
) -> None:
    """
    Check if user is active and raise 403 if inactive.

    Args:
        user: User object to check (UsersRead schema).

    Returns:
        None

    Raises:
        HTTPException: 403 if user is not active.
    """
    if not user.active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Inactive user",
            headers={"WWW-Authenticate": "Bearer"},
        )

create_signup_user 

create_signup_user(user, server_settings, identity_service, db, persist_credential=True)

Create a new user during signup process.

Parameters:

Name Type Description Default
user UsersSignup

User signup data.

 required
server_settings ServerSettingsRead

Server config for signup requirements.

 required
identity_service IdentityService

Identity service dependency.

 required
db Session

SQLAlchemy database session.

 required
persist_credential bool

When True (default), validate the supplied password and store its hash in the auth-owned credential table. SSO-created accounts pass False so they get no local credential row and remain SSO-only (has_local_password then correctly reports False).

 True

Returns:

Type Description
UsersRead

Created user schema.

Raises:

Type Description
HTTPException

409 if email/username already exists. Abstract message to reduce information leakage.
HTTPException

500 if database error occurs.
Source code in backend/app/modules/users/users/crud.py 
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
@core_decorators.handle_db_errors
def create_signup_user(
    user: users_schema.UsersSignup,
    server_settings: server_settings_schema.ServerSettingsRead,
    identity_service: "auth_identity_service.IdentityService",
    db: Session,
    persist_credential: bool = True,
) -> users_schema.UsersRead:
    """
    Create a new user during signup process.

    Args:
        user: User signup data.
        server_settings: Server config for signup requirements.
        identity_service: Identity service dependency.
        db: SQLAlchemy database session.
        persist_credential: When ``True`` (default), validate the supplied
            password and store its hash in the auth-owned credential table.
            SSO-created accounts pass ``False`` so they get no local
            credential row and remain SSO-only (``has_local_password`` then
            correctly reports ``False``).

    Returns:
        Created user schema.

    Raises:
        HTTPException: 409 if email/username already exists. Abstract message
            to reduce information leakage.
        HTTPException: 500 if database error occurs.
    """
    try:
        # Determine user status based on server settings
        active = True
        email_verified = False
        pending_admin_approval = False

        if server_settings.signup_require_email_verification:
            email_verified = False
            active = False  # Inactive until email verified

        if server_settings.signup_require_admin_approval:
            pending_admin_approval = True
            active = False  # Inactive until approved

        # If both email verification and admin approval are disabled, user is immediately active
        if not server_settings.signup_require_email_verification and not server_settings.signup_require_admin_approval:
            active = True
            email_verified = True

        # Create a new user
        db_users = users_models.Users(
            **user.model_dump(
                exclude={
                    "username",
                    "email",
                    "access_type",
                    "active",
                    "email_verified",
                    "pending_admin_approval",
                    "password",
                }
            ),
            username=user.username.lower(),
            email=user.email.lower(),
            access_type=users_schema.UserAccessType.REGULAR.value,
            active=active,
            email_verified=email_verified,
            pending_admin_approval=pending_admin_approval,
        )

        # Hash the signup password with the configured policy. SSO-created
        # accounts opt out (persist_credential=False) so the supplied
        # placeholder password is never validated or hashed.
        hashed_password: str | None = None
        if persist_credential:
            hashed_password = auth_password_policy.validate_and_hash_for_user(
                identity_service,
                server_settings,
                users_schema.UserAccessType.REGULAR.value,
                user.password,
            )

        # Add the user to the database
        db.add(db_users)
        db.commit()
        db.refresh(db_users)

        # Persist the password hash in the auth-owned credential table. Skipped
        # for SSO-only accounts so that ``has_local_password`` stays a true row
        # existence check.
        if hashed_password is not None:
            identity_service.set_local_password_hash(db_users.id, hashed_password)

        # Return user
        return _transform_users(db_users)
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Unable to create user."),
        ) from integrity_error

create_user 

create_user(user, identity_service, db)

Create a new user with hashed password.

Parameters:

Name Type Description Default
user UsersCreate

User creation data with plain text password.

 required
identity_service IdentityService

Identity service dependency.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersRead

Created user schema with hashed password.

Raises:

Type Description
HTTPException

409 if email/username already exists.
HTTPException

500 if database error occurs.
Source code in backend/app/modules/users/users/crud.py 
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
@core_decorators.handle_db_errors
def create_user(
    user: users_schema.UsersCreate,
    identity_service: "auth_identity_service.IdentityService",
    db: Session,
) -> users_schema.UsersRead:
    """
    Create a new user with hashed password.

    Args:
        user: User creation data with plain text password.
        identity_service: Identity service dependency.
        db: SQLAlchemy database session.

    Returns:
        Created user schema with hashed password.

    Raises:
        HTTPException: 409 if email/username already exists.
        HTTPException: 500 if database error occurs.
    """
    try:
        user.username = user.username.lower()
        user.email = user.email.lower()

        # Get server settings to determine password policy
        server_settings = server_settings_utils.get_server_settings_or_404(db)

        # Normalize access_type to string value
        access_type_value = users_schema.normalize_access_type(user.access_type)

        # Hash the password with configurable policy and length
        hashed_password = auth_password_policy.validate_and_hash_for_user(
            identity_service,
            server_settings,
            access_type_value,
            user.password,
        )

        # Create a new user
        db_users = users_models.Users(
            **user.model_dump(exclude={"password", "access_type", "mfa_enabled"}),
            access_type=access_type_value,
        )

        # Add the user to the database
        db.add(db_users)
        db.commit()
        db.refresh(db_users)

        # Persist the password hash in the auth-owned credential table.
        identity_service.set_local_password_hash(db_users.id, hashed_password)

        # Return user
        return _transform_users(db_users)
    except HTTPException:
        # Rollback the transaction
        db.rollback()
        raise
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Duplicate entry error. Check if email and username are unique"),
        ) from integrity_error

create_user_default_data 

create_user_default_data(user_id, identity_service, db)

Create default data for newly created user.

Parameters:

Name Type Description Default
user_id int

ID of user to create default data for.

 required
identity_service IdentityService

Identity service used to initialise the auth-owned MFA row through the auth boundary.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None
Source code in backend/app/modules/users/users/utils.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def create_user_default_data(
    user_id: int,
    identity_service: "auth_identity_service.IdentityService",
    db: Session,
) -> None:
    """
    Create default data for newly created user.

    Args:
        user_id: ID of user to create default data for.
        identity_service: Identity service used to initialise the
            auth-owned MFA row through the auth boundary.
        db: SQLAlchemy database session.

    Returns:
        None
    """
    # Create the user integrations in the database
    user_integrations_crud.create_user_integrations(user_id, db)

    # Create the user privacy settings
    users_privacy_settings_crud.create_user_privacy_settings(user_id, db)

    # Create the user health targets
    health_targets_crud.create_health_targets(user_id, db)

    # Create the user default gear
    user_default_gear_crud.create_user_default_gear(user_id, db)

    # Create the user's MFA row (disabled by default) via the auth boundary.
    identity_service.initialize_user_mfa(user_id)

delete_user async 

delete_user(user_id, db)

Delete a user from the database.

Parameters:

Name Type Description Default
user_id int

ID of user to delete.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 if user not found.
HTTPException

500 if database error occurs.
Source code in backend/app/modules/users/users/crud.py 
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
@core_decorators.handle_db_errors
async def delete_user(user_id: int, db: Session) -> None:
    """
    Delete a user from the database.

    Args:
        user_id: ID of user to delete.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 500 if database error occurs.
    """
    # Delete the user row off the event loop (blocking sync DB calls),
    # keeping the API event loop responsive.
    await run_in_threadpool(_delete_user_row, user_id, db)

    # Delete the user photo in the filesystem
    await users_utils.delete_user_photo_filesystem(user_id)

delete_user_photo_filesystem async 

delete_user_photo_filesystem(user_id)

Delete a user's stored photo blobs.

Parameters:

Name Type Description Default
user_id int

ID of user whose photo files to delete.

 required

Returns:

Type Description
None

None
Source code in backend/app/modules/users/users/utils.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
async def delete_user_photo_filesystem(user_id: int) -> None:
    """
    Delete a user's stored photo blobs.

    Args:
        user_id: ID of user whose photo files to delete.

    Returns:
        None
    """
    storage = platform_runtime.get_active_platform().storage
    for key in storage.list_keys(users_signing.USER_IMAGE_STORAGE_AREA, f"{user_id}."):
        storage.delete(users_signing.USER_IMAGE_STORAGE_AREA, key)

edit_user async 

edit_user(user_id, user, db)

Update an existing user's information.

Note

This dynamic-assignment helper is intended for admin endpoints that legitimately need to set fields like access_type, active, or pending_admin_approval. Self-service profile updates MUST NOT call this — use :func:edit_profile_user instead, which enforces an explicit allow-list and prevents privilege escalation through mass assignment.

Parameters:

Name Type Description Default
user_id int

ID of user to update.

 required
user UsersRead

User data to update with.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersRead

users_schema.UsersRead

Raises:

Type Description
HTTPException

404 if user not found.
HTTPException

409 if email/username conflict.
HTTPException

500 if database error occurs.
Source code in backend/app/modules/users/users/crud.py 
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
@core_decorators.handle_db_errors
async def edit_user(user_id: int, user: users_schema.UsersRead, db: Session) -> users_schema.UsersRead:
    """
    Update an existing user's information.

    Note:
        This dynamic-assignment helper is intended for **admin**
        endpoints that legitimately need to set fields like
        ``access_type``, ``active``, or ``pending_admin_approval``.
        Self-service profile updates MUST NOT call this — use
        :func:`edit_profile_user` instead, which enforces an
        explicit allow-list and prevents privilege escalation
        through mass assignment.

    Args:
        user_id: ID of user to update.
        user: User data to update with.
        db: SQLAlchemy database session.

    Returns:
        users_schema.UsersRead

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 409 if email/username conflict.
        HTTPException: 500 if database error occurs.
    """
    try:
        # Fetch the user off the event loop (blocking DB read).
        db_users = await run_in_threadpool(_get_user_model_by_id_or_404, user_id, db)

        height_before = db_users.height

        user.username = user.username.lower()

        # ``photo_path`` is excluded because the read schema exposes a
        # resolved, signed URL while the column stores a storage key.
        # Echoing the read model back would overwrite the key with its own
        # URL; the photo is owned solely by the upload/delete endpoints.
        user_data = user.model_dump(
            exclude_unset=True,
            exclude={"password", "external_auth_count", "mfa_enabled", "photo_path"},
        )
        # Iterate over the fields and update the db_users dynamically
        for key, value in user_data.items():
            # Skip attributes that are read-only properties on the ORM model
            # with no setter. Response-only fields (e.g. mfa_enabled,
            # external_auth_count) are already excluded above via
            # model_dump, this is a defensive guard for any future
            # computed property added to Users.
            class_attr = getattr(type(db_users), key, None)
            if isinstance(class_attr, property) and class_attr.fset is None:
                continue
            setattr(db_users, key, value)

        # Persist the changes off the event loop. The commit/refresh and the
        # bulk BMI recalculation are blocking SQLAlchemy calls, so run them
        # in a worker thread to keep the API event loop responsive.
        return await run_in_threadpool(
            _persist_user_edits,
            db_users,
            height_before,
            True,
            db,
        )
    except HTTPException:
        raise
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Duplicate entry error. Check if email and username are unique"),
        ) from integrity_error

get_admin_users_or_404 

get_admin_users_or_404(db)

Retrieve all admin users from database or raise 404 error.

Parameters:

Name Type Description Default
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
list[UsersRead]

List of all admin User schemas.

Raises:

Type Description
HTTPException

404 if no admin users found.
Source code in backend/app/modules/users/users/utils.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_admin_users_or_404(db: Session) -> list[users_schema.UsersRead]:
    """
    Retrieve all admin users from database or raise 404 error.

    Args:
        db: SQLAlchemy database session.

    Returns:
        List of all admin User schemas.

    Raises:
        HTTPException: 404 if no admin users found.
    """
    admins: list[users_schema.UsersRead] = users_crud.get_users_admin(db)

    if not admins:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No admin users found",
        )

    return admins

get_all_users 

get_all_users(db)

Retrieve all users from the database.

Parameters:

Name Type Description Default
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
list[UsersRead]

List of all user schemas.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users/crud.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
@core_decorators.handle_db_errors
def get_all_users(db: Session) -> list[users_schema.UsersRead]:
    """
    Retrieve all users from the database.

    Args:
        db: SQLAlchemy database session.

    Returns:
        List of all user schemas.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(users_models.Users)
    users: list[users_models.Users] = list(db.execute(stmt).scalars().all())

    return _transform_users(users)

get_user_by_email 

get_user_by_email(email, db)

Retrieve user by email address.

Parameters:

Name Type Description Default
email str

Email address to search for (case-insensitive).

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersRead | None

User schema if found, None otherwise.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users/crud.py 
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
@core_decorators.handle_db_errors
def get_user_by_email(email: str, db: Session) -> users_schema.UsersRead | None:
    """
    Retrieve user by email address.

    Args:
        email: Email address to search for (case-insensitive).
        db: SQLAlchemy database session.

    Returns:
        User schema if found, None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(users_models.Users).where(users_models.Users.email == email.lower())
    user = db.execute(stmt).scalar_one_or_none()
    return _transform_users(user) if user else None

get_user_by_id 

get_user_by_id(user_id, db, public_check=False)

Retrieve user by ID.

Parameters:

Name Type Description Default
user_id int

User ID to search for.

 required
db Session

SQLAlchemy database session.

 required
public_check bool

If True, only returns user when public sharing is enabled in server settings.

 False

Returns:

Type Description
UsersRead | None

User schema if found (and public sharing enabled if
UsersRead | None

public_check=True), None otherwise.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users/crud.py 
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
@core_decorators.handle_db_errors
def get_user_by_id(user_id: int, db: Session, public_check: bool = False) -> users_schema.UsersRead | None:
    """
    Retrieve user by ID.

    Args:
        user_id: User ID to search for.
        db: SQLAlchemy database session.
        public_check: If True, only returns user when public sharing
                      is enabled in server settings.

    Returns:
        User schema if found (and public sharing enabled if
        public_check=True), None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    if public_check:
        # Check if public sharable links are enabled in server settings
        server_settings = server_settings_utils.get_server_settings_or_404(db)

        # Return None if public sharable links are disabled
        if not server_settings.public_shareable_links or not server_settings.public_shareable_links_user_info:
            return None

    stmt = select(users_models.Users).where(users_models.Users.id == user_id)
    user = db.execute(stmt).scalar_one_or_none()
    return _transform_users(user) if user else None

get_user_by_id_or_404 

get_user_by_id_or_404(user_id, db)

Retrieve user by ID or raise 404 error.

Parameters:

Name Type Description Default
user_id int

User ID to search for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersRead

Users schema (guaranteed non-None).

Raises:

Type Description
HTTPException

404 if user not found.
Source code in backend/app/modules/users/users/utils.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def get_user_by_id_or_404(user_id: int, db: Session) -> users_schema.UsersRead:
    """
    Retrieve user by ID or raise 404 error.

    Args:
        user_id: User ID to search for.
        db: SQLAlchemy database session.

    Returns:
        Users schema (guaranteed non-None).

    Raises:
        HTTPException: 404 if user not found.
    """
    # Get the user from the database
    db_user: users_schema.UsersRead | None = users_crud.get_user_by_id(user_id, db)

    if db_user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return db_user

get_user_by_username 

get_user_by_username(username: str, db: Session, contains: Literal[False] = False) -> users_schema.UsersRead | None

get_user_by_username(username: str, db: Session, contains: Literal[True]) -> list[users_schema.UsersRead]

get_user_by_username(username, db, contains=False)

Retrieve user by username.

Parameters:

Name Type Description Default
username str

Username to search for.

 required
db Session

SQLAlchemy database session.

 required
contains bool

If True, performs partial match search and returns list of matching users. If False, performs exact match and returns single user or None.

 False

Returns:

Type Description
list[UsersRead] | UsersRead | None

If contains=False: User schema if found, None otherwise.
list[UsersRead] | UsersRead | None

If contains=True: List of user schemas matching the search.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users/crud.py 
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
@core_decorators.handle_db_errors
def get_user_by_username(
    username: str, db: Session, contains: bool = False
) -> list[users_schema.UsersRead] | users_schema.UsersRead | None:
    """
    Retrieve user by username.

    Args:
        username: Username to search for.
        db: SQLAlchemy database session.
        contains: If True, performs partial match search and returns
                  list of matching users. If False, performs exact
                  match and returns single user or None.

    Returns:
        If contains=False: User schema if found, None otherwise.
        If contains=True: List of user schemas matching the search.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    # Decode and normalize search term (needed for both exact and partial matches)
    normalized_username = unquote(username).replace("+", " ").lower()

    if contains:
        # Escape LIKE special characters to prevent SQL injection
        escaped_username = normalized_username.replace("\\", "\\\\").replace("%", r"\%").replace("_", r"\_")

        # Query users with username containing the search term
        stmt = select(users_models.Users).where(
            func.lower(users_models.Users.username).like(f"%{escaped_username}%", escape="\\")
        )
        users: list[users_models.Users] = list(db.execute(stmt).scalars().all())
        return _transform_users(users)
    else:
        # Exact match - no LIKE escaping needed
        stmt = select(users_models.Users).where(users_models.Users.username == normalized_username)
        user = db.execute(stmt).scalar_one_or_none()
        return _transform_users(user) if user else None

get_users_admin 

get_users_admin(db)

Retrieve all admin users from the database.

Parameters:

Name Type Description Default
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
list[UsersRead]

List of admin user schemas.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users/crud.py 
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
@core_decorators.handle_db_errors
def get_users_admin(db: Session) -> list[users_schema.UsersRead]:
    """
    Retrieve all admin users from the database.

    Args:
        db: SQLAlchemy database session.

    Returns:
        List of admin user schemas.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(users_models.Users).where(users_models.Users.access_type == users_schema.UserAccessType.ADMIN.value)
    users: list[users_models.Users] = list(db.execute(stmt).scalars().all())
    return _transform_users(users)

get_users_number 

get_users_number(db, show_inactive=True, show_email_unverified=True, show_pending_approval=True)

Count users matching the optional list filters.

Parameters:

Name Type Description Default
db Session

SQLAlchemy database session.

 required
show_inactive bool | None

If False, excludes inactive users. Defaults to True.

 True
show_email_unverified bool | None

If False, excludes users with unverified emails. Defaults to True.

 True
show_pending_approval bool | None

If False, excludes users pending admin approval. Defaults to True.

 True

Returns:

Type Description
int

Number of users matching the filters.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users/crud.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
@core_decorators.handle_db_errors
def get_users_number(
    db: Session,
    show_inactive: bool | None = True,
    show_email_unverified: bool | None = True,
    show_pending_approval: bool | None = True,
) -> int:
    """
    Count users matching the optional list filters.

    Args:
        db: SQLAlchemy database session.
        show_inactive: If False, excludes inactive users.
            Defaults to True.
        show_email_unverified: If False, excludes users with
            unverified emails. Defaults to True.
        show_pending_approval: If False, excludes users pending
            admin approval. Defaults to True.

    Returns:
        Number of users matching the filters.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    stmt = select(func.count(users_models.Users.id))

    if show_inactive is False:
        stmt = stmt.where(users_models.Users.active.is_(True))
    if show_email_unverified is False:
        stmt = stmt.where(users_models.Users.email_verified.is_(True))
    if show_pending_approval is False:
        stmt = stmt.where(users_models.Users.pending_admin_approval.is_(False))

    return db.execute(stmt).scalar_one()

get_users_with_pagination 

get_users_with_pagination(db, page_number=None, num_records=None, show_inactive=True, show_email_unverified=True, show_pending_approval=True)

Retrieve a paginated list of users with optional filtering.

Parameters:

Name Type Description Default
db Session

Database session for executing queries.

 required
page_number int | None

The page number for pagination (1-indexed). If None, pagination is not applied. Defaults to None.

 None
num_records int | None

The number of records per page. If None, pagination is not applied. Defaults to None.

 None
show_inactive bool | None

If False, excludes inactive users. Defaults to True (includes inactive users).

 True
show_email_unverified bool | None

If False, excludes users with unverified emails. Defaults to True (includes email unverified users).

 True
show_pending_approval bool | None

If False, excludes users pending admin approval. Defaults to True (includes pending approval users).

 True

Returns:

Type Description
list[UsersRead]

list[users_schema.UsersRead]: A list of User schemas matching the specified criteria, ordered by username. Returns an empty list if no users match the filters.
Source code in backend/app/modules/users/users/crud.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
@core_decorators.handle_db_errors
def get_users_with_pagination(
    db: Session,
    page_number: int | None = None,
    num_records: int | None = None,
    show_inactive: bool | None = True,
    show_email_unverified: bool | None = True,
    show_pending_approval: bool | None = True,
) -> list[users_schema.UsersRead]:
    """
    Retrieve a paginated list of users with optional filtering.

    Args:
        db (Session): Database session for executing queries.
        page_number (int | None): The page number for pagination (1-indexed).
            If None, pagination is not applied. Defaults to None.
        num_records (int | None): The number of records per page.
            If None, pagination is not applied. Defaults to None.
        show_inactive (bool | None): If False, excludes inactive users.
            Defaults to True (includes inactive users).
        show_email_unverified (bool | None): If False, excludes users with
            unverified emails. Defaults to True (includes email unverified
            users).
        show_pending_approval (bool | None): If False, excludes users pending
            admin approval. Defaults to True (includes pending approval users).

    Returns:
        list[users_schema.UsersRead]: A list of User schemas matching the specified
            criteria, ordered by username. Returns an empty list if no users
            match the filters.
    """
    stmt = select(users_models.Users)

    if show_inactive is False:
        stmt = stmt.where(users_models.Users.active.is_(True))
    if show_email_unverified is False:
        stmt = stmt.where(users_models.Users.email_verified.is_(True))
    if show_pending_approval is False:
        stmt = stmt.where(users_models.Users.pending_admin_approval.is_(False))

    stmt = stmt.order_by(users_models.Users.username)

    if page_number is not None and num_records is not None:
        stmt = stmt.offset((page_number - 1) * num_records).limit(num_records)

    users: list[users_models.Users] = list(db.execute(stmt).scalars().all())
    return _transform_users(users)

save_user_image_file async 

save_user_image_file(user_id, file, db)

Save user image file with security validation and update DB.

Uses centralized file upload handler for validation and async I/O, then updates user photo path in database.

Parameters:

Name Type Description Default
user_id int

ID of user whose image is being saved.

 required
file UploadFile

Uploaded image file (UploadFile).

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
str

Path to saved image file.

Raises:

Type Description
HTTPException

400 if filename or extension is invalid, 413 if too large, 500 if upload fails.
Source code in backend/app/modules/users/users/utils.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
async def save_user_image_file(user_id: int, file: UploadFile, db: Session) -> str:
    """
    Save user image file with security validation and update DB.

    Uses centralized file upload handler for validation and async
    I/O, then updates user photo path in database.

    Args:
        user_id: ID of user whose image is being saved.
        file: Uploaded image file (UploadFile).
        db: SQLAlchemy database session.

    Returns:
        Path to saved image file.

    Raises:
        HTTPException: 400 if filename or extension is invalid,
            413 if too large, 500 if upload fails.
    """
    # Validate the user exists off the event loop (blocking sync DB read).
    # A missing user raises HTTPException 404 inside the thread, which
    # propagates through the await unchanged.
    await run_in_threadpool(get_user_by_id_or_404, user_id, db)

    if not file.filename:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Filename is required",
        )

    # Defense-in-depth allow-list on the user-supplied extension.
    # SafeUploads still validates the magic number afterwards, so a
    # mismatched signature is rejected even if the extension passes.
    _, file_extension = os.path.splitext(file.filename)
    file_extension = file_extension.lower()
    if file_extension not in _ALLOWED_USER_IMAGE_EXTENSIONS:
        raise HTTPException(
            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
            detail="Unsupported user image file type",
        )

    filename: str = f"{user_id}{file_extension}"

    # Validated in memory, then written through the platform StorageProvider so
    # the photo is not pinned to this node's disk and has no public path.
    data = await core_file_uploads.read_validated_upload(file, kind=core_file_uploads.UploadKind.IMAGE)
    platform_runtime.get_active_platform().storage.save(
        users_signing.USER_IMAGE_STORAGE_AREA,
        filename,
        data,
        file.content_type,
    )

    # The row stores the storage key; the read schema resolves it to a URL.
    return str(await users_crud.update_user_photo(user_id, db, filename))

update_user_photo async 

update_user_photo(user_id, db, photo_path=None)

Update a user's photo path.

Parameters:

Name Type Description Default
user_id int

ID of user to update photo for.

 required
db Session

SQLAlchemy database session.

 required
photo_path str | None

New photo path. If None, removes photo.

 None

Returns:

Type Description
str | None

The updated photo path, or None if removed.

Raises:

Type Description
HTTPException

404 if user not found.
HTTPException

500 if database error occurs.
Source code in backend/app/modules/users/users/crud.py 
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
@core_decorators.handle_db_errors
async def update_user_photo(user_id: int, db: Session, photo_path: str | None = None) -> str | None:
    """
    Update a user's photo path.

    Args:
        user_id: ID of user to update photo for.
        db: SQLAlchemy database session.
        photo_path: New photo path. If None, removes photo.

    Returns:
        The updated photo path, or None if removed.

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 500 if database error occurs.
    """
    # Persist the photo path change off the event loop (blocking sync
    # DB calls), keeping the API event loop responsive.
    await run_in_threadpool(_apply_user_photo_update, user_id, photo_path, db)

    if photo_path:
        # Return the photo path
        return photo_path

    # Delete the user photo in the filesystem
    await users_utils.delete_user_photo_filesystem(user_id)

    return None

verify_user_email 

verify_user_email(user_id, server_settings, db)

Verify user email and conditionally activate account.

Parameters:

Name Type Description Default
user_id int

ID of user to verify.

 required
server_settings ServerSettingsRead

Server config determining activation policy.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 if user not found.
HTTPException

500 if database error occurs.
Source code in backend/app/modules/users/users/crud.py 
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
@core_decorators.handle_db_errors
def verify_user_email(
    user_id: int,
    server_settings: server_settings_schema.ServerSettingsRead,
    db: Session,
) -> None:
    """
    Verify user email and conditionally activate account.

    Args:
        user_id: ID of user to verify.
        server_settings: Server config determining activation policy.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 if user not found.
        HTTPException: 500 if database error occurs.
    """
    # Get the user from the database
    db_users = _get_user_model_by_id_or_404(user_id, db)

    db_users.email_verified = True
    if not server_settings.signup_require_admin_approval:
        db_users.pending_admin_approval = False
        db_users.active = True

    # Commit the transaction
    db.commit()
    db.refresh(db_users)

Users profile

Profile module for user profile management and operations.

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

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

ActivityLimitError

Bases: ProfileImportError

Raised when import contains too many activities.

Source code in backend/app/modules/users/users_profile/exceptions.py 
136
137
138
139
140
141
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/modules/users/users_profile/exceptions.py 
71
72
73
74
75
76
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/modules/users/users_profile/exceptions.py 
104
105
106
107
108
109
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/modules/users/users_profile/exceptions.py 
39
40
41
42
43
44
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/modules/users/users_profile/exceptions.py 
120
121
122
123
124
125
class DiskSpaceError(ProfileImportError):
    """
    Raised when insufficient disk space is available.
    """

    pass

ExportError

Bases: ProfileOperationError

Base exception for export operations.

Source code in backend/app/modules/users/users_profile/exceptions.py 
22
23
24
25
26
27
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/modules/users/users_profile/export_service.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
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 = users_profile_utils.initialize_operation_counts(include_user_count=True)
        self.performance_config: ExportPerformanceConfig = (
            performance_config or ExportPerformanceConfig.get_auto_config()
        )

        logger.info(
            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}"
        )

    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:
            users_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

            logger.info(f"Starting batched activity collection with batch_size={batch_size}")

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

                if not batch_activities:
                    break

                all_activities.extend(batch_activities)
                offset += batch_size

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

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

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

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

            logger.info(f"Written {len(activities_dicts)} activities to ZIP")

            # 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:
                logger.warning(f"No valid activity IDs found for user {self.user_id}")
            else:
                self._collect_and_write_activity_components(zipf, activity_ids, all_activities)

            self._collect_and_write_global_activity_components(zipf)

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

        return all_activities

    def _collect_and_write_global_activity_components(self, zipf: zipfile.ZipFile) -> None:
        """Collect and write file-global activity package records once."""
        for contributor in activities_contributor_registry.profile_global_contributors():
            try:
                records = contributor.export(self.db)
                record_dicts = [users_profile_utils.sqlalchemy_obj_to_dict(record) for record in records]
                users_profile_utils.write_json_to_zip(zipf, contributor.archive_path, record_dicts, self.counts)
                self.counts[contributor.count_key] = len(record_dicts)
            except Exception as err:
                logger.error(f"Failed to collect {contributor.key}: {err}", exc_info=err)
                raise DataCollectionError(f"Failed to collect {contributor.key}: {err}") from err

    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

            return activity_integration.list_user_activities_page(
                self.user_id,
                self.db,
                page_number=page_number,
                num_records=limit,
            )

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

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

        Args:
            zipf: ZipFile instance to write to.
            activity_ids: List of activity IDs to process.
            user_activities: List of activity objects.
        """
        if not activity_ids or not user_activities:
            logger.warning("No activity IDs or activities provided for component collection")
            return
        # Process activity IDs in smaller batches to reduce memory usage
        batch_size = self.performance_config.batch_size // 2  # Smaller batches for components

        for contributor in activities_contributor_registry.profile_activity_contributors():
            # For large splittable components, write in chunks during collection
            if contributor.split:
                self._collect_and_write_component_chunked(
                    zipf,
                    contributor.count_key,
                    contributor.archive_path,
                    contributor.export,
                    activity_ids,
                    user_activities,
                    batch_size,
                )
            else:
                # For small components, collect all then write
                self._collect_and_write_component_simple(
                    zipf,
                    contributor.count_key,
                    contributor.archive_path,
                    contributor.export,
                    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]

            users_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.db)
                if data:
                    # Convert to dicts and add to chunk buffer
                    batch_dicts = [users_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}"

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

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

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

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

        # Write remaining data in buffer
        if chunk_buffer:
            if file_counter == 0:
                # Only one chunk, use original filename
                users_profile_utils.write_json_to_zip(zipf, base_filename, chunk_buffer, self.counts)
                logger.info(f"Written {len(chunk_buffer)} {component_key} items to single file")
            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}"

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

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

    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]

            users_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.db)
                if data:
                    all_component_data.extend(data)
            except Exception as err:
                logger.error(f"Failed to collect batch for {component_key}: {err}", exc_info=err)
                raise DataCollectionError(f"Failed to collect {component_key}: {err}") from err

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

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

    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 = [users_profile_utils.sqlalchemy_obj_to_dict(g) for g in gears]
                    users_profile_utils.write_json_to_zip(zipf, "data/gears.json", gears_dicts, self.counts)
                else:
                    users_profile_utils.write_json_to_zip(zipf, "data/gears.json", [], self.counts)
            except Exception as err:
                logger.error(f"Failed to collect gears: {err}", exc_info=err)
                raise DataCollectionError(f"Failed to collect gears: {err}") from err

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

        except SQLAlchemyError as err:
            logger.error(f"Database error collecting gear data: {err}", exc_info=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 = [users_profile_utils.sqlalchemy_obj_to_dict(hd) for hd in health_weight]
                    users_profile_utils.write_json_to_zip(
                        zipf,
                        "data/health_weight.json",
                        health_weight_dicts,
                        self.counts,
                    )
                else:
                    users_profile_utils.write_json_to_zip(zipf, "data/health_weight.json", [], self.counts)
            except Exception as err:
                logger.error(f"Failed to collect health data: {err}", exc_info=err)
                raise DataCollectionError(f"Failed to collect health data: {err}") from err

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

        except SQLAlchemyError as err:
            logger.error(f"Database error collecting health data: {err}", exc_info=err)
            raise DatabaseConnectionError(f"Failed to collect health data: {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_schema.UsersDefaultGearRead | None = (
                    user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)
                )
                if user_default_gear:
                    default_gear_dict = [user_default_gear.model_dump(mode="json")]
                    users_profile_utils.write_json_to_zip(
                        zipf,
                        "data/user_default_gear.json",
                        default_gear_dict,
                        self.counts,
                    )
                else:
                    users_profile_utils.write_json_to_zip(zipf, "data/user_default_gear.json", [], self.counts)
            except Exception as err:
                logger.error(f"Failed to collect user default gear: {err}", exc_info=err)
                raise DataCollectionError(f"Failed to collect user default gear: {err}") from err

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

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

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

        except SQLAlchemyError as err:
            logger.error(f"Database error collecting user settings: {err}", exc_info=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.

        Reads each activity's retained source file through the platform
        ``StorageProvider`` (local disk or object storage) and writes it into the
        archive under ``activity_files/{id}{ext}``.

        Args:
            zipf: ZipFile instance to write to.
            user_activities: List of activity objects.
        """
        if not user_activities:
            return

        storage = platform_runtime.get_active_platform().storage
        for activity in user_activities:
            try:
                stored = activity_file_storage_integration.get_activity_file(activity.id, storage)
                if stored is None:
                    continue
                key, data = stored
                zipf.writestr(os.path.join("activity_files", key), data)
                self.counts["activity_files"] += 1
            except MemoryAllocationError:
                raise
            except OSError as err:
                logger.warning(f"Failed to add activity file for activity {activity.id}: {err}", exc_info=err)
                continue
            except Exception as err:
                logger.warning(f"Unexpected error adding activity file for activity {activity.id}: {err}", exc_info=err)
                continue

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

        Read through the platform ``StorageProvider`` rather than by walking a
        local directory, so an export is complete regardless of which storage
        backend is configured. Keys are ``{activity_id}_{uuid}{ext}``, so each
        activity's blobs are listed by its own prefix instead of scanning every
        stored file and filtering.

        Args:
            zipf: ZipFile instance to write to.
            user_activities: List of activity objects.
        """
        if not user_activities:
            return

        storage = platform_runtime.get_active_platform().storage

        for activity in user_activities:
            try:
                blobs = activity_media_integration.list_media_blobs(activity.id, storage)
            except Exception as err:
                logger.warning(f"Failed to list media for activity {activity.id}: {err}", exc_info=err)
                continue

            for key, data in blobs:
                try:
                    zipf.writestr(os.path.join("activity_media", key), data)
                    self.counts["media"] += 1
                except MemoryAllocationError:
                    raise
                except Exception as err:
                    logger.warning(f"Unexpected error adding media file {key}: {err}", exc_info=err)
                    continue

    def add_user_images_to_zip(self, zipf: zipfile.ZipFile):
        """
        Add the user's profile image to the ZIP archive.

        Read through the platform ``StorageProvider`` rather than by walking a
        local directory, so an export is complete regardless of which storage
        backend is configured. Keys are ``{user_id}{ext}``, so the caller's
        blobs are listed by their own prefix instead of scanning every stored
        image and filtering by filename.

        Args:
            zipf: ZipFile instance to write to.
        """
        storage = platform_runtime.get_active_platform().storage
        area = users_signing.USER_IMAGE_STORAGE_AREA

        try:
            keys = storage.list_keys(area, f"{self.user_id}.")
        except Exception as err:
            logger.warning(f"Failed to list images for user {self.user_id}: {err}", exc_info=err)
            return

        for key in keys:
            try:
                data = storage.get(area, key)
                if data is None:
                    logger.warning(f"User image blob not found for key: {key}")
                    continue

                size = len(data)
                if size > 10 * 1024 * 1024:
                    logger.warning(f"Large image file: {key} ({size / (1024 * 1024):.1f}MB)")
                if size > 5 * 1024 * 1024:
                    users_profile_utils.check_memory_usage(
                        f"before image {key}",
                        self.performance_config.max_memory_mb,
                        self.performance_config.enable_memory_monitoring,
                    )

                zipf.writestr(os.path.join("user_images", key), data)
                self.counts["user_images"] += 1
            except MemoryAllocationError:
                raise
            except Exception as err:
                logger.warning(f"Unexpected error adding user image {key}: {err}", exc_info=err)

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

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

        Yields:
            Chunks of ZIP archive as bytes.

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

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

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

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

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

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

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

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

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

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

                        users_profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                        logger.info("Adding user images to archive...")
                        self.add_user_images_to_zip(zipf)

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

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

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

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

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

                # Stream the file with error handling
                tmp.seek(0)
                chunk_count = 0
                while True:
                    try:
                        users_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:
                        logger.error(f"Memory error during streaming: {err}", exc_info=err)
                        raise MemoryAllocationError(f"Insufficient memory to stream export: {err}") from err

                logger.info(f"Successfully streamed {chunk_count} chunks for user {self.user_id}")
        except MemoryAllocationError as err:
            raise err
        except OSError as err:
            logger.error(f"File system error during export: {err}", exc_info=err)
            raise FileSystemError(f"File system error during export: {err}") from err
        except MemoryError as err:
            logger.error(f"Memory allocation error during export: {err}", exc_info=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.

Reads each activity's retained source file through the platform StorageProvider (local disk or object storage) and writes it into the archive under activity_files/{id}{ext}.

Parameters:

Name Type Description Default
zipf ZipFile

ZipFile instance to write to.

 required
user_activities list[Any]

List of activity objects.

 required
Source code in backend/app/modules/users/users_profile/export_service.py 
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
def add_activity_files_to_zip(self, zipf: zipfile.ZipFile, user_activities: list[Any]):
    """
    Add activity files to ZIP archive.

    Reads each activity's retained source file through the platform
    ``StorageProvider`` (local disk or object storage) and writes it into the
    archive under ``activity_files/{id}{ext}``.

    Args:
        zipf: ZipFile instance to write to.
        user_activities: List of activity objects.
    """
    if not user_activities:
        return

    storage = platform_runtime.get_active_platform().storage
    for activity in user_activities:
        try:
            stored = activity_file_storage_integration.get_activity_file(activity.id, storage)
            if stored is None:
                continue
            key, data = stored
            zipf.writestr(os.path.join("activity_files", key), data)
            self.counts["activity_files"] += 1
        except MemoryAllocationError:
            raise
        except OSError as err:
            logger.warning(f"Failed to add activity file for activity {activity.id}: {err}", exc_info=err)
            continue
        except Exception as err:
            logger.warning(f"Unexpected error adding activity file for activity {activity.id}: {err}", exc_info=err)
            continue

add_activity_media_to_zip 

add_activity_media_to_zip(zipf, user_activities)

Add activity media files to ZIP archive.

Read through the platform StorageProvider rather than by walking a local directory, so an export is complete regardless of which storage backend is configured. Keys are {activity_id}_{uuid}{ext}, so each activity's blobs are listed by its own prefix instead of scanning every stored file and filtering.

Parameters:

Name Type Description Default
zipf ZipFile

ZipFile instance to write to.

 required
user_activities list[Any]

List of activity objects.

 required
Source code in backend/app/modules/users/users_profile/export_service.py 
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
def add_activity_media_to_zip(self, zipf: zipfile.ZipFile, user_activities: list[Any]):
    """
    Add activity media files to ZIP archive.

    Read through the platform ``StorageProvider`` rather than by walking a
    local directory, so an export is complete regardless of which storage
    backend is configured. Keys are ``{activity_id}_{uuid}{ext}``, so each
    activity's blobs are listed by its own prefix instead of scanning every
    stored file and filtering.

    Args:
        zipf: ZipFile instance to write to.
        user_activities: List of activity objects.
    """
    if not user_activities:
        return

    storage = platform_runtime.get_active_platform().storage

    for activity in user_activities:
        try:
            blobs = activity_media_integration.list_media_blobs(activity.id, storage)
        except Exception as err:
            logger.warning(f"Failed to list media for activity {activity.id}: {err}", exc_info=err)
            continue

        for key, data in blobs:
            try:
                zipf.writestr(os.path.join("activity_media", key), data)
                self.counts["media"] += 1
            except MemoryAllocationError:
                raise
            except Exception as err:
                logger.warning(f"Unexpected error adding media file {key}: {err}", exc_info=err)
                continue

add_user_images_to_zip 

add_user_images_to_zip(zipf)

Add the user's profile image to the ZIP archive.

Read through the platform StorageProvider rather than by walking a local directory, so an export is complete regardless of which storage backend is configured. Keys are {user_id}{ext}, so the caller's blobs are listed by their own prefix instead of scanning every stored image and filtering by filename.

Parameters:

Name Type Description Default
zipf ZipFile

ZipFile instance to write to.

 required
Source code in backend/app/modules/users/users_profile/export_service.py 
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
def add_user_images_to_zip(self, zipf: zipfile.ZipFile):
    """
    Add the user's profile image to the ZIP archive.

    Read through the platform ``StorageProvider`` rather than by walking a
    local directory, so an export is complete regardless of which storage
    backend is configured. Keys are ``{user_id}{ext}``, so the caller's
    blobs are listed by their own prefix instead of scanning every stored
    image and filtering by filename.

    Args:
        zipf: ZipFile instance to write to.
    """
    storage = platform_runtime.get_active_platform().storage
    area = users_signing.USER_IMAGE_STORAGE_AREA

    try:
        keys = storage.list_keys(area, f"{self.user_id}.")
    except Exception as err:
        logger.warning(f"Failed to list images for user {self.user_id}: {err}", exc_info=err)
        return

    for key in keys:
        try:
            data = storage.get(area, key)
            if data is None:
                logger.warning(f"User image blob not found for key: {key}")
                continue

            size = len(data)
            if size > 10 * 1024 * 1024:
                logger.warning(f"Large image file: {key} ({size / (1024 * 1024):.1f}MB)")
            if size > 5 * 1024 * 1024:
                users_profile_utils.check_memory_usage(
                    f"before image {key}",
                    self.performance_config.max_memory_mb,
                    self.performance_config.enable_memory_monitoring,
                )

            zipf.writestr(os.path.join("user_images", key), data)
            self.counts["user_images"] += 1
        except MemoryAllocationError:
            raise
        except Exception as err:
            logger.warning(f"Unexpected error adding user image {key}: {err}", exc_info=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/modules/users/users_profile/export_service.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
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 = [users_profile_utils.sqlalchemy_obj_to_dict(g) for g in gears]
                users_profile_utils.write_json_to_zip(zipf, "data/gears.json", gears_dicts, self.counts)
            else:
                users_profile_utils.write_json_to_zip(zipf, "data/gears.json", [], self.counts)
        except Exception as err:
            logger.error(f"Failed to collect gears: {err}", exc_info=err)
            raise DataCollectionError(f"Failed to collect gears: {err}") from err

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

    except SQLAlchemyError as err:
        logger.error(f"Database error collecting gear data: {err}", exc_info=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/modules/users/users_profile/export_service.py 
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
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 = [users_profile_utils.sqlalchemy_obj_to_dict(hd) for hd in health_weight]
                users_profile_utils.write_json_to_zip(
                    zipf,
                    "data/health_weight.json",
                    health_weight_dicts,
                    self.counts,
                )
            else:
                users_profile_utils.write_json_to_zip(zipf, "data/health_weight.json", [], self.counts)
        except Exception as err:
            logger.error(f"Failed to collect health data: {err}", exc_info=err)
            raise DataCollectionError(f"Failed to collect health data: {err}") from err

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

    except SQLAlchemyError as err:
        logger.error(f"Database error collecting health data: {err}", exc_info=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/modules/users/users_profile/export_service.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
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:
        users_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

        logger.info(f"Starting batched activity collection with batch_size={batch_size}")

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

            if not batch_activities:
                break

            all_activities.extend(batch_activities)
            offset += batch_size

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

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

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

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

        logger.info(f"Written {len(activities_dicts)} activities to ZIP")

        # 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:
            logger.warning(f"No valid activity IDs found for user {self.user_id}")
        else:
            self._collect_and_write_activity_components(zipf, activity_ids, all_activities)

        self._collect_and_write_global_activity_components(zipf)

    except SQLAlchemyError as err:
        logger.error(f"Database error collecting activities: {err}", exc_info=err)
        raise DatabaseConnectionError(f"Failed to collect activity data: {err}") from err
    except MemoryAllocationError as err:
        logger.error(f"Memory limit exceeded while collecting activities: {err}. ", exc_info=err)
        raise err
    except DataCollectionError:
        raise
    except Exception as err:
        logger.error(f"Unexpected error collecting activities: {err}", exc_info=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/modules/users/users_profile/export_service.py 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
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_schema.UsersDefaultGearRead | None = (
                user_default_gear_crud.get_user_default_gear_by_user_id(self.user_id, self.db)
            )
            if user_default_gear:
                default_gear_dict = [user_default_gear.model_dump(mode="json")]
                users_profile_utils.write_json_to_zip(
                    zipf,
                    "data/user_default_gear.json",
                    default_gear_dict,
                    self.counts,
                )
            else:
                users_profile_utils.write_json_to_zip(zipf, "data/user_default_gear.json", [], self.counts)
        except Exception as err:
            logger.error(f"Failed to collect user default gear: {err}", exc_info=err)
            raise DataCollectionError(f"Failed to collect user default gear: {err}") from err

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

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

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

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

generate_export_archive 

generate_export_archive(user_dict, timeout_seconds=300)

Generate and stream export archive as bytes.

Parameters:

Name Type Description Default
user_dict dict[str, Any]

User data dictionary to export.

 required
timeout_seconds int | None

Optional timeout in seconds.

 300

Yields:

Type Description
Generator[bytes]

Chunks of ZIP archive as bytes.

Raises:

Type Description
ExportTimeoutError

If operation times out.
ZipCreationError

If ZIP creation fails.
MemoryAllocationError

If memory limit exceeded.
FileSystemError

If file system error occurs.
Source code in backend/app/modules/users/users_profile/export_service.py 
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
def generate_export_archive(self, user_dict: dict[str, Any], timeout_seconds: int | None = 300) -> Generator[bytes]:
    """
    Generate and stream export archive as bytes.

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

    Yields:
        Chunks of ZIP archive as bytes.

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

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

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

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

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

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

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

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

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

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

                    users_profile_utils.check_timeout(timeout_seconds, start_time, ExportTimeoutError, "Export")
                    logger.info("Adding user images to archive...")
                    self.add_user_images_to_zip(zipf)

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

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

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

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

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

            # Stream the file with error handling
            tmp.seek(0)
            chunk_count = 0
            while True:
                try:
                    users_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:
                    logger.error(f"Memory error during streaming: {err}", exc_info=err)
                    raise MemoryAllocationError(f"Insufficient memory to stream export: {err}") from err

            logger.info(f"Successfully streamed {chunk_count} chunks for user {self.user_id}")
    except MemoryAllocationError as err:
        raise err
    except OSError as err:
        logger.error(f"File system error during export: {err}", exc_info=err)
        raise FileSystemError(f"File system error during export: {err}") from err
    except MemoryError as err:
        logger.error(f"Memory allocation error during export: {err}", exc_info=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/modules/users/users_profile/exceptions.py 
79
80
81
82
83
84
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/modules/users/users_profile/exceptions.py 
 96
 97
 98
 99
100
101
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/modules/users/users_profile/exceptions.py 
128
129
130
131
132
133
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/modules/users/users_profile/exceptions.py 
47
48
49
50
51
52
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/modules/users/users_profile/import_service.py 
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
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()
        )

        logger.info(
            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}"
        )

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

        Args:
            zip_data: ZIP file content as bytes.

        Returns:
            Dictionary with import results and counts.

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

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

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

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

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

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

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

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

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

                self._restore_global_activity_components(zipf)

                # 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 health data
                profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
                health_weight_data = self._load_single_json(zipf, "data/health_weight.json")
                health_targets_data = self._load_single_json(zipf, "data/health_targets.json")

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

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

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

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

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

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

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

        Returns:
            Parsed JSON data as list.

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

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

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

            logger.debug(f"Loaded {len(data)} items from {filename}")

            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}"
            logger.error(error_msg)
            raise JSONParseError(error_msg) from err

    def _restore_global_activity_components(self, zipf: zipfile.ZipFile) -> None:
        """Load and restore every global activity contribution once."""
        for contributor in activities_contributor_registry.profile_global_contributors():
            records = self._load_single_json(zipf, contributor.archive_path, check_memory=False)
            restored = contributor.restore(records, self.db)
            self.counts[contributor.count_key] += restored
        # The restore functions stage; the importer owns the commit.
        self.db.commit()

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

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

        Returns:
            Entry bytes.

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

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

        Args:
            gears_data: List of gear data dictionaries.

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

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

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

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

        logger.info(f"Imported {self.counts['gears']} gears")
        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:
            logger.info("No gear components data to import")
            return

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

            gear_component_data.pop("id", None)

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

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

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

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

            # The archive carries the exporter's photo key (or, for
            # archives predating storage keys, an absolute filesystem path).
            # Either way only the extension is trusted: the key is rebuilt for
            # the importing user so a crafted ZIP cannot repoint the avatar at
            # somebody else's blob.
            photo_path = user_profile.get("photo_path")
            extension = os.path.splitext(photo_path)[1].lstrip(".").lower() if isinstance(photo_path, str) else ""
            if extension in _IMPORTABLE_IMAGE_EXTENSIONS:
                user_profile["photo_path"] = f"{self.user_id}.{extension}"
            else:
                user_profile["photo_path"] = None

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

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

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

        Args:
            user_default_gear_data: Default gear data.
            gears_id_mapping: Mapping of old to new gear IDs.
        """
        if not user_default_gear_data:
            logger.info("No user default gear data to import")
            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:
            logger.info("No existing user default gear, creating new record")
            current_user_default_gear = user_default_gear_crud.create_user_default_gear(self.user_id, self.db)

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

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

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

        user_default_gear = user_default_gear_schema.UsersDefaultGearUpdate(**gear_data)
        user_default_gear_crud.edit_user_default_gear(user_default_gear, self.user_id, self.db)
        logger.info("Imported user default gear")
        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:
            logger.info("No user integrations data to import")
            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)
        logger.info("Imported user integrations")
        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:
            logger.info("No user goals data to import")
            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

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

    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:
            logger.info("No user privacy settings data to import")
            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)
        logger.info("Imported user privacy settings")
        self.counts["user_privacy_settings"] += 1

    async def collect_and_import_activity_components(
        self,
        component_data: dict[str, list[dict[str, Any]]],
        original_activity_id: int,
        new_activity: activity_schema.ActivityRead,
    ) -> None:
        """Import contributed components for a single restored activity.

        Args:
            component_data: Records keyed by profile contributor key.
            original_activity_id: Old activity ID.
            new_activity: New activity object.
        """
        for contributor in activities_contributor_registry.profile_activity_contributors():
            restored = contributor.restore(
                component_data.get(contributor.key, []),
                original_activity_id,
                new_activity,
                self.db,
            )
            self.counts[contributor.count_key] += restored

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

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

        Returns:
            Dictionary mapping old activity IDs to new IDs.

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

        # Load activities list
        activities_data = self._load_single_json(zipf, "data/activities.json")
        if not activities_data:
            logger.info("No activities data to import")
            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}"
            )

        contributors = activities_contributor_registry.profile_activity_contributors()
        component_files: dict[str, list[str]] = {}
        component_records: dict[str, list[Any]] = {}
        for contributor in contributors:
            if contributor.split:
                component_files[contributor.key] = self._get_split_files_list(
                    file_list,
                    contributor.archive_path.removesuffix(".json"),
                )
            else:
                component_records[contributor.key] = self._load_single_json(
                    zipf,
                    contributor.archive_path,
                    check_memory=False,
                )

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

        # 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]

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

            batch_components: dict[str, list[dict[str, Any]]] = {}
            for contributor in contributors:
                if contributor.split:
                    batch_components[contributor.key] = self._load_components_for_batch(
                        zipf,
                        component_files[contributor.key],
                        activities_batch,
                        contributor.key,
                    )
                else:
                    batch_components[contributor.key] = self._filter_components_for_batch(
                        component_records.get(contributor.key, []),
                        activities_batch,
                    )

            # 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_contracts.ActivityCore(**activity_data)
                # Bulk restore deliberately does not publish activity.created;
                # see integration_service.restore_activity for why.
                new_activity = activity_integration.restore_activity(activity, self.db)

                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_components,
                        original_activity_id,
                        new_activity,
                    )

                # The activity and its components are one unit: the restore
                # functions stage, and the importer commits them together.
                self.db.commit()
                self.counts["activities"] += 1

            # Clear batch data from memory
            del batch_components
            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,
            )

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

    @staticmethod
    def _filter_components_for_batch(
        records: list[Any],
        activities_batch: list[Any],
    ) -> list[dict[str, Any]]:
        """Filter in-memory component records to the current activity batch."""
        activity_ids = {
            activity.get("id")
            for activity in activities_batch
            if isinstance(activity, dict) and activity.get("id") is not None
        }
        return [record for record in records if isinstance(record, dict) and record.get("activity_id") in activity_ids]

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

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

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

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

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

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

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

        all_components = []

        # Load and filter components from each file
        for filename in component_files:
            try:
                components = json.loads(self._read_zip_entry(zipf, filename))
                if not isinstance(components, list):
                    logger.warning(f"Expected list in {filename}, got {type(components).__name__}")
                    continue
                # Only keep components for activities in this batch
                filtered = [
                    component
                    for component in components
                    if isinstance(component, dict) and component.get("activity_id") in batch_activity_ids
                ]
                all_components.extend(filtered)

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

        return all_components

    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
            logger.info(f"Imported {self.counts['health_weight']} health weight records")
        else:
            logger.debug("No health weight data to import")

        # 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
            logger.info(f"Imported {self.counts['health_targets']} health targets")
        else:
            logger.debug("No health targets to import")

    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, validate as an activity file, then
                    # persist through the platform StorageProvider (the same store
                    # the ingestion path writes to) rather than the local
                    # processed directory. The blocking save is offloaded so it
                    # never stalls the import's event loop.
                    activity_limit = file_uploads.file_validator.config.limits.max_activity_file_size
                    file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=activity_limit)
                    try:
                        await file_uploads.validate_bytes(
                            file_bytes,
                            kind=file_uploads.UploadKind.ACTIVITY,
                            filename=new_file_name,
                        )
                    except core_exceptions.DomainError as err:
                        logger.warning(f"Profile import dropped invalid activity file {new_file_name}: {err.detail}")
                        continue
                    await asyncio.to_thread(
                        activity_file_storage_integration.store_activity_file,
                        new_id,
                        ext,
                        file_bytes,
                        platform_runtime.get_active_platform().storage,
                    )
                    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 validate them as an image
                        # before storing, then write through the platform
                        # StorageProvider under the media area (the same place
                        # an upload lands) rather than to a local directory.
                        image_limit = file_uploads.file_validator.config.limits.max_image_size
                        file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
                        try:
                            await file_uploads.validate_bytes(
                                file_bytes,
                                kind=file_uploads.UploadKind.IMAGE,
                                filename=new_file_name,
                            )
                        except core_exceptions.DomainError as err:
                            logger.warning(
                                f"Profile import dropped invalid activity media {new_file_name}: {err.detail}"
                            )
                            continue
                        activity_media_integration.store_media_blob(
                            new_id,
                            suffix,
                            ext,
                            file_bytes,
                            platform_runtime.get_active_platform().storage,
                        )
                        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/"):
                logger.debug(f"Processing user image file: {file_path}")
                ext = os.path.splitext(path)[1]
                new_file_name = f"{self.user_id}{ext}"

                # Read bytes from ZIP and save through the unified
                # validated-write pipeline so the entry is re-checked
                # as an image.
                image_limit = file_uploads.file_validator.config.limits.max_image_size
                file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
                try:
                    await file_uploads.validate_bytes(
                        file_bytes,
                        kind=file_uploads.UploadKind.IMAGE,
                        filename=new_file_name,
                    )
                except core_exceptions.DomainError as err:
                    logger.warning(f"Profile import dropped invalid user image {new_file_name}: {err.detail}")
                    continue
                platform_runtime.get_active_platform().storage.save(
                    users_signing.USER_IMAGE_STORAGE_AREA,
                    new_file_name,
                    file_bytes,
                )
                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/modules/users/users_profile/import_service.py 
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
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, validate as an activity file, then
                # persist through the platform StorageProvider (the same store
                # the ingestion path writes to) rather than the local
                # processed directory. The blocking save is offloaded so it
                # never stalls the import's event loop.
                activity_limit = file_uploads.file_validator.config.limits.max_activity_file_size
                file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=activity_limit)
                try:
                    await file_uploads.validate_bytes(
                        file_bytes,
                        kind=file_uploads.UploadKind.ACTIVITY,
                        filename=new_file_name,
                    )
                except core_exceptions.DomainError as err:
                    logger.warning(f"Profile import dropped invalid activity file {new_file_name}: {err.detail}")
                    continue
                await asyncio.to_thread(
                    activity_file_storage_integration.store_activity_file,
                    new_id,
                    ext,
                    file_bytes,
                    platform_runtime.get_active_platform().storage,
                )
                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/modules/users/users_profile/import_service.py 
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
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 validate them as an image
                    # before storing, then write through the platform
                    # StorageProvider under the media area (the same place
                    # an upload lands) rather than to a local directory.
                    image_limit = file_uploads.file_validator.config.limits.max_image_size
                    file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
                    try:
                        await file_uploads.validate_bytes(
                            file_bytes,
                            kind=file_uploads.UploadKind.IMAGE,
                            filename=new_file_name,
                        )
                    except core_exceptions.DomainError as err:
                        logger.warning(
                            f"Profile import dropped invalid activity media {new_file_name}: {err.detail}"
                        )
                        continue
                    activity_media_integration.store_media_blob(
                        new_id,
                        suffix,
                        ext,
                        file_bytes,
                        platform_runtime.get_active_platform().storage,
                    )
                    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/modules/users/users_profile/import_service.py 
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
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/"):
            logger.debug(f"Processing user image file: {file_path}")
            ext = os.path.splitext(path)[1]
            new_file_name = f"{self.user_id}{ext}"

            # Read bytes from ZIP and save through the unified
            # validated-write pipeline so the entry is re-checked
            # as an image.
            image_limit = file_uploads.file_validator.config.limits.max_image_size
            file_bytes = self._read_zip_entry(zipf, file_path, max_bytes=image_limit)
            try:
                await file_uploads.validate_bytes(
                    file_bytes,
                    kind=file_uploads.UploadKind.IMAGE,
                    filename=new_file_name,
                )
            except core_exceptions.DomainError as err:
                logger.warning(f"Profile import dropped invalid user image {new_file_name}: {err.detail}")
                continue
            platform_runtime.get_active_platform().storage.save(
                users_signing.USER_IMAGE_STORAGE_AREA,
                new_file_name,
                file_bytes,
            )
            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/modules/users/users_profile/import_service.py 
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
async def collect_and_import_activities_data_batched(
    self,
    zipf: zipfile.ZipFile,
    file_list: set[str],
    gears_id_mapping: dict[int, int],
    start_time: float,
    timeout_seconds: int,
) -> dict[int, int]:
    """
    Import activities in batches to manage memory.

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

    Returns:
        Dictionary mapping old activity IDs to new IDs.

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

    # Load activities list
    activities_data = self._load_single_json(zipf, "data/activities.json")
    if not activities_data:
        logger.info("No activities data to import")
        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}"
        )

    contributors = activities_contributor_registry.profile_activity_contributors()
    component_files: dict[str, list[str]] = {}
    component_records: dict[str, list[Any]] = {}
    for contributor in contributors:
        if contributor.split:
            component_files[contributor.key] = self._get_split_files_list(
                file_list,
                contributor.archive_path.removesuffix(".json"),
            )
        else:
            component_records[contributor.key] = self._load_single_json(
                zipf,
                contributor.archive_path,
                check_memory=False,
            )

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

    # 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]

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

        batch_components: dict[str, list[dict[str, Any]]] = {}
        for contributor in contributors:
            if contributor.split:
                batch_components[contributor.key] = self._load_components_for_batch(
                    zipf,
                    component_files[contributor.key],
                    activities_batch,
                    contributor.key,
                )
            else:
                batch_components[contributor.key] = self._filter_components_for_batch(
                    component_records.get(contributor.key, []),
                    activities_batch,
                )

        # 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_contracts.ActivityCore(**activity_data)
            # Bulk restore deliberately does not publish activity.created;
            # see integration_service.restore_activity for why.
            new_activity = activity_integration.restore_activity(activity, self.db)

            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_components,
                    original_activity_id,
                    new_activity,
                )

            # The activity and its components are one unit: the restore
            # functions stage, and the importer commits them together.
            self.db.commit()
            self.counts["activities"] += 1

        # Clear batch data from memory
        del batch_components
        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,
        )

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

collect_and_import_activity_components async 

collect_and_import_activity_components(component_data, original_activity_id, new_activity)

Import contributed components for a single restored activity.

Parameters:

Name Type Description Default
component_data dict[str, list[dict[str, Any]]]

Records keyed by profile contributor key.

 required
original_activity_id int

Old activity ID.

 required
new_activity ActivityRead

New activity object.

 required
Source code in backend/app/modules/users/users_profile/import_service.py 
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
async def collect_and_import_activity_components(
    self,
    component_data: dict[str, list[dict[str, Any]]],
    original_activity_id: int,
    new_activity: activity_schema.ActivityRead,
) -> None:
    """Import contributed components for a single restored activity.

    Args:
        component_data: Records keyed by profile contributor key.
        original_activity_id: Old activity ID.
        new_activity: New activity object.
    """
    for contributor in activities_contributor_registry.profile_activity_contributors():
        restored = contributor.restore(
            component_data.get(contributor.key, []),
            original_activity_id,
            new_activity,
            self.db,
        )
        self.counts[contributor.count_key] += restored

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/modules/users/users_profile/import_service.py 
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
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:
        logger.info("No gear components data to import")
        return

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

        gear_component_data.pop("id", None)

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

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

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/modules/users/users_profile/import_service.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
async def collect_and_import_gears_data(self, gears_data: list[Any]) -> dict[int, int]:
    """
    Import gear data and create ID mappings.

    Args:
        gears_data: List of gear data dictionaries.

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

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

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

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

    logger.info(f"Imported {self.counts['gears']} gears")
    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/modules/users/users_profile/import_service.py 
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
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
        logger.info(f"Imported {self.counts['health_weight']} health weight records")
    else:
        logger.debug("No health weight data to import")

    # 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
        logger.info(f"Imported {self.counts['health_targets']} health targets")
    else:
        logger.debug("No health targets to import")

collect_and_import_user_data async 

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

Import user profile and related settings.

Parameters:

Name Type Description Default
user_data list[Any]

User profile data.

 required
user_default_gear_data list[Any]

Default gear settings.

 required
user_goals_data list[Any]

User goals data.

 required
user_integrations_data list[Any]

Integration settings.

 required
user_privacy_settings_data list[Any]

Privacy settings.

 required
gears_id_mapping dict[int, int]

Mapping of old to new gear IDs.

 required
Source code in backend/app/modules/users/users_profile/import_service.py 
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
async def collect_and_import_user_data(
    self,
    user_data: list[Any],
    user_default_gear_data: list[Any],
    user_goals_data: list[Any],
    user_integrations_data: list[Any],
    user_privacy_settings_data: list[Any],
    gears_id_mapping: dict[int, int],
) -> None:
    """
    Import user profile and related settings.

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

        # The archive carries the exporter's photo key (or, for
        # archives predating storage keys, an absolute filesystem path).
        # Either way only the extension is trusted: the key is rebuilt for
        # the importing user so a crafted ZIP cannot repoint the avatar at
        # somebody else's blob.
        photo_path = user_profile.get("photo_path")
        extension = os.path.splitext(photo_path)[1].lstrip(".").lower() if isinstance(photo_path, str) else ""
        if extension in _IMPORTABLE_IMAGE_EXTENSIONS:
            user_profile["photo_path"] = f"{self.user_id}.{extension}"
        else:
            user_profile["photo_path"] = None

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

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

collect_and_import_user_default_gear async 

collect_and_import_user_default_gear(user_default_gear_data, gears_id_mapping)

Import user default gear settings with ID remapping.

Parameters:

Name Type Description Default
user_default_gear_data list[Any]

Default gear data.

 required
gears_id_mapping dict[int, int]

Mapping of old to new gear IDs.

 required
Source code in backend/app/modules/users/users_profile/import_service.py 
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
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:
        logger.info("No user default gear data to import")
        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:
        logger.info("No existing user default gear, creating new record")
        current_user_default_gear = user_default_gear_crud.create_user_default_gear(self.user_id, self.db)

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

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

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

    user_default_gear = user_default_gear_schema.UsersDefaultGearUpdate(**gear_data)
    user_default_gear_crud.edit_user_default_gear(user_default_gear, self.user_id, self.db)
    logger.info("Imported user default gear")
    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/modules/users/users_profile/import_service.py 
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
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:
        logger.info("No user goals data to import")
        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

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

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/modules/users/users_profile/import_service.py 
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
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:
        logger.info("No user integrations data to import")
        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)
    logger.info("Imported user integrations")
    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/modules/users/users_profile/import_service.py 
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
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:
        logger.info("No user privacy settings data to import")
        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)
    logger.info("Imported user privacy settings")
    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/modules/users/users_profile/import_service.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
async def import_from_zip_data(self, zip_data: bytes) -> dict[str, Any]:
    """
    Import profile data from ZIP file bytes.

    Args:
        zip_data: ZIP file content as bytes.

    Returns:
        Dictionary with import results and counts.

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

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

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

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

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

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

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

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

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

            self._restore_global_activity_components(zipf)

            # 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 health data
            profile_utils.check_timeout(timeout_seconds, start_time, ImportTimeoutError, "Import")
            health_weight_data = self._load_single_json(zipf, "data/health_weight.json")
            health_targets_data = self._load_single_json(zipf, "data/health_targets.json")

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

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

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

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

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

ImportTimeoutError

Bases: ProfileImportError

Raised when import operation exceeds time limit.

Source code in backend/app/modules/users/users_profile/exceptions.py 
112
113
114
115
116
117
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/modules/users/users_profile/exceptions.py 
88
89
90
91
92
93
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/modules/users/users_profile/exceptions.py 
152
153
154
155
156
157
class JSONParseError(ProfileImportError):
    """
    Raised when JSON data cannot be parsed.
    """

    pass

MemoryAllocationError

Bases: ProfileOperationError

Raised when memory allocation fails.

Source code in backend/app/modules/users/users_profile/exceptions.py 
63
64
65
66
67
68
class MemoryAllocationError(ProfileOperationError):
    """
    Raised when memory allocation fails.
    """

    pass

ProfileImportError

Bases: ProfileOperationError

Base exception for import operations.

Source code in backend/app/modules/users/users_profile/exceptions.py 
30
31
32
33
34
35
class ProfileImportError(ProfileOperationError):
    """
    Base exception for import operations.
    """

    pass

ProfileOperationError

Bases: Exception

Base exception for all profile operations.

Source code in backend/app/modules/users/users_profile/exceptions.py 
14
15
16
17
18
19
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/modules/users/users_profile/exceptions.py 
160
161
162
163
164
165
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/modules/users/users_profile/exceptions.py 
55
56
57
58
59
60
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/modules/users/users_profile/exceptions.py 
144
145
146
147
148
149
class ZipStructureError(ProfileImportError):
    """
    Raised when ZIP file structure is invalid.
    """

    pass

Users goals

User goals module for managing user fitness goals.

This module provides CRUD operations and data models for user goal tracking including activity type goals, intervals, and progress calculation.

Exports
  • CRUD: get_user_goals_by_user_id, get_user_goal_by_user_and_goal_id, create_user_goal, update_user_goal, delete_user_goal
  • Schemas: UsersGoalBase, UsersGoalCreate, UsersGoalUpdate, UsersGoalRead, UsersGoalProgress
  • Enums: Interval, ActivityType, GoalType
  • Utils: calculate_user_goals

ActivityType

Bases: StrEnum

Supported activity types for user goals.

Attributes:

Name Type Description
RUN

Running activities.
BIKE

Cycling activities.
SWIM

Swimming activities.
WALK

Walking or hiking activities.
STRENGTH

Strength or resistance training.
CARDIO

Cardiovascular training.
Source code in backend/app/modules/users/users_goals/schema.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class ActivityType(StrEnum):
    """
    Supported activity types for user goals.

    Attributes:
        RUN: Running activities.
        BIKE: Cycling activities.
        SWIM: Swimming activities.
        WALK: Walking or hiking activities.
        STRENGTH: Strength or resistance training.
        CARDIO: Cardiovascular training.
    """

    RUN = "run"
    BIKE = "bike"
    SWIM = "swim"
    WALK = "walk"
    STRENGTH = "strength"
    CARDIO = "cardio"

GoalType

Bases: StrEnum

Types of measurable user goals.

Attributes:

Name Type Description
CALORIES

Target calories burned.
ACTIVITIES

Target count of completed activities.
DISTANCE

Target distance traveled.
ELEVATION

Target elevation gain.
DURATION

Target total duration.
Source code in backend/app/modules/users/users_goals/schema.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class GoalType(StrEnum):
    """
    Types of measurable user goals.

    Attributes:
        CALORIES: Target calories burned.
        ACTIVITIES: Target count of completed activities.
        DISTANCE: Target distance traveled.
        ELEVATION: Target elevation gain.
        DURATION: Target total duration.
    """

    CALORIES = "calories"
    ACTIVITIES = "activities"
    DISTANCE = "distance"
    ELEVATION = "elevation"
    DURATION = "duration"

Interval

Bases: StrEnum

Recurrence intervals for user goals.

Attributes:

Name Type Description
DAILY

Daily recurrence interval.
WEEKLY

Weekly recurrence interval.
MONTHLY

Monthly recurrence interval.
YEARLY

Yearly recurrence interval.
Source code in backend/app/modules/users/users_goals/schema.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Interval(StrEnum):
    """
    Recurrence intervals for user goals.

    Attributes:
        DAILY: Daily recurrence interval.
        WEEKLY: Weekly recurrence interval.
        MONTHLY: Monthly recurrence interval.
        YEARLY: Yearly recurrence interval.
    """

    DAILY = "daily"
    WEEKLY = "weekly"
    MONTHLY = "monthly"
    YEARLY = "yearly"

UsersGoalBase

Bases: BaseModel

Base schema for users fitness goals.

Attributes:

Name Type Description
interval Interval

Goal time interval (daily, weekly, monthly, yearly).
activity_type ActivityType

Type of activity for the goal.
goal_type GoalType

Type of goal metric being tracked.
goal_calories StrictInt | None

Target calories in kcal.
goal_activities_number StrictInt | None

Target number of activities.
goal_distance StrictInt | None

Target distance in meters.
goal_elevation StrictInt | None

Target elevation gain in meters.
goal_duration StrictInt | None

Target duration in seconds.
Source code in backend/app/modules/users/users_goals/schema.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class UsersGoalBase(BaseModel):
    """
    Base schema for users fitness goals.

    Attributes:
        interval: Goal time interval (daily, weekly, monthly,
            yearly).
        activity_type: Type of activity for the goal.
        goal_type: Type of goal metric being tracked.
        goal_calories: Target calories in kcal.
        goal_activities_number: Target number of activities.
        goal_distance: Target distance in meters.
        goal_elevation: Target elevation gain in meters.
        goal_duration: Target duration in seconds.
    """

    interval: Interval = Field(..., description="Goal time interval")
    activity_type: ActivityType = Field(..., description="Type of activity for the goal")
    goal_type: GoalType = Field(..., description="Type of goal metric being tracked")
    goal_calories: StrictInt | None = Field(default=None, ge=0, description="Target calories in kcal")
    goal_activities_number: StrictInt | None = Field(default=None, ge=0, description="Target number of activities")
    goal_distance: StrictInt | None = Field(default=None, ge=0, description="Target distance in meters")
    goal_elevation: StrictInt | None = Field(default=None, ge=0, description="Target elevation gain in meters")
    goal_duration: StrictInt | None = Field(default=None, ge=0, description="Target duration in seconds")

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

    @model_validator(mode="after")
    def ensure_correct_goal_field(self) -> "UsersGoalBase":
        """
        Validate exactly one goal field matches goal_type.

        Returns:
            Self with validated goal field.

        Raises:
            PydanticCustomError: If required field missing or
                extra fields set.
        """
        goal_type_value = self.goal_type if isinstance(self.goal_type, str) else self.goal_type.value

        required_field = TYPE_TO_FIELD.get(goal_type_value)
        if required_field is None:
            return self

        if getattr(self, required_field) is None:
            raise PydanticCustomError(
                "missing_goal_value",
                "{field} is required when goal_type={goal_type}",
                {
                    "field": required_field,
                    "goal_type": goal_type_value,
                },
            )

        for name in TYPE_TO_FIELD.values():
            if name != required_field and getattr(self, name) is not None:
                raise PydanticCustomError(
                    "exclusive_goal_value",
                    "Only {field} may be set when goal_type={goal_type}",
                    {
                        "field": required_field,
                        "goal_type": goal_type_value,
                    },
                )
        return self

ensure_correct_goal_field 

ensure_correct_goal_field()

Validate exactly one goal field matches goal_type.

Returns:

Type Description
UsersGoalBase

Self with validated goal field.

Raises:

Type Description
PydanticCustomError

If required field missing or extra fields set.
Source code in backend/app/modules/users/users_goals/schema.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@model_validator(mode="after")
def ensure_correct_goal_field(self) -> "UsersGoalBase":
    """
    Validate exactly one goal field matches goal_type.

    Returns:
        Self with validated goal field.

    Raises:
        PydanticCustomError: If required field missing or
            extra fields set.
    """
    goal_type_value = self.goal_type if isinstance(self.goal_type, str) else self.goal_type.value

    required_field = TYPE_TO_FIELD.get(goal_type_value)
    if required_field is None:
        return self

    if getattr(self, required_field) is None:
        raise PydanticCustomError(
            "missing_goal_value",
            "{field} is required when goal_type={goal_type}",
            {
                "field": required_field,
                "goal_type": goal_type_value,
            },
        )

    for name in TYPE_TO_FIELD.values():
        if name != required_field and getattr(self, name) is not None:
            raise PydanticCustomError(
                "exclusive_goal_value",
                "Only {field} may be set when goal_type={goal_type}",
                {
                    "field": required_field,
                    "goal_type": goal_type_value,
                },
            )
    return self

UsersGoalCreate

Bases: UsersGoalBase

Schema for creating a new users goal.

Inherits all validation from UsersGoalBase.

Source code in backend/app/modules/users/users_goals/schema.py 
156
157
158
159
160
161
class UsersGoalCreate(UsersGoalBase):
    """
    Schema for creating a new users goal.

    Inherits all validation from UsersGoalBase.
    """

UsersGoalProgress

Bases: BaseModel

Schema for user goal progress tracking.

Attributes:

Name Type Description
goal_id StrictInt

Goal identifier.
interval Interval

Goal time interval.
activity_type ActivityType

Activity type for this goal.
goal_type GoalType

Type of goal metric.
start_date StrictStr

Period start date.
end_date StrictStr

Period end date.
percentage_completed StrictInt | None

Completion percentage.
total_calories StrictInt | None

Total calories achieved.
total_activities_number StrictInt | None

Total activities completed.
total_distance StrictInt | None

Total distance achieved.
total_elevation StrictInt | None

Total elevation gained.
total_duration StrictInt | None

Total duration achieved.
goal_calories StrictInt | None

Target calories.
goal_activities_number StrictInt | None

Target activities count.
goal_distance StrictInt | None

Target distance.
goal_elevation StrictInt | None

Target elevation.
goal_duration StrictInt | None

Target duration.
Source code in backend/app/modules/users/users_goals/schema.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class UsersGoalProgress(BaseModel):
    """
    Schema for user goal progress tracking.

    Attributes:
        goal_id: Goal identifier.
        interval: Goal time interval.
        activity_type: Activity type for this goal.
        goal_type: Type of goal metric.
        start_date: Period start date.
        end_date: Period end date.
        percentage_completed: Completion percentage.
        total_calories: Total calories achieved.
        total_activities_number: Total activities completed.
        total_distance: Total distance achieved.
        total_elevation: Total elevation gained.
        total_duration: Total duration achieved.
        goal_calories: Target calories.
        goal_activities_number: Target activities count.
        goal_distance: Target distance.
        goal_elevation: Target elevation.
        goal_duration: Target duration.
    """

    goal_id: StrictInt = Field(..., ge=1, description="Goal identifier")
    interval: Interval = Field(..., description="Goal time interval")
    activity_type: ActivityType = Field(..., description="Activity type for this goal")
    goal_type: GoalType
    start_date: StrictStr = Field(..., description="Period start date")
    end_date: StrictStr = Field(..., description="Period end date")
    percentage_completed: StrictInt | None = Field(default=None, ge=0, le=100, description="Completion percentage")
    # total
    total_calories: StrictInt | None = Field(default=None, ge=0, description="Total calories achieved")
    total_activities_number: StrictInt | None = Field(default=None, ge=0, description="Total activities completed")
    total_distance: StrictInt | None = Field(default=None, ge=0, description="Total distance achieved")
    total_elevation: StrictInt | None = Field(default=None, ge=0, description="Total elevation gained")
    total_duration: StrictInt | None = Field(default=None, ge=0, description="Total duration achieved")
    # goal
    goal_calories: StrictInt | None = Field(default=None, ge=0, description="Target calories")
    goal_activities_number: StrictInt | None = Field(default=None, ge=0, description="Target activities count")
    goal_distance: StrictInt | None = Field(default=None, ge=0, description="Target distance")
    goal_elevation: StrictInt | None = Field(default=None, ge=0, description="Target elevation")
    goal_duration: StrictInt | None = Field(default=None, ge=0, description="Target duration")

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

UsersGoalRead

Bases: UsersGoalBase

Schema for reading a user goal.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the goal.
user_id StrictInt

User who owns this goal.
Source code in backend/app/modules/users/users_goals/schema.py 
164
165
166
167
168
169
170
171
172
173
174
class UsersGoalRead(UsersGoalBase):
    """
    Schema for reading a user goal.

    Attributes:
        id: Unique identifier for the goal.
        user_id: User who owns this goal.
    """

    id: StrictInt = Field(..., ge=1, description="Unique identifier for the goal")
    user_id: StrictInt = Field(..., ge=1, description="User who owns this goal")

UsersGoalUpdate

Bases: UsersGoalRead

Schema for updating an existing users goal.

Supports partial updates. Omitted fields remain unchanged.

Source code in backend/app/modules/users/users_goals/schema.py 
177
178
179
180
181
182
class UsersGoalUpdate(UsersGoalRead):
    """
    Schema for updating an existing users goal.

    Supports partial updates. Omitted fields remain unchanged.
    """

calculate_user_goals 

calculate_user_goals(user_id, date, db, first_day_of_week)

Calculate progress for all user goals on a specified date.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
date str | None

Date in YYYY-MM-DD format. If None, uses current date.

 required
db Session

SQLAlchemy database session.

 required
first_day_of_week WeekdayValue

User's configured first weekday.

 required

Returns:

Type Description
list[UsersGoalProgress] | None

List of UsersGoalProgress objects, or None if no goals found.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/users/users_goals/utils.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def calculate_user_goals(
    user_id: int,
    date: str | None,
    db: Session,
    first_day_of_week: core_calendar.WeekdayValue,
) -> list[user_goals_schema.UsersGoalProgress] | None:
    """
    Calculate progress for all user goals on a specified date.

    Args:
        user_id: The ID of the user.
        date: Date in YYYY-MM-DD format. If None, uses
            current date.
        db: SQLAlchemy database session.
        first_day_of_week: User's configured first weekday.

    Returns:
        List of UsersGoalProgress objects, or None if no
            goals found.

    Raises:
        HTTPException: If database error occurs.
    """
    if not date:
        # The athlete's own calendar day, not the server's. Anchoring on
        # ``datetime.now(UTC)`` rolled daily goals over at UTC midnight and
        # flipped weekly/monthly/yearly windows up to a day early or late, so
        # progress bars visibly reset at the wrong time for anyone off UTC.
        date = users_utils.user_local_today(user_id, db).strftime("%Y-%m-%d")
    try:
        goals: list[user_goals_schema.UsersGoalRead] = user_goals_crud.get_user_goals_by_user_id(user_id, db)

        if not goals:
            return None

        return [
            calculate_goal_progress_by_activity_type(
                goal,
                date,
                db,
                first_day_of_week,
            )
            for goal in goals
        ]
    except HTTPException as http_err:
        raise http_err
    except (ValueError, TypeError) as err:
        # Log the exception
        logger.error(f"Error in calculate_user_goals: {err}", exc_info=err)
        # Raise an HTTPException with a 400 status code
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Invalid data provided",
        ) from err
    except Exception as err:
        # Log unexpected exceptions
        logger.error(f"Unexpected error in calculate_user_goals: {err}", exc_info=err)
        # Raise an HTTPException with a 500 status code
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Internal Server Error",
        ) from err

create_user_goal 

create_user_goal(user_id, user_goal, db)

Create a new user goal.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
user_goal UsersGoalCreate

Goal data to create.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersGoalRead

The created UsersGoal schema.

Raises:

Type Description
HTTPException

If goal already exists (409) or database error occurs (500).
Source code in backend/app/modules/users/users_goals/crud.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@core_decorators.handle_db_errors
def create_user_goal(
    user_id: int,
    user_goal: user_goals_schema.UsersGoalCreate,
    db: Session,
) -> user_goals_schema.UsersGoalRead:
    """
    Create a new user goal.

    Args:
        user_id: The ID of the user.
        user_goal: Goal data to create.
        db: SQLAlchemy database session.

    Returns:
        The created UsersGoal schema.

    Raises:
        HTTPException: If goal already exists (409) or database
            error occurs (500).
    """
    try:
        # Check if goal already exists
        stmt = select(user_goals_models.UsersGoal).where(
            user_goals_models.UsersGoal.user_id == user_id,
            user_goals_models.UsersGoal.interval == user_goal.interval,
            user_goals_models.UsersGoal.activity_type == user_goal.activity_type,
            user_goals_models.UsersGoal.goal_type == user_goal.goal_type,
        )
        existing_goal = db.execute(stmt).scalar_one_or_none()

        if existing_goal:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=("User already has a goal for this activity type, interval, and goal type."),
            )

        db_users_goal = user_goals_models.UsersGoal(
            user_id=user_id,
            interval=user_goal.interval,
            activity_type=user_goal.activity_type,
            goal_type=user_goal.goal_type,
            goal_calories=user_goal.goal_calories,
            goal_activities_number=user_goal.goal_activities_number,
            goal_distance=user_goal.goal_distance,
            goal_elevation=user_goal.goal_elevation,
            goal_duration=user_goal.goal_duration,
        )
        db.add(db_users_goal)
        db.commit()
        db.refresh(db_users_goal)
        return _transform_users_goals(db_users_goal)
    except HTTPException:
        raise

delete_user_goal 

delete_user_goal(user_id, goal_id, db)

Delete a user's goal from the database.

Parameters:

Name Type Description Default
user_id int

ID of the user who owns the goal.

 required
goal_id int

ID of the goal to delete.

 required
db Session

SQLAlchemy database session.

 required

Raises:

Type Description
HTTPException

If goal not found (404) or database error occurs (500).
Source code in backend/app/modules/users/users_goals/crud.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
@core_decorators.handle_db_errors
def delete_user_goal(user_id: int, goal_id: int, db: Session) -> None:
    """
    Delete a user's goal from the database.

    Args:
        user_id: ID of the user who owns the goal.
        goal_id: ID of the goal to delete.
        db: SQLAlchemy database session.

    Raises:
        HTTPException: If goal not found (404) or database
            error occurs (500).
    """
    db_users_goal = _get_user_goal_model_by_user_and_goal_id_or_404(user_id, goal_id, db)

    db.delete(db_users_goal)
    db.commit()

get_user_goal_by_user_and_goal_id 

get_user_goal_by_user_and_goal_id(user_id, goal_id, db)

Retrieve a specific goal by user ID and goal ID.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
goal_id int

The ID of the goal.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersGoalRead | None

UsersGoal schema if found, None otherwise.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/users/users_goals/crud.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@core_decorators.handle_db_errors
def get_user_goal_by_user_and_goal_id(
    user_id: int, goal_id: int, db: Session
) -> user_goals_schema.UsersGoalRead | None:
    """
    Retrieve a specific goal by user ID and goal ID.

    Args:
        user_id: The ID of the user.
        goal_id: The ID of the goal.
        db: SQLAlchemy database session.

    Returns:
        UsersGoal schema if found, None otherwise.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = select(user_goals_models.UsersGoal).where(
        user_goals_models.UsersGoal.user_id == user_id,
        user_goals_models.UsersGoal.id == goal_id,
    )
    result = db.execute(stmt).scalar_one_or_none()
    return _transform_users_goals(result) if result else None

get_user_goals_by_user_id 

get_user_goals_by_user_id(user_id, db, interval=None, activity_type=None, goal_type=None)

Retrieve goals for a specific user with optional filters.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
db Session

SQLAlchemy database session.

 required
interval Interval | None

Optional filter by goal interval.

 None
activity_type ActivityType | None

Optional filter by activity type.

 None
goal_type GoalType | None

Optional filter by goal type.

 None

Returns:

Type Description
list[UsersGoalRead]

List of UsersGoal schemas matching the filters.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/users/users_goals/crud.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@core_decorators.handle_db_errors
def get_user_goals_by_user_id(
    user_id: int,
    db: Session,
    interval: user_goals_schema.Interval | None = None,
    activity_type: user_goals_schema.ActivityType | None = None,
    goal_type: user_goals_schema.GoalType | None = None,
) -> list[user_goals_schema.UsersGoalRead]:
    """
    Retrieve goals for a specific user with optional filters.

    Args:
        user_id: The ID of the user.
        db: SQLAlchemy database session.
        interval: Optional filter by goal interval.
        activity_type: Optional filter by activity type.
        goal_type: Optional filter by goal type.

    Returns:
        List of UsersGoal schemas matching the filters.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = select(user_goals_models.UsersGoal).where(user_goals_models.UsersGoal.user_id == user_id)

    if interval is not None:
        stmt = stmt.where(user_goals_models.UsersGoal.interval == interval.value)
    if activity_type is not None:
        stmt = stmt.where(user_goals_models.UsersGoal.activity_type == activity_type.value)
    if goal_type is not None:
        stmt = stmt.where(user_goals_models.UsersGoal.goal_type == goal_type.value)

    return _transform_users_goals(list(db.execute(stmt).scalars().all()))

update_user_goal 

update_user_goal(user_id, user_goal, db)

Update a user's goal.

Parameters:

Name Type Description Default
user_id int

ID of the user.

 required
user_goal UsersGoalUpdate

Schema with fields to update.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersGoalRead

The updated UsersGoal schema.

Raises:

Type Description
HTTPException

If goal not found (404) or database error occurs (500).
Source code in backend/app/modules/users/users_goals/crud.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
@core_decorators.handle_db_errors
def update_user_goal(
    user_id: int,
    user_goal: user_goals_schema.UsersGoalUpdate,
    db: Session,
) -> user_goals_schema.UsersGoalRead:
    """
    Update a user's goal.

    Args:
        user_id: ID of the user.
        user_goal: Schema with fields to update.
        db: SQLAlchemy database session.

    Returns:
        The updated UsersGoal schema.

    Raises:
        HTTPException: If goal not found (404) or database
            error occurs (500).
    """
    if user_goal.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Cannot edit user goal for another user",
        )

    db_users_goal = _get_user_goal_model_by_user_and_goal_id_or_404(user_id, user_goal.id, db)

    # Update only provided fields
    update_data = user_goal.model_dump(exclude_unset=True, exclude={"user_id", "id"})
    for field, value in update_data.items():
        setattr(db_users_goal, field, value)

    db.commit()
    db.refresh(db_users_goal)
    return _transform_users_goals(db_users_goal)

Users default gear

User default gear module for activity type gear assignments.

This module provides CRUD operations and data models for managing user default gear settings per activity type (run, ride, swim, etc.).

Exports
  • CRUD: get_user_default_gear_by_user_id, create_user_default_gear, edit_user_default_gear
  • Schemas: UsersDefaultGearBase, UsersDefaultGearUpdate, UsersDefaultGearRead
  • Utils: get_user_default_gear_by_activity_type
  • Constants: ACTIVITY_TYPE_TO_GEAR_ATTR

UsersDefaultGearBase

Bases: BaseModel

Base schema for users default gear assignments.

Attributes:

Name Type Description
run_gear_id StrictInt | None

Default gear for run activities.
trail_run_gear_id StrictInt | None

Default gear for trail run activities.
virtual_run_gear_id StrictInt | None

Default gear for virtual run activities.
ride_gear_id StrictInt | None

Default gear for ride activities.
gravel_ride_gear_id StrictInt | None

Default gear for gravel ride activities.
mtb_ride_gear_id StrictInt | None

Default gear for MTB ride activities.
virtual_ride_gear_id StrictInt | None

Default gear for virtual ride activities.
ows_gear_id StrictInt | None

Default gear for open water swim activities.
walk_gear_id StrictInt | None

Default gear for walk activities.
hike_gear_id StrictInt | None

Default gear for hike activities.
tennis_gear_id StrictInt | None

Default gear for tennis activities.
alpine_ski_gear_id StrictInt | None

Default gear for alpine ski activities.
nordic_ski_gear_id StrictInt | None

Default gear for nordic ski activities.
snowboard_gear_id StrictInt | None

Default gear for snowboard activities.
windsurf_gear_id StrictInt | None

Default gear for windsurf activities.
Source code in backend/app/modules/users/users_default_gear/schema.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
class UsersDefaultGearBase(BaseModel):
    """
    Base schema for users default gear assignments.

    Attributes:
        run_gear_id: Default gear for run activities.
        trail_run_gear_id: Default gear for trail run
            activities.
        virtual_run_gear_id: Default gear for virtual run
            activities.
        ride_gear_id: Default gear for ride activities.
        gravel_ride_gear_id: Default gear for gravel ride
            activities.
        mtb_ride_gear_id: Default gear for MTB ride activities.
        virtual_ride_gear_id: Default gear for virtual ride
            activities.
        ows_gear_id: Default gear for open water swim
            activities.
        walk_gear_id: Default gear for walk activities.
        hike_gear_id: Default gear for hike activities.
        tennis_gear_id: Default gear for tennis activities.
        alpine_ski_gear_id: Default gear for alpine ski
            activities.
        nordic_ski_gear_id: Default gear for nordic ski
            activities.
        snowboard_gear_id: Default gear for snowboard
            activities.
        windsurf_gear_id: Default gear for windsurf activities.
    """

    run_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for run activities",
    )
    trail_run_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for trail run activities",
    )
    virtual_run_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for virtual run activities",
    )
    ride_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for ride activities",
    )
    gravel_ride_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for gravel ride activities",
    )
    mtb_ride_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for MTB ride activities",
    )
    virtual_ride_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for virtual ride activities",
    )
    ows_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for OWS activities",
    )
    walk_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for walk activities",
    )
    hike_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for hike activities",
    )
    tennis_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for tennis activities",
    )
    alpine_ski_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for alpine ski activities",
    )
    nordic_ski_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for nordic ski activities",
    )
    snowboard_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for snowboard activities",
    )
    windsurf_gear_id: StrictInt | None = Field(
        default=None,
        ge=1,
        description="Default gear for windsurf activities",
    )

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

UsersDefaultGearRead

Bases: UsersDefaultGearBase

Schema for reading user default gear settings.

Extends base with identifier fields.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the record.
user_id StrictInt

User ID reference.
Source code in backend/app/modules/users/users_default_gear/schema.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
class UsersDefaultGearRead(UsersDefaultGearBase):
    """
    Schema for reading user default gear settings.

    Extends base with identifier fields.

    Attributes:
        id: Unique identifier for the record.
        user_id: User ID reference.
    """

    id: StrictInt = Field(..., ge=1, description="Unique identifier")
    user_id: StrictInt = Field(..., ge=1, description="User ID")

UsersDefaultGearUpdate

Bases: UsersDefaultGearRead

Schema for updating user default gear settings.

Inherits all gear assignment fields from read schema. All fields are optional for partial updates.

Source code in backend/app/modules/users/users_default_gear/schema.py 
139
140
141
142
143
144
145
class UsersDefaultGearUpdate(UsersDefaultGearRead):
    """
    Schema for updating user default gear settings.

    Inherits all gear assignment fields from read schema.
    All fields are optional for partial updates.
    """

create_user_default_gear 

create_user_default_gear(user_id, db)

Create default gear settings for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to create settings for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersDefaultGearRead

The created UsersDefaultGear schema.

Raises:

Type Description
HTTPException

409 error if settings already exist.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_default_gear/crud.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@core_decorators.handle_db_errors
def create_user_default_gear(
    user_id: int,
    db: Session,
) -> users_default_gear_schema.UsersDefaultGearRead:
    """
    Create default gear settings for a user.

    Args:
        user_id: The ID of the user to create settings for.
        db: SQLAlchemy database session.

    Returns:
        The created UsersDefaultGear schema.

    Raises:
        HTTPException: 409 error if settings already exist.
        HTTPException: 500 error if database operation fails.
    """
    try:
        db_users_default_gear = users_default_gear_models.UsersDefaultGear(
            user_id=user_id,
        )

        db.add(db_users_default_gear)
        db.commit()
        db.refresh(db_users_default_gear)

        return _transform_users_default_gear(db_users_default_gear)
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Default gear settings already exist for this user"),
        ) from integrity_error

edit_user_default_gear 

edit_user_default_gear(user_default_gear, user_id, db)

Update default gear settings for a user.

Parameters:

Name Type Description Default
user_default_gear UsersDefaultGearUpdate

Schema with gear fields to update.

 required
user_id int

The ID of the user.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersDefaultGearRead

The updated UsersDefaultGear schema.

Raises:

Type Description
HTTPException

404 error if settings not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_default_gear/crud.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
@core_decorators.handle_db_errors
def edit_user_default_gear(
    user_default_gear: users_default_gear_schema.UsersDefaultGearUpdate,
    user_id: int,
    db: Session,
) -> users_default_gear_schema.UsersDefaultGearRead:
    """
    Update default gear settings for a user.

    Args:
        user_default_gear: Schema with gear fields to update.
        user_id: The ID of the user.
        db: SQLAlchemy database session.

    Returns:
        The updated UsersDefaultGear schema.

    Raises:
        HTTPException: 404 error if settings not found.
        HTTPException: 500 error if database operation fails.
    """
    if user_default_gear.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Cannot edit default gear for another user.",
        )

    db_users_default_gear = _get_user_default_gear_model_by_user_id_or_404(user_id, db)

    user_default_gear_data: dict[str, Any] = user_default_gear.model_dump(exclude_unset=True, exclude={"user_id", "id"})
    gears_dependencies.require_owned_gears(user_id, user_default_gear_data.values(), db)
    for key, value in user_default_gear_data.items():
        setattr(db_users_default_gear, key, value)

    db.commit()
    db.refresh(db_users_default_gear)

    return _transform_users_default_gear(db_users_default_gear)

get_user_default_gear_by_activity_type 

get_user_default_gear_by_activity_type(user_id, activity_type, db)

Get default gear ID for a specific activity type.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
activity_type int

The activity type code.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
int | None

The gear ID for the activity type, or None if not set
int | None

or activity type is unknown.

Raises:

Type Description
HTTPException

If user default gear not found or database error occurs.
Source code in backend/app/modules/users/users_default_gear/utils.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def get_user_default_gear_by_activity_type(
    user_id: int,
    activity_type: int,
    db: Session,
) -> int | None:
    """
    Get default gear ID for a specific activity type.

    Args:
        user_id: The ID of the user.
        activity_type: The activity type code.
        db: SQLAlchemy database session.

    Returns:
        The gear ID for the activity type, or None if not set
        or activity type is unknown.

    Raises:
        HTTPException: If user default gear not found or
            database error occurs.
    """
    try:
        user_default_gear: users_default_gear_schema.UsersDefaultGearRead | None = (
            user_default_gear_crud.get_user_default_gear_by_user_id(user_id, db)
        )

        if not user_default_gear:
            return None

        attr_name = ACTIVITY_TYPE_TO_GEAR_ATTR.get(activity_type)
        if attr_name is None:
            return None

        return getattr(user_default_gear, attr_name, None)
    except HTTPException:
        raise
    except Exception as err:
        logger.error(f"Error in get_user_default_gear_by_activity_type: {err}", exc_info=err)
        raise

get_user_default_gear_by_user_id 

get_user_default_gear_by_user_id(user_id, db)

Retrieve default gear settings for a specific user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to fetch settings for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersDefaultGearRead | None

The UsersDefaultGear model for the user or None if not found.

Raises:

Type Description
HTTPException

404 error if settings not found.
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users_default_gear/crud.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@core_decorators.handle_db_errors
def get_user_default_gear_by_user_id(
    user_id: int,
    db: Session,
) -> users_default_gear_schema.UsersDefaultGearRead | None:
    """
    Retrieve default gear settings for a specific user.

    Args:
        user_id: The ID of the user to fetch settings for.
        db: SQLAlchemy database session.

    Returns:
        The UsersDefaultGear model for the user or None if not found.

    Raises:
        HTTPException: 404 error if settings not found.
        HTTPException: 500 error if database query fails.
    """
    stmt = select(users_default_gear_models.UsersDefaultGear).where(
        users_default_gear_models.UsersDefaultGear.user_id == user_id
    )
    db_users_default_gear = db.execute(stmt).scalar_one_or_none()
    return _transform_users_default_gear(db_users_default_gear) if db_users_default_gear else None

Users integrations

User integrations module for third-party service OAuth management.

This module provides CRUD operations and data models for user integration settings including Strava and Garmin Connect OAuth tokens and synchronization preferences.

Exports
  • CRUD: get_user_integrations_by_user_id, create_user_integrations, edit_user_integrations, link_strava_account, unlink_strava_account, set_user_strava_client, set_user_strava_state, set_user_strava_sync_gear, link_garminconnect_account, unlink_garminconnect_account, set_user_garminconnect_sync_gear, get_user_integrations_by_strava_state
  • Schemas: UsersIntegrationsBase, UsersIntegrationsCreate, UsersIntegrationsRead, UsersIntegrationsUpdate

UsersIntegrationsBase

Bases: BaseModel

Base schema for user integrations.

Attributes:

Name Type Description
strava_client_id StrictStr | None

Encrypted Strava client ID.
strava_client_secret StrictStr | None

Encrypted Strava client secret.
strava_state StrictStr | None

Temporary state for Strava OAuth flow.
strava_token StrictStr | None

Encrypted Strava access token.
strava_refresh_token StrictStr | None

Encrypted Strava refresh token.
strava_token_expires_at datetime | None

Strava token expiration time.
strava_sync_gear StrictBool

Enable Strava gear synchronization.
garminconnect_token dict[str, Any] | None

Garmin Connect token data.
garminconnect_sync_gear StrictBool

Enable Garmin Connect gear synchronization.
Source code in backend/app/modules/users/users_integrations/schema.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class UsersIntegrationsBase(BaseModel):
    """
    Base schema for user integrations.

    Attributes:
        strava_client_id: Encrypted Strava client ID.
        strava_client_secret: Encrypted Strava client secret.
        strava_state: Temporary state for Strava OAuth flow.
        strava_token: Encrypted Strava access token.
        strava_refresh_token: Encrypted Strava refresh token.
        strava_token_expires_at: Strava token expiration time.
        strava_sync_gear: Enable Strava gear synchronization.
        garminconnect_token: Garmin Connect token data.
        garminconnect_sync_gear: Enable Garmin Connect gear
            synchronization.
    """

    strava_client_id: StrictStr | None = Field(
        default=None,
        max_length=512,
        description=("Strava client ID encrypted at rest with Fernet key"),
    )
    strava_client_secret: StrictStr | None = Field(
        default=None,
        max_length=512,
        description=("Strava client secret encrypted at rest with Fernet key"),
    )
    strava_state: StrictStr | None = Field(
        default=None,
        max_length=45,
        description="Strava temporary state for link process",
    )
    strava_token: StrictStr | None = Field(
        default=None,
        max_length=512,
        description=("Strava token after link process encrypted at rest with Fernet key"),
    )
    strava_refresh_token: StrictStr | None = Field(
        default=None,
        max_length=512,
        description=("Strava refresh token after link process encrypted at rest with Fernet key"),
    )
    strava_token_expires_at: datetime | None = Field(default=None, description="Strava token expiration date")
    strava_sync_gear: StrictBool = Field(default=False, description="Whether Strava gear is to be synced")
    garminconnect_token: dict[str, Any] | None = Field(default=None, description="Garmin Connect token")
    garminconnect_sync_gear: StrictBool = Field(
        default=False,
        description="Whether Garmin Connect gear is to be synced",
    )

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

UsersIntegrationsCreate

Bases: UsersIntegrationsBase

Pydantic model for creating user integrations.

Inherits all attributes from UsersIntegrationsBase.

Source code in backend/app/modules/users/users_integrations/schema.py 
73
74
75
76
77
78
class UsersIntegrationsCreate(UsersIntegrationsBase):
    """
    Pydantic model for creating user integrations.

    Inherits all attributes from UsersIntegrationsBase.
    """

UsersIntegrationsRead

Bases: UsersIntegrationsBase

Schema for reading user integrations.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the integrations record.
user_id StrictInt

Foreign key reference to the user.
Source code in backend/app/modules/users/users_integrations/schema.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class UsersIntegrationsRead(UsersIntegrationsBase):
    """
    Schema for reading user integrations.

    Attributes:
        id: Unique identifier for the integrations record.
        user_id: Foreign key reference to the user.
    """

    id: StrictInt = Field(..., ge=1, description="Unique identifier for integrations")
    user_id: StrictInt = Field(..., ge=1, description="Foreign key reference to the user")

UsersIntegrationsUpdate

Bases: UsersIntegrationsBase

Schema for updating user integrations.

Inherits all validation from UsersIntegrationsBase. All fields are optional for partial updates.

Source code in backend/app/modules/users/users_integrations/schema.py 
81
82
83
84
85
86
87
class UsersIntegrationsUpdate(UsersIntegrationsBase):
    """
    Schema for updating user integrations.

    Inherits all validation from UsersIntegrationsBase.
    All fields are optional for partial updates.
    """

create_user_integrations 

create_user_integrations(user_id, db)

Create integration settings for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to create integrations for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersIntegrationsRead

The created UsersIntegrationsRead schema.

Raises:

Type Description
HTTPException

409 error if integrations already exist.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
@core_decorators.handle_db_errors
def create_user_integrations(user_id: int, db: Session) -> user_integrations_schema.UsersIntegrationsRead:
    """
    Create integration settings for a user.

    Args:
        user_id: The ID of the user to create integrations for.
        db: SQLAlchemy database session.

    Returns:
        The created UsersIntegrationsRead schema.

    Raises:
        HTTPException: 409 error if integrations already exist.
        HTTPException: 500 error if database operation fails.
    """
    try:
        # Create a new user integrations with model defaults
        user_integrations = user_integrations_models.UsersIntegrations(
            user_id=user_id,
        )

        # Add the user integrations to the database
        db.add(user_integrations)
        db.commit()
        db.refresh(user_integrations)

        # Return the user integrations
        return _transform_users_integrations(user_integrations)
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Integrations already exist for this user",
        ) from integrity_error

edit_user_integrations 

edit_user_integrations(user_integrations, user_id, db)

Update user integration settings.

Parameters:

Name Type Description Default
user_integrations UsersIntegrationsUpdate

Schema with fields to update.

 required
user_id int

The ID of the user to update.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersIntegrationsRead

The updated UsersIntegrations model.

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
@core_decorators.handle_db_errors
def edit_user_integrations(
    user_integrations: user_integrations_schema.UsersIntegrationsUpdate,
    user_id: int,
    db: Session,
) -> user_integrations_schema.UsersIntegrationsRead:
    """
    Update user integration settings.

    Args:
        user_integrations: Schema with fields to update.
        user_id: The ID of the user to update.
        db: SQLAlchemy database session.

    Returns:
        The updated UsersIntegrations model.

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations from the database
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Get fields to update
    user_integrations_data = user_integrations.model_dump(exclude_unset=True, exclude={"user_id", "id"})
    # Update fields dynamically
    for key, value in user_integrations_data.items():
        setattr(db_user_integrations, key, value)

    # Commit the transaction
    db.commit()
    db.refresh(db_user_integrations)
    return _transform_users_integrations(db_user_integrations)

get_user_integrations_by_strava_state 

get_user_integrations_by_strava_state(strava_state, db)

Retrieve integrations by Strava OAuth state token.

Parameters:

Name Type Description Default
strava_state str

The Strava OAuth state to search for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersIntegrationsRead | None

The UsersIntegrationsRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@core_decorators.handle_db_errors
def get_user_integrations_by_strava_state(
    strava_state: str, db: Session
) -> user_integrations_schema.UsersIntegrationsRead | None:
    """
    Retrieve integrations by Strava OAuth state token.

    Args:
        strava_state: The Strava OAuth state to search for.
        db: SQLAlchemy database session.

    Returns:
        The UsersIntegrationsRead schema if found, None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    # Get user integrations based on the strava state
    stmt = select(user_integrations_models.UsersIntegrations).where(
        user_integrations_models.UsersIntegrations.strava_state == strava_state
    )
    user_integrations = db.execute(stmt).scalar_one_or_none()
    return _transform_users_integrations(user_integrations) if user_integrations else None

get_user_integrations_by_user_id 

get_user_integrations_by_user_id(user_id, db)

Retrieve integrations for a specific user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to fetch integrations for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersIntegrationsRead | None

The UsersIntegrationsRead schema for the user.

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@core_decorators.handle_db_errors
def get_user_integrations_by_user_id(
    user_id: int, db: Session
) -> user_integrations_schema.UsersIntegrationsRead | None:
    """
    Retrieve integrations for a specific user.

    Args:
        user_id: The ID of the user to fetch integrations for.
        db: SQLAlchemy database session.

    Returns:
        The UsersIntegrationsRead schema for the user.

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database query fails.
    """
    # Get the user integrations by the user id
    stmt = select(user_integrations_models.UsersIntegrations).where(
        user_integrations_models.UsersIntegrations.user_id == user_id
    )
    user_integrations = db.execute(stmt).scalar_one_or_none()
    return _transform_users_integrations(user_integrations) if user_integrations else None

link_garminconnect_account(user_id, token, db)

Link a Garmin Connect account by storing token.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
token dict

Garmin Connect token data.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
@core_decorators.handle_db_errors
def link_garminconnect_account(
    user_id: int,
    token: dict,
    db: Session,
) -> None:
    """
    Link a Garmin Connect account by storing token.

    Args:
        user_id: The ID of the user.
        token: Garmin Connect token data.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Store Garmin Connect token
    db_user_integrations.garminconnect_token = token

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

link_strava_account(user_id, tokens, db)

Link a Strava account by storing OAuth tokens.

Parameters:

Name Type Description Default
user_id int

The ID of the user to update.

 required
tokens dict

Dictionary containing access_token, refresh_token, and expires_at.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
@core_decorators.handle_db_errors
def link_strava_account(
    user_id: int,
    tokens: dict,
    db: Session,
) -> None:
    """
    Link a Strava account by storing OAuth tokens.

    Args:
        user_id: The ID of the user to update.
        tokens: Dictionary containing access_token,
            refresh_token, and expires_at.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Update the user integrations with the tokens
    db_user_integrations.strava_token = core_cryptography.encrypt_token_fernet(tokens["access_token"])
    db_user_integrations.strava_refresh_token = core_cryptography.encrypt_token_fernet(tokens["refresh_token"])
    db_user_integrations.strava_token_expires_at = datetime.fromtimestamp(tokens["expires_at"], tz=UTC)

    # Set the strava state to None
    db_user_integrations.strava_state = None

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

set_user_garminconnect_sync_gear 

set_user_garminconnect_sync_gear(user_id, garminconnect_sync_gear, db)

Set Garmin Connect gear synchronization preference.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
garminconnect_sync_gear bool

Whether to sync gear from Garmin Connect.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
@core_decorators.handle_db_errors
def set_user_garminconnect_sync_gear(user_id: int, garminconnect_sync_gear: bool, db: Session) -> None:
    """
    Set Garmin Connect gear synchronization preference.

    Args:
        user_id: The ID of the user.
        garminconnect_sync_gear: Whether to sync gear from
            Garmin Connect.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Set the Garmin Connect gear sync preference
    db_user_integrations.garminconnect_sync_gear = garminconnect_sync_gear

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

set_user_strava_client 

set_user_strava_client(user_id, client_id, client_secret, db)

Set Strava client credentials for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
client_id str

Strava client ID to encrypt and store.

 required
client_secret str

Strava client secret to encrypt and store.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
@core_decorators.handle_db_errors
def set_user_strava_client(user_id: int, client_id: str, client_secret: str, db: Session) -> None:
    """
    Set Strava client credentials for a user.

    Args:
        user_id: The ID of the user.
        client_id: Strava client ID to encrypt and store.
        client_secret: Strava client secret to encrypt and store.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Encrypt and store Strava client credentials
    db_user_integrations.strava_client_id = core_cryptography.encrypt_token_fernet(client_id)
    db_user_integrations.strava_client_secret = core_cryptography.encrypt_token_fernet(client_secret)

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

set_user_strava_state 

set_user_strava_state(user_id, state, db)

Set or clear Strava OAuth state for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
state str | None

Strava OAuth state to set, or None to clear.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
@core_decorators.handle_db_errors
def set_user_strava_state(user_id: int, state: str | None, db: Session) -> None:
    """
    Set or clear Strava OAuth state for a user.

    Args:
        user_id: The ID of the user.
        state: Strava OAuth state to set, or None to clear.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Set the user Strava state
    db_user_integrations.strava_state = None if state in ("null", None) else state

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

set_user_strava_sync_gear 

set_user_strava_sync_gear(user_id, strava_sync_gear, db)

Set Strava gear synchronization preference for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user.

 required
strava_sync_gear bool

Whether to sync gear from Strava.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
@core_decorators.handle_db_errors
def set_user_strava_sync_gear(user_id: int, strava_sync_gear: bool, db: Session) -> None:
    """
    Set Strava gear synchronization preference for a user.

    Args:
        user_id: The ID of the user.
        strava_sync_gear: Whether to sync gear from Strava.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Set the Strava gear sync preference
    db_user_integrations.strava_sync_gear = strava_sync_gear

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

unlink_garminconnect_account(user_id, db)

Unlink a Garmin Connect account by clearing OAuth tokens.

Parameters:

Name Type Description Default
user_id int

The ID of the user to unlink.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
@core_decorators.handle_db_errors
def unlink_garminconnect_account(user_id: int, db: Session) -> None:
    """
    Unlink a Garmin Connect account by clearing OAuth tokens.

    Args:
        user_id: The ID of the user to unlink.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Clear all Garmin Connect integration data
    db_user_integrations.garminconnect_token = None
    db_user_integrations.garminconnect_sync_gear = False

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

unlink_strava_account(user_id, db)

Unlink a Strava account by clearing OAuth tokens.

Parameters:

Name Type Description Default
user_id int

The ID of the user to unlink.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 error if integrations not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_integrations/crud.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
@core_decorators.handle_db_errors
def unlink_strava_account(user_id: int, db: Session) -> None:
    """
    Unlink a Strava account by clearing OAuth tokens.

    Args:
        user_id: The ID of the user to unlink.
        db: SQLAlchemy database session.

    Returns:
        None

    Raises:
        HTTPException: 404 error if integrations not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user integrations by the user id
    db_user_integrations = _get_user_integrations_model_by_user_id_or_404(user_id, db)

    # Clear all Strava integration data
    db_user_integrations.strava_state = None
    db_user_integrations.strava_token = None
    db_user_integrations.strava_refresh_token = None
    db_user_integrations.strava_token_expires_at = None
    db_user_integrations.strava_sync_gear = False
    db_user_integrations.strava_client_id = None
    db_user_integrations.strava_client_secret = None

    # Commit the changes to the database
    db.commit()
    db.refresh(db_user_integrations)

Users privacy settings

User privacy settings module for activity visibility control.

This module provides CRUD operations and data models for user privacy settings including activity visibility levels and various data hiding options for activities.

Exports
  • CRUD: get_user_privacy_settings_by_user_id, create_user_privacy_settings, edit_user_privacy_settings
  • Schemas: UsersPrivacySettingsBase, UsersPrivacySettingsCreate, UsersPrivacySettingsRead, UsersPrivacySettingsUpdate
  • Models: UsersPrivacySettings (ORM model)
  • Enums: ActivityVisibility

ActivityVisibility

Bases: Enum

Activity visibility levels.

Attributes:

Name Type Description
PUBLIC

Visible to everyone.
FOLLOWERS

Visible only to followers.
PRIVATE

Visible only to the user.
Source code in backend/app/modules/users/users_privacy_settings/schema.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
class ActivityVisibility(Enum):
    """
    Activity visibility levels.

    Attributes:
        PUBLIC: Visible to everyone.
        FOLLOWERS: Visible only to followers.
        PRIVATE: Visible only to the user.
    """

    PUBLIC = "public"
    FOLLOWERS = "followers"
    PRIVATE = "private"

UsersPrivacySettingsBase

Bases: BaseModel

Base schema for user privacy settings.

Attributes:

Name Type Description
default_activity_visibility ActivityVisibility | None

Default activity visibility level.
hide_activity_start_time StrictBool | None

Hide start time from activities.
hide_activity_location StrictBool | None

Hide location data from activities.
hide_activity_map StrictBool | None

Hide map visualization from activities.
hide_activity_hr StrictBool | None

Hide heart rate data from activities.
hide_activity_power StrictBool | None

Hide power data from activities.
hide_activity_cadence StrictBool | None

Hide cadence data from activities.
hide_activity_elevation StrictBool | None

Hide elevation data from activities.
hide_activity_speed StrictBool | None

Hide speed data from activities.
hide_activity_pace StrictBool | None

Hide pace data from activities.
hide_activity_laps StrictBool | None

Hide lap data from activities.
hide_activity_workout_sets_steps StrictBool | None

Hide workout sets and steps from activities.
hide_activity_gear StrictBool | None

Hide gear information from activities.
Source code in backend/app/modules/users/users_privacy_settings/schema.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
class UsersPrivacySettingsBase(BaseModel):
    """
    Base schema for user privacy settings.

    Attributes:
        default_activity_visibility: Default activity visibility
            level.
        hide_activity_start_time: Hide start time from
            activities.
        hide_activity_location: Hide location data from
            activities.
        hide_activity_map: Hide map visualization from
            activities.
        hide_activity_hr: Hide heart rate data from activities.
        hide_activity_power: Hide power data from activities.
        hide_activity_cadence: Hide cadence data from
            activities.
        hide_activity_elevation: Hide elevation data from
            activities.
        hide_activity_speed: Hide speed data from activities.
        hide_activity_pace: Hide pace data from activities.
        hide_activity_laps: Hide lap data from activities.
        hide_activity_workout_sets_steps: Hide workout sets and
            steps from activities.
        hide_activity_gear: Hide gear information from
            activities.
    """

    default_activity_visibility: ActivityVisibility | None = Field(
        default=ActivityVisibility.PUBLIC,
        description="Default activity visibility (public, followers, private)",
    )
    hide_activity_start_time: StrictBool | None = Field(default=False, description="Hide activity start time")
    hide_activity_location: StrictBool | None = Field(default=False, description="Hide activity location")
    hide_activity_map: StrictBool | None = Field(default=False, description="Hide activity map")
    hide_activity_hr: StrictBool | None = Field(default=False, description="Hide activity heart rate")
    hide_activity_power: StrictBool | None = Field(default=False, description="Hide activity power")
    hide_activity_cadence: StrictBool | None = Field(default=False, description="Hide activity cadence")
    hide_activity_elevation: StrictBool | None = Field(default=False, description="Hide activity elevation")
    hide_activity_speed: StrictBool | None = Field(default=False, description="Hide activity speed")
    hide_activity_pace: StrictBool | None = Field(default=False, description="Hide activity pace")
    hide_activity_laps: StrictBool | None = Field(default=False, description="Hide activity laps")
    hide_activity_workout_sets_steps: StrictBool | None = Field(
        default=False, description="Hide activity workout sets and steps"
    )
    hide_activity_gear: StrictBool | None = Field(default=False, description="Hide activity gear")

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

UsersPrivacySettingsCreate

Bases: UsersPrivacySettingsBase

Pydantic model for creating user privacy settings.

Inherits all attributes from UsersPrivacySettingsBase.

Source code in backend/app/modules/users/users_privacy_settings/schema.py 
84
85
86
87
88
89
class UsersPrivacySettingsCreate(UsersPrivacySettingsBase):
    """
    Pydantic model for creating user privacy settings.

    Inherits all attributes from UsersPrivacySettingsBase.
    """

UsersPrivacySettingsRead

Bases: UsersPrivacySettingsBase

Schema for reading user privacy settings.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the privacy settings record.
user_id StrictInt

Foreign key reference to the user.
Source code in backend/app/modules/users/users_privacy_settings/schema.py 
101
102
103
104
105
106
107
108
109
110
111
class UsersPrivacySettingsRead(UsersPrivacySettingsBase):
    """
    Schema for reading user privacy settings.

    Attributes:
        id: Unique identifier for the privacy settings record.
        user_id: Foreign key reference to the user.
    """

    id: StrictInt = Field(..., ge=1, description="Unique identifier for privacy settings")
    user_id: StrictInt = Field(..., ge=1, description="Foreign key reference to the user")

UsersPrivacySettingsUpdate

Bases: UsersPrivacySettingsBase

Schema for updating user privacy settings.

Inherits all validation from UsersPrivacySettingsBase. All fields are optional for partial updates.

Source code in backend/app/modules/users/users_privacy_settings/schema.py 
92
93
94
95
96
97
98
class UsersPrivacySettingsUpdate(UsersPrivacySettingsBase):
    """
    Schema for updating user privacy settings.

    Inherits all validation from UsersPrivacySettingsBase.
    All fields are optional for partial updates.
    """

create_user_privacy_settings 

create_user_privacy_settings(user_id, db)

Create privacy settings for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to create settings for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersPrivacySettingsRead

The created UsersPrivacySettingsRead schema.

Raises:

Type Description
HTTPException

409 error if settings already exist.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_privacy_settings/crud.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@core_decorators.handle_db_errors
def create_user_privacy_settings(user_id: int, db: Session) -> users_privacy_settings_schema.UsersPrivacySettingsRead:
    """
    Create privacy settings for a user.

    Args:
        user_id: The ID of the user to create settings for.
        db: SQLAlchemy database session.

    Returns:
        The created UsersPrivacySettingsRead schema.

    Raises:
        HTTPException: 409 error if settings already exist.
        HTTPException: 500 error if database operation fails.
    """
    try:
        # Create a new user privacy settings with model defaults
        db_privacy_settings = users_privacy_settings_models.UsersPrivacySettings(
            user_id=user_id,
        )

        # Add the user privacy settings to the database
        db.add(db_privacy_settings)
        db.commit()
        db.refresh(db_privacy_settings)

        # Return the user privacy settings
        return _transform_users_privacy_settings(db_privacy_settings)
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Conflict status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Privacy settings already exist for this user",
        ) from integrity_error

edit_user_privacy_settings 

edit_user_privacy_settings(user_id, user_privacy_settings_data, db)

Update privacy settings for a specific user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to update settings for.

 required
user_privacy_settings_data UsersPrivacySettingsUpdate

Schema with fields to update.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersPrivacySettingsRead

The updated UsersPrivacySettingsRead schema.

Raises:

Type Description
HTTPException

404 error if settings not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/users/users_privacy_settings/crud.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@core_decorators.handle_db_errors
def edit_user_privacy_settings(
    user_id: int,
    user_privacy_settings_data: users_privacy_settings_schema.UsersPrivacySettingsUpdate,
    db: Session,
) -> users_privacy_settings_schema.UsersPrivacySettingsRead:
    """
    Update privacy settings for a specific user.

    Args:
        user_id: The ID of the user to update settings for.
        user_privacy_settings_data: Schema with fields to update.
        db: SQLAlchemy database session.

    Returns:
        The updated UsersPrivacySettingsRead schema.

    Raises:
        HTTPException: 404 error if settings not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user privacy settings by the user id
    db_user_privacy_settings = _get_user_privacy_settings_model_by_user_id_or_404(user_id, db)

    # Dictionary of the fields to update if they are not None
    privacy_settings_dict = user_privacy_settings_data.model_dump(exclude_unset=True, exclude={"user_id", "id"})
    # Iterate over the fields and update dynamically
    for key, value in privacy_settings_dict.items():
        setattr(db_user_privacy_settings, key, value)

    # Commit the transaction
    db.commit()
    db.refresh(db_user_privacy_settings)

    # Return the updated user privacy settings
    return _transform_users_privacy_settings(db_user_privacy_settings)

get_user_privacy_settings_by_user_id 

get_user_privacy_settings_by_user_id(user_id, db)

Retrieve privacy settings for a specific user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to fetch settings for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
UsersPrivacySettingsRead | None

The UsersPrivacySettingsRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/users/users_privacy_settings/crud.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@core_decorators.handle_db_errors
def get_user_privacy_settings_by_user_id(
    user_id: int, db: Session
) -> users_privacy_settings_schema.UsersPrivacySettingsRead | None:
    """
    Retrieve privacy settings for a specific user.

    Args:
        user_id: The ID of the user to fetch settings for.
        db: SQLAlchemy database session.

    Returns:
        The UsersPrivacySettingsRead schema if found, None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    # Get the user privacy settings by the user id
    stmt = select(users_privacy_settings_models.UsersPrivacySettings).where(
        users_privacy_settings_models.UsersPrivacySettings.user_id == user_id
    )
    db_users_privacy_settings = db.execute(stmt).scalar_one_or_none()
    return _transform_users_privacy_settings(db_users_privacy_settings) if db_users_privacy_settings else None