Skip to content

Health API reference

Health sleep

Health sleep module for managing user sleep data.

This module provides CRUD operations and data models for user sleep tracking including sleep duration, stages, quality metrics, heart rate, SpO2, and sleep scoring.

Exports
  • CRUD: get_health_sleep_number_by_user_id, get_health_sleep_by_user_id, get_health_sleep_by_date_and_user_id, create_health_sleep, edit_health_sleep, delete_health_sleep
  • Schemas: HealthSleepBase, HealthSleepCreate, HealthSleepUpdate, HealthSleepRead, HealthSleepListResponse, HealthSleepStage
  • Enums: Source, SleepStageType, HRVStatus, SleepScore
  • Models: HealthSleep (ORM model)

HRVStatus

Bases: Enum

Enum representing the Heart Rate Variability (HRV) status classification.

Attributes:

Name Type Description
BALANCED

Indicates healthy HRV levels with good cardiovascular autonomic balance.
UNBALANCED

Indicates HRV levels showing some autonomic imbalance or stress.
LOW

Indicates HRV levels that are lower than optimal, suggesting fatigue or overtraining.
POOR

Indicates critically low HRV levels, suggesting significant stress, illness, or recovery issues.
Source code in backend/app/modules/health/health_sleep/schema.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class HRVStatus(Enum):
    """
    Enum representing the Heart Rate Variability (HRV) status classification.

    Attributes:
        BALANCED: Indicates healthy HRV levels with good cardiovascular autonomic balance.
        UNBALANCED: Indicates HRV levels showing some autonomic imbalance or stress.
        LOW: Indicates HRV levels that are lower than optimal, suggesting fatigue or overtraining.
        POOR: Indicates critically low HRV levels, suggesting significant stress, illness, or recovery issues.
    """

    BALANCED = "BALANCED"
    UNBALANCED = "UNBALANCED"
    LOW = "LOW"
    POOR = "POOR"

HealthSleepBase

Bases: BaseModel

Pydantic model representing the base schema for health sleep data.

This model defines the structure and validation rules for sleep-related health metrics, including sleep duration, quality scores, heart rate variability, oxygen saturation, respiration rates, and sleep stages.

Attributes:

Name Type Description
date date | None

Calendar date of the sleep session.
sleep_start_time_gmt datetime | None

Start time of sleep in GMT.
sleep_end_time_gmt datetime | None

End time of sleep in GMT.
sleep_start_time_local datetime | None

Start time of sleep in local time.
sleep_end_time_local datetime | None

End time of sleep in local time.
total_sleep_seconds StrictInt | None

Total duration of sleep in seconds (>= 0).
nap_time_seconds StrictInt | None

Duration of naps in seconds (>= 0).
unmeasurable_sleep_seconds StrictInt | None

Unmeasurable sleep duration in seconds (>= 0).
deep_sleep_seconds StrictInt | None

Duration of deep sleep in seconds (>= 0).
light_sleep_seconds StrictInt | None

Duration of light sleep in seconds (>= 0).
rem_sleep_seconds StrictInt | None

Duration of REM sleep in seconds (>= 0).
awake_sleep_seconds StrictInt | None

Duration of awake time in seconds (>= 0).
avg_heart_rate StrictInt | None

Average heart rate during sleep (20-220 bpm).
min_heart_rate StrictInt | None

Minimum heart rate during sleep (20-220 bpm).
max_heart_rate StrictInt | None

Maximum heart rate during sleep (20-220 bpm).
avg_spo2 StrictInt | None

Average SpO2 oxygen saturation percentage (70-100%).
lowest_spo2 StrictInt | None

Lowest SpO2 reading during sleep (70-100%).
highest_spo2 StrictInt | None

Highest SpO2 reading during sleep (70-100%).
avg_respiration StrictInt | None

Average respiration rate (>= 0).
lowest_respiration StrictInt | None

Lowest respiration rate (>= 0).
highest_respiration StrictInt | None

Highest respiration rate (>= 0).
avg_stress_level StrictInt | None

Average stress level (0-100).
awake_count StrictInt | None

Number of times awake during sleep (>= 0).
restless_moments_count StrictInt | None

Number of restless moments during sleep (>= 0).
sleep_score_overall StrictInt | None

Overall sleep score (0-100).
sleep_score_duration SleepScore | None

Sleep duration score.
sleep_score_quality SleepScore | None

Sleep quality score.
garminconnect_sleep_id StrictStr | None

Garmin Connect sleep record ID.
sleep_stages list[SleepStage] | None

List of sleep stages.
source Source | None

Source of the sleep data.
hrv_status HRVStatus | None

Heart rate variability status.
resting_heart_rate StrictInt | None

Resting heart rate (20-220 bpm).
avg_skin_temp_deviation StrictFloat | None

Average skin temperature deviation.
awake_count_score SleepScore | None

Awake count score.
rem_percentage_score SleepScore | None

REM percentage score.
deep_percentage_score SleepScore | None

Deep sleep percentage score.
light_percentage_score SleepScore | None

Light sleep percentage score.
avg_sleep_stress StrictInt | None

Average sleep stress (0-100).
sleep_stress_score SleepScore | None

Sleep stress score.
Model Configuration
  • Allows population from ORM attributes
  • Forbids extra fields
  • Validates on assignment
  • Uses enum values in serialization
Validators

validate_sleep_times: Ensures sleep start time is before sleep end time for both GMT and local time zones.

Source code in backend/app/modules/health/health_sleep/schema.py 
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
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
class HealthSleepBase(BaseModel):
    """
    Pydantic model representing the base schema for health sleep data.

    This model defines the structure and validation rules for sleep-related health metrics,
    including sleep duration, quality scores, heart rate variability, oxygen saturation,
    respiration rates, and sleep stages.

    Attributes:
        date (datetime_date | None): Calendar date of the sleep session.
        sleep_start_time_gmt (datetime | None): Start time of sleep in GMT.
        sleep_end_time_gmt (datetime | None): End time of sleep in GMT.
        sleep_start_time_local (datetime | None): Start time of sleep in local time.
        sleep_end_time_local (datetime | None): End time of sleep in local time.
        total_sleep_seconds (StrictInt | None): Total duration of sleep in seconds (>= 0).
        nap_time_seconds (StrictInt | None): Duration of naps in seconds (>= 0).
        unmeasurable_sleep_seconds (StrictInt | None): Unmeasurable sleep duration in seconds (>= 0).
        deep_sleep_seconds (StrictInt | None): Duration of deep sleep in seconds (>= 0).
        light_sleep_seconds (StrictInt | None): Duration of light sleep in seconds (>= 0).
        rem_sleep_seconds (StrictInt | None): Duration of REM sleep in seconds (>= 0).
        awake_sleep_seconds (StrictInt | None): Duration of awake time in seconds (>= 0).
        avg_heart_rate (StrictInt | None): Average heart rate during sleep (20-220 bpm).
        min_heart_rate (StrictInt | None): Minimum heart rate during sleep (20-220 bpm).
        max_heart_rate (StrictInt | None): Maximum heart rate during sleep (20-220 bpm).
        avg_spo2 (StrictInt | None): Average SpO2 oxygen saturation percentage (70-100%).
        lowest_spo2 (StrictInt | None): Lowest SpO2 reading during sleep (70-100%).
        highest_spo2 (StrictInt | None): Highest SpO2 reading during sleep (70-100%).
        avg_respiration (StrictInt | None): Average respiration rate (>= 0).
        lowest_respiration (StrictInt | None): Lowest respiration rate (>= 0).
        highest_respiration (StrictInt | None): Highest respiration rate (>= 0).
        avg_stress_level (StrictInt | None): Average stress level (0-100).
        awake_count (StrictInt | None): Number of times awake during sleep (>= 0).
        restless_moments_count (StrictInt | None): Number of restless moments during sleep (>= 0).
        sleep_score_overall (StrictInt | None): Overall sleep score (0-100).
        sleep_score_duration (SleepScore | None): Sleep duration score.
        sleep_score_quality (SleepScore | None): Sleep quality score.
        garminconnect_sleep_id (StrictStr | None): Garmin Connect sleep record ID.
        sleep_stages (list[SleepStage] | None): List of sleep stages.
        source (Source | None): Source of the sleep data.
        hrv_status (HRVStatus | None): Heart rate variability status.
        resting_heart_rate (StrictInt | None): Resting heart rate (20-220 bpm).
        avg_skin_temp_deviation (StrictFloat | None): Average skin temperature deviation.
        awake_count_score (SleepScore | None): Awake count score.
        rem_percentage_score (SleepScore | None): REM percentage score.
        deep_percentage_score (SleepScore | None): Deep sleep percentage score.
        light_percentage_score (SleepScore | None): Light sleep percentage score.
        avg_sleep_stress (StrictInt | None): Average sleep stress (0-100).
        sleep_stress_score (SleepScore | None): Sleep stress score.

    Model Configuration:
        - Allows population from ORM attributes
        - Forbids extra fields
        - Validates on assignment
        - Uses enum values in serialization

    Validators:
        validate_sleep_times: Ensures sleep start time is before sleep end time
                             for both GMT and local time zones.
    """

    date: datetime_date | None = Field(default=None, description="Calendar date of the sleep session")
    sleep_start_time_gmt: datetime | None = Field(default=None, description="Start time of sleep in GMT")
    sleep_end_time_gmt: datetime | None = Field(default=None, description="End time of sleep in GMT")
    sleep_start_time_local: datetime | None = Field(default=None, description="Start time of sleep in local time")
    sleep_end_time_local: datetime | None = Field(default=None, description="End time of sleep in local time")
    total_sleep_seconds: StrictInt | None = Field(default=None, ge=0, description="Total duration of sleep in seconds")
    nap_time_seconds: StrictInt | None = Field(default=None, ge=0, description="Duration of naps in seconds")
    unmeasurable_sleep_seconds: StrictInt | None = Field(
        default=None, ge=0, description="Unmeasurable sleep duration in seconds"
    )
    deep_sleep_seconds: StrictInt | None = Field(default=None, ge=0, description="Duration of deep sleep in seconds")
    light_sleep_seconds: StrictInt | None = Field(default=None, ge=0, description="Duration of light sleep in seconds")
    rem_sleep_seconds: StrictInt | None = Field(default=None, ge=0, description="Duration of REM sleep in seconds")
    awake_sleep_seconds: StrictInt | None = Field(default=None, ge=0, description="Duration of awake time in seconds")
    avg_heart_rate: StrictInt | None = Field(default=None, ge=20, le=220, description="Average heart rate during sleep")
    min_heart_rate: StrictInt | None = Field(default=None, ge=20, le=220, description="Minimum heart rate during sleep")
    max_heart_rate: StrictInt | None = Field(default=None, ge=20, le=220, description="Maximum heart rate during sleep")
    avg_spo2: StrictInt | None = Field(
        default=None,
        ge=70,
        le=100,
        description="Average SpO2 oxygen saturation percentage",
    )
    lowest_spo2: StrictInt | None = Field(default=None, ge=70, le=100, description="Lowest SpO2 reading during sleep")
    highest_spo2: StrictInt | None = Field(default=None, ge=70, le=100, description="Highest SpO2 reading during sleep")
    avg_respiration: StrictInt | None = Field(default=None, ge=0, description="Average respiration rate")
    lowest_respiration: StrictInt | None = Field(default=None, ge=0, description="Lowest respiration rate")
    highest_respiration: StrictInt | None = Field(default=None, ge=0, description="Highest respiration rate")
    avg_stress_level: StrictInt | None = Field(default=None, ge=0, le=100, description="Average stress level")
    awake_count: StrictInt | None = Field(default=None, ge=0, description="Number of times awake during sleep")
    restless_moments_count: StrictInt | None = Field(
        default=None, ge=0, description="Number of restless moments during sleep"
    )
    sleep_score_overall: StrictInt | None = Field(default=None, ge=0, le=100, description="Overall sleep score (0-100)")
    sleep_score_duration: SleepScore | None = Field(default=None, description="Sleep duration score")
    sleep_score_quality: SleepScore | None = Field(default=None, description="Sleep quality score")
    garminconnect_sleep_id: StrictStr | None = Field(
        default=None, min_length=1, description="Garmin Connect sleep record ID"
    )
    sleep_stages: list[HealthSleepStage] | None = Field(default=None, description="List of sleep stages")
    source: Source | None = Field(default=None, description="Source of the sleep data")
    hrv_status: HRVStatus | None = Field(default=None, description="Heart rate variability status")
    resting_heart_rate: StrictInt | None = Field(default=None, ge=20, le=220, description="Resting heart rate")
    avg_skin_temp_deviation: StrictFloat | None = Field(default=None, description="Average skin temperature deviation")
    awake_count_score: SleepScore | None = Field(default=None, description="Awake count score")
    rem_percentage_score: SleepScore | None = Field(default=None, description="REM percentage score")
    deep_percentage_score: SleepScore | None = Field(default=None, description="Deep sleep percentage score")
    light_percentage_score: SleepScore | None = Field(default=None, description="Light sleep percentage score")
    avg_sleep_stress: StrictInt | None = Field(default=None, ge=0, le=100, description="Average sleep stress")
    sleep_stress_score: SleepScore | None = Field(default=None, description="Sleep stress score")

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

    @model_validator(mode="after")
    def validate_sleep_times(self) -> "HealthSleepBase":
        """Validate sleep start < end."""
        # Validate sleep start < sleep end (GMT)
        if (
            self.sleep_start_time_gmt is not None and self.sleep_end_time_gmt is not None
        ) and self.sleep_start_time_gmt >= self.sleep_end_time_gmt:
            raise ValueError("Sleep start time must be before sleep end time")

        # Validate sleep start < sleep end (Local)
        if (
            self.sleep_start_time_local is not None and self.sleep_end_time_local is not None
        ) and self.sleep_start_time_local >= self.sleep_end_time_local:
            raise ValueError("Sleep start time (local) must be before sleep end time (local)")

        return self

validate_sleep_times 

validate_sleep_times()

Validate sleep start < end.

Source code in backend/app/modules/health/health_sleep/schema.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
@model_validator(mode="after")
def validate_sleep_times(self) -> "HealthSleepBase":
    """Validate sleep start < end."""
    # Validate sleep start < sleep end (GMT)
    if (
        self.sleep_start_time_gmt is not None and self.sleep_end_time_gmt is not None
    ) and self.sleep_start_time_gmt >= self.sleep_end_time_gmt:
        raise ValueError("Sleep start time must be before sleep end time")

    # Validate sleep start < sleep end (Local)
    if (
        self.sleep_start_time_local is not None and self.sleep_end_time_local is not None
    ) and self.sleep_start_time_local >= self.sleep_end_time_local:
        raise ValueError("Sleep start time (local) must be before sleep end time (local)")

    return self

HealthSleepCreate

Bases: HealthSleepBase

Validator for HealthSleepCreate model that automatically sets the date field.

This validator runs after model initialization to ensure that if no date is provided, it defaults to today's date.

Source code in backend/app/modules/health/health_sleep/schema.py 
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
class HealthSleepCreate(HealthSleepBase):
    """
    Validator for HealthSleepCreate model that automatically sets the date field.

    This validator runs after model initialization to ensure that if no date
    is provided, it defaults to today's date.
    """

    @model_validator(mode="after")
    def set_default_date(self) -> "HealthSleepCreate":
        """
        Set date to today if not provided.

        Clients should always send their own local date — the request carries no
        timezone, so this fallback can only use the server's configured zone and
        will file an entry logged near local midnight under the wrong day for an
        athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
        than ``date.today()``, which silently depends on the container's clock.

        Returns:
            The validated model instance with date set.
        """
        if self.date is None:
            self.date = core_timezone.today_in(core_config.settings.TZ)
        return self

set_default_date 

set_default_date()

Set date to today if not provided.

Clients should always send their own local date — the request carries no timezone, so this fallback can only use the server's configured zone and will file an entry logged near local midnight under the wrong day for an athlete elsewhere. Resolved through settings.TZ explicitly rather than date.today(), which silently depends on the container's clock.

Returns:

Type Description
HealthSleepCreate

The validated model instance with date set.
Source code in backend/app/modules/health/health_sleep/schema.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@model_validator(mode="after")
def set_default_date(self) -> "HealthSleepCreate":
    """
    Set date to today if not provided.

    Clients should always send their own local date — the request carries no
    timezone, so this fallback can only use the server's configured zone and
    will file an entry logged near local midnight under the wrong day for an
    athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
    than ``date.today()``, which silently depends on the container's clock.

    Returns:
        The validated model instance with date set.
    """
    if self.date is None:
        self.date = core_timezone.today_in(core_config.settings.TZ)
    return self

HealthSleepListResponse

Bases: HealthListResponse

Response model for listing health sleep records.

Extends the base HealthListResponse to provide a typed response containing a list of sleep records. This model is used when returning multiple sleep entries from API endpoints.

Attributes:

Name Type Description
records list[HealthSleepRead]

A list of HealthSleepRead objects representing individual sleep records.
Source code in backend/app/modules/health/health_sleep/schema.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
class HealthSleepListResponse(health_schema.HealthListResponse):
    """
    Response model for listing health sleep records.

    Extends the base HealthListResponse to provide a typed response containing
    a list of sleep records. This model is used when returning multiple sleep
    entries from API endpoints.

    Attributes:
        records: A list of HealthSleepRead objects representing individual
            sleep records.
    """

    records: list[HealthSleepRead] = Field(..., description="List of health sleep records")

HealthSleepRead

Bases: HealthSleepBase

Read schema for health sleep records.

Extends the base health sleep schema with an identifier field for retrieving or referencing existing sleep records in the database.

Attributes:

Name Type Description
user_id StrictInt

Foreign key reference to the user.
id StrictInt

Unique identifier for the sleep record to update or retrieve.
Source code in backend/app/modules/health/health_sleep/schema.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
class HealthSleepRead(HealthSleepBase):
    """
    Read schema for health sleep records.

    Extends the base health sleep schema with an identifier field for retrieving
    or referencing existing sleep records in the database.

    Attributes:
        user_id (StrictInt): Foreign key reference to the user.
        id (StrictInt): Unique identifier for the sleep record to update or retrieve.
    """

    id: StrictInt = Field(..., description="Unique identifier for the sleep record to update")
    user_id: StrictInt = Field(..., description="Foreign key reference to the user")

HealthSleepStage

Bases: BaseModel

Represents a sleep stage with timing and duration information.

This model captures details about individual sleep stages within a sleep session, including the type of stage, when it occurred, and how long it lasted.

Attributes:

Name Type Description
stage_type SleepStageType | None

The classification of the sleep stage (e.g., light, deep, REM).
start_time_gmt datetime | None

The beginning timestamp of the sleep stage in GMT timezone.
end_time_gmt datetime | None

The ending timestamp of the sleep stage in GMT timezone.
duration_seconds StrictInt | None

The length of the sleep stage in seconds. Must be non-negative.
Source code in backend/app/modules/health/health_sleep/schema.py 
 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
class HealthSleepStage(BaseModel):
    """
    Represents a sleep stage with timing and duration information.

    This model captures details about individual sleep stages within a sleep session,
    including the type of stage, when it occurred, and how long it lasted.

    Attributes:
        stage_type: The classification of the sleep stage (e.g., light, deep, REM).
        start_time_gmt: The beginning timestamp of the sleep stage in GMT timezone.
        end_time_gmt: The ending timestamp of the sleep stage in GMT timezone.
        duration_seconds: The length of the sleep stage in seconds. Must be non-negative.
    """

    stage_type: SleepStageType | None = Field(default=None, description="Type of sleep stage")
    start_time_gmt: datetime | None = Field(default=None, description="Start time of the stage in GMT")
    end_time_gmt: datetime | None = Field(default=None, description="End time of the stage in GMT")
    duration_seconds: StrictInt | None = Field(default=None, ge=0, description="Duration of the stage in seconds")

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

HealthSleepUpdate

Bases: HealthSleepRead

Schema for updating health sleep records.

Inherits all fields from HealthSleepRead, allowing clients to update existing sleep data while maintaining consistency with the read schema.

Used in PUT/PATCH requests to modify sleep tracking information such as sleep duration, quality, and related health metrics.

Source code in backend/app/modules/health/health_sleep/schema.py 
288
289
290
291
292
293
294
295
296
297
class HealthSleepUpdate(HealthSleepRead):
    """
    Schema for updating health sleep records.

    Inherits all fields from HealthSleepRead, allowing clients to update
    existing sleep data while maintaining consistency with the read schema.

    Used in PUT/PATCH requests to modify sleep tracking information such as
    sleep duration, quality, and related health metrics.
    """

SleepScore

Bases: Enum

Enumeration of sleep quality score levels.

Attributes:

Name Type Description
EXCELLENT str

Represents excellent sleep quality.
GOOD str

Represents good sleep quality.
FAIR str

Represents fair sleep quality.
POOR str

Represents poor sleep quality.
Source code in backend/app/modules/health/health_sleep/schema.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
class SleepScore(Enum):
    """
    Enumeration of sleep quality score levels.

    Attributes:
        EXCELLENT (str): Represents excellent sleep quality.
        GOOD (str): Represents good sleep quality.
        FAIR (str): Represents fair sleep quality.
        POOR (str): Represents poor sleep quality.
    """

    EXCELLENT = "EXCELLENT"
    GOOD = "GOOD"
    FAIR = "FAIR"
    POOR = "POOR"

SleepStageType

Bases: Enum

Enum representing different stages of sleep.

Attributes:

Name Type Description
DEEP int

Deep sleep stage, value 0. Characterized by slow brain waves and muscle relaxation.
LIGHT int

Light sleep stage, value 1. Transitional sleep between wakefulness and deep sleep.
REM int

Rapid Eye Movement sleep stage, value 2. Associated with vivid dreams and mental activity.
AWAKE int

Awake state, value 3. The stage when the person is conscious and not sleeping.
Source code in backend/app/modules/health/health_sleep/schema.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class SleepStageType(Enum):
    """
    Enum representing different stages of sleep.

    Attributes:
        DEEP (int): Deep sleep stage, value 0. Characterized by slow brain waves and muscle relaxation.
        LIGHT (int): Light sleep stage, value 1. Transitional sleep between wakefulness and deep sleep.
        REM (int): Rapid Eye Movement sleep stage, value 2. Associated with vivid dreams and mental activity.
        AWAKE (int): Awake state, value 3. The stage when the person is conscious and not sleeping.
    """

    DEEP = 0
    LIGHT = 1
    REM = 2
    AWAKE = 3

Source

Bases: Enum

Enum representing the source of sleep health data.

Attributes:

Name Type Description
GARMIN

Sleep data sourced from Garmin devices or services.
Source code in backend/app/modules/health/health_sleep/schema.py 
20
21
22
23
24
25
26
27
28
class Source(Enum):
    """
    Enum representing the source of sleep health data.

    Attributes:
        GARMIN: Sleep data sourced from Garmin devices or services.
    """

    GARMIN = "garmin"

create_health_sleep 

create_health_sleep(user_id, health_sleep, db)

Create a new health sleep record for a user.

Parameters:

Name Type Description Default
user_id int

User ID for the sleep record.

 required
health_sleep HealthSleepCreate

Health sleep data to create.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthSleepRead

Created health sleep record with assigned ID.

Raises:

Type Description
HTTPException

409 if duplicate entry, 500 if database error.
Source code in backend/app/modules/health/health_sleep/crud.py 
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
@core_decorators.handle_db_errors
def create_health_sleep(
    user_id: int,
    health_sleep: health_sleep_schema.HealthSleepCreate,
    db: Session,
) -> health_sleep_schema.HealthSleepRead:
    """
    Create a new health sleep record for a user.

    Args:
        user_id: User ID for the sleep record.
        health_sleep: Health sleep data to create.
        db: Database session.

    Returns:
        Created health sleep record with assigned ID.

    Raises:
        HTTPException: 409 if duplicate entry, 500 if database
            error.
    """
    try:
        # Create a new health_sleep
        db_health_sleep = health_sleep_models.HealthSleep(
            **health_sleep.model_dump(
                exclude_none=False,
                mode="json",
            ),
            user_id=user_id,
        )

        # Add the health_sleep to the database
        db.add(db_health_sleep)
        db.commit()
        db.refresh(db_health_sleep)

        # Return the created health_sleep model
        return _transform_health_sleep(db_health_sleep)
    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=(f"Duplicate entry error. Check if there is already an entry created for {health_sleep.date}"),
        ) from integrity_error

delete_health_sleep 

delete_health_sleep(user_id, health_sleep_id, db)

Delete a health sleep record for a specific user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the sleep record.

 required
health_sleep_id int

Sleep record ID to delete.

 required
db Session

Database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

404 if not found, 500 if database error.
Source code in backend/app/modules/health/health_sleep/crud.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@core_decorators.handle_db_errors
def delete_health_sleep(user_id: int, health_sleep_id: int, db: Session) -> None:
    """
    Delete a health sleep record for a specific user.

    Args:
        user_id: User ID who owns the sleep record.
        health_sleep_id: Sleep record ID to delete.
        db: Database session.

    Returns:
        None

    Raises:
        HTTPException: 404 if not found, 500 if database error.
    """
    # Get the record to delete
    db_health_sleep = _get_health_sleep_model_by_id_and_user_id_or_404(health_sleep_id, user_id, db)

    # Delete the record
    db.delete(db_health_sleep)
    db.commit()

edit_health_sleep 

edit_health_sleep(user_id, health_sleep, db)

Edit an existing health sleep record for a user.

Parameters:

Name Type Description Default
user_id int

User ID whose sleep record is being edited.

 required
health_sleep HealthSleepUpdate

Updated health sleep data.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthSleepRead

Updated HealthSleepRead schema.

Raises:

Type Description
HTTPException

403 if trying to edit other user record, 404 if not found, 500 if database error.
Source code in backend/app/modules/health/health_sleep/crud.py 
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
@core_decorators.handle_db_errors
def edit_health_sleep(
    user_id: int,
    health_sleep: health_sleep_schema.HealthSleepUpdate,
    db: Session,
) -> health_sleep_schema.HealthSleepRead:
    """
    Edit an existing health sleep record for a user.

    Args:
        user_id: User ID whose sleep record is being edited.
        health_sleep: Updated health sleep data.
        db: Database session.

    Returns:
        Updated HealthSleepRead schema.

    Raises:
        HTTPException: 403 if trying to edit other user record, 404 if not
            found, 500 if database error.
    """
    # Ensure the health_sleep belongs to the user
    if health_sleep.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Cannot edit health sleep for another user.",
        )

    # Get the health_sleep from the database
    db_health_sleep = _get_health_sleep_model_by_id_and_user_id_or_404(health_sleep.id, user_id, db)

    # Dictionary of the fields to update if they are not None
    health_sleep_data = health_sleep.model_dump(exclude_unset=True, mode="json")
    # Iterate over fields and update dynamically
    for key, value in health_sleep_data.items():
        setattr(db_health_sleep, key, value)

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

    return _transform_health_sleep(db_health_sleep)

get_health_sleep_by_date_and_user_id 

get_health_sleep_by_date_and_user_id(user_id, date, db)

Retrieve health sleep record for a user on a specific date.

Parameters:

Name Type Description Default
user_id int

User ID.

 required
date str

Date string for the sleep record.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthSleepRead | None

HealthSleepRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_sleep/crud.py 
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
@core_decorators.handle_db_errors
def get_health_sleep_by_date_and_user_id(
    user_id: int, date: str, db: Session
) -> health_sleep_schema.HealthSleepRead | None:
    """
    Retrieve health sleep record for a user on a specific date.

    Args:
        user_id: User ID.
        date: Date string for the sleep record.
        db: Database session.

    Returns:
        HealthSleepRead schema if found, None otherwise.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the health_sleep from the database
    stmt = select(health_sleep_models.HealthSleep).where(
        health_sleep_models.HealthSleep.date == func.date(date),
        health_sleep_models.HealthSleep.user_id == user_id,
    )
    db_health_sleep = db.execute(stmt).scalar_one_or_none()
    return _transform_health_sleep(db_health_sleep) if db_health_sleep else None

get_health_sleep_by_user_id 

get_health_sleep_by_user_id(user_id, db, page_number=None, num_records=None, interval=None)

Retrieve health sleep records for a specific user with optional pagination and filtering.

Parameters:

Name Type Description Default
user_id int

The ID of the user whose health sleep records are to be retrieved.

 required
db Session

The database session used to execute the query.

 required
page_number int | None

The page number for pagination (1-indexed). If provided, num_records must also be provided. Defaults to None.

 None
num_records int | None

The number of records per page. If provided, page_number must also be provided. Defaults to None.

 None
interval Interval | None

The time interval to filter records. If provided, only records from the start of the interval to present are returned. Defaults to None.

 None

Returns:

Type Description
list[HealthSleepRead]

list[health_sleep_schema.HealthSleepRead]: A list of health sleep records sorted by date in descending order, optionally paginated.
Source code in backend/app/modules/health/health_sleep/crud.py 
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
@core_decorators.handle_db_errors
def get_health_sleep_by_user_id(
    user_id: int,
    db: Session,
    page_number: int | None = None,
    num_records: int | None = None,
    interval: health_constants.Interval | None = None,
) -> list[health_sleep_schema.HealthSleepRead]:
    """
    Retrieve health sleep records for a specific user with optional pagination
        and filtering.

    Args:
        user_id (int): The ID of the user whose health sleep records are to be
            retrieved.
        db (Session): The database session used to execute the query.
        page_number (int | None, optional): The page number for pagination
            (1-indexed).
            If provided, num_records must also be provided. Defaults to None.
        num_records (int | None, optional): The number of records per page.
            If provided, page_number must also be provided. Defaults to None.
        interval (health_constants.Interval | None, optional): The time
            interval to filter records.
            If provided, only records from the start of the interval to present
            are returned. Defaults to None.

    Returns:
        list[health_sleep_schema.HealthSleepRead]: A list of health sleep
            records sorted by date in descending order, optionally paginated.
    """
    # Get the health_sleep from the database
    stmt = select(health_sleep_models.HealthSleep).where(health_sleep_models.HealthSleep.user_id == user_id)

    if interval is not None:
        stmt = stmt.where(
            health_sleep_models.HealthSleep.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    stmt = stmt.order_by(desc(health_sleep_models.HealthSleep.date))

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

    db_health_sleep = db.execute(stmt).scalars().all()

    return _transform_health_sleep(list(db_health_sleep))

get_health_sleep_number_by_user_id 

get_health_sleep_number_by_user_id(user_id, db, interval=None)

Retrieve total count of health sleep records for a user. If interval is provided, count only records starting from the calculated start date.

Parameters:

Name Type Description Default
user_id int

User ID to count records for.

 required
db Session

Database session.

 required
interval Interval | None

Optional filter by goal interval.

 None

Returns:

Type Description
int

Total number of health sleep records.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_sleep/crud.py 
 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
@core_decorators.handle_db_errors
def get_health_sleep_number_by_user_id(
    user_id: int, db: Session, interval: health_constants.Interval | None = None
) -> int:
    """
    Retrieve total count of health sleep records for a user. If interval is
    provided, count only records starting from the calculated start date.

    Args:
        user_id: User ID to count records for.
        db: Database session.
        interval: Optional filter by goal interval.

    Returns:
        Total number of health sleep records.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the number of health_sleep from the database
    stmt = (
        select(func.count())
        .select_from(health_sleep_models.HealthSleep)
        .where(health_sleep_models.HealthSleep.user_id == user_id)
    )

    if interval is not None:
        stmt = stmt.where(
            health_sleep_models.HealthSleep.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    return db.execute(stmt).scalar_one()

Health steps

Health steps module for managing user step count data.

This module provides CRUD operations and data models for user step tracking including daily step counts and data sources.

Exports
  • CRUD: get_health_steps_number_by_user_id, get_health_steps_by_user_id, get_health_steps_by_date_and_user_id, create_health_steps, edit_health_steps, delete_health_steps
  • Schemas: HealthStepsBase, HealthStepsCreate, HealthStepsUpdate, HealthStepsRead, HealthStepsListResponse
  • Enums: Source
  • Models: HealthSteps (ORM model)

HealthStepsBase

Bases: BaseModel

Base model for health steps data.

Represents the core attributes of a user's step count record, including the user reference, date of the record, number of steps taken, and the source of the data.

Attributes:

Name Type Description
date date | None

Calendar date of the steps record. Optional field.
steps StrictInt | None

Number of steps taken. Must be a non-negative integer. Optional field.
source Source | None

Source of the steps data (e.g., device, API, manual entry). Optional field.
Configuration
  • from_attributes: Allows model to be populated from ORM objects.
  • extra: Forbids any extra fields not defined in the model.
  • validate_assignment: Validates field values when assigned after model creation.
  • use_enum_values: Uses enum values instead of enum objects in serialization.
Source code in backend/app/modules/health/health_steps/schema.py 
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
class HealthStepsBase(BaseModel):
    """
    Base model for health steps data.

    Represents the core attributes of a user's step count record, including the user reference,
    date of the record, number of steps taken, and the source of the data.

    Attributes:
        date (datetime_date | None): Calendar date of the steps record. Optional field.
        steps (StrictInt | None): Number of steps taken. Must be a non-negative integer. Optional field.
        source (Source | None): Source of the steps data (e.g., device, API, manual entry). Optional field.

    Configuration:
        - from_attributes: Allows model to be populated from ORM objects.
        - extra: Forbids any extra fields not defined in the model.
        - validate_assignment: Validates field values when assigned after model creation.
        - use_enum_values: Uses enum values instead of enum objects in serialization.
    """

    date: datetime_date | None = Field(default=None, description="Calendar date of the steps")
    steps: StrictInt | None = Field(default=None, ge=0, description="Number of steps taken")
    source: Source | None = Field(default=None, description="Source of the steps data")

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

HealthStepsCreate

Bases: HealthStepsBase

Validator for HealthStepsCreate model that automatically sets the date field.

This validator runs after model initialization to ensure that if no date is provided, it defaults to today's date.

Source code in backend/app/modules/health/health_steps/schema.py 
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
class HealthStepsCreate(HealthStepsBase):
    """
    Validator for HealthStepsCreate model that automatically sets the date field.

    This validator runs after model initialization to ensure that if no date
    is provided, it defaults to today's date.
    """

    @model_validator(mode="after")
    def set_default_date(self) -> "HealthStepsCreate":
        """
        Set date to today if not provided.

        Clients should always send their own local date — the request carries no
        timezone, so this fallback can only use the server's configured zone and
        will file an entry logged near local midnight under the wrong day for an
        athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
        than ``date.today()``, which silently depends on the container's clock.

        Returns:
            The validated model instance with date set.
        """
        if self.date is None:
            self.date = core_timezone.today_in(core_config.settings.TZ)
        return self

set_default_date 

set_default_date()

Set date to today if not provided.

Clients should always send their own local date — the request carries no timezone, so this fallback can only use the server's configured zone and will file an entry logged near local midnight under the wrong day for an athlete elsewhere. Resolved through settings.TZ explicitly rather than date.today(), which silently depends on the container's clock.

Returns:

Type Description
HealthStepsCreate

The validated model instance with date set.
Source code in backend/app/modules/health/health_steps/schema.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@model_validator(mode="after")
def set_default_date(self) -> "HealthStepsCreate":
    """
    Set date to today if not provided.

    Clients should always send their own local date — the request carries no
    timezone, so this fallback can only use the server's configured zone and
    will file an entry logged near local midnight under the wrong day for an
    athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
    than ``date.today()``, which silently depends on the container's clock.

    Returns:
        The validated model instance with date set.
    """
    if self.date is None:
        self.date = core_timezone.today_in(core_config.settings.TZ)
    return self

HealthStepsListResponse

Bases: HealthListResponse

Response model for listing health steps records.

Extends the base HealthListResponse to provide a typed response containing a list of steps records. This model is used when returning multiple steps entries from API endpoints.

Attributes:

Name Type Description
records list[HealthStepsRead]

A list of HealthStepsRead objects representing individual steps records.
Source code in backend/app/modules/health/health_steps/schema.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class HealthStepsListResponse(health_schema.HealthListResponse):
    """
    Response model for listing health steps records.

    Extends the base HealthListResponse to provide a typed response containing
    a list of steps records. This model is used when returning multiple steps
    entries from API endpoints.

    Attributes:
        records: A list of HealthStepsRead objects representing individual
            steps records.
    """

    records: list[HealthStepsRead] = Field(..., description="List of health steps records")

HealthStepsRead

Bases: HealthStepsBase

Schema for reading health steps records.

Extends the base health steps schema with an identifier field for retrieving or referencing existing steps records in the database.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the steps record to update. Required field.
user_id StrictInt

Foreign key reference to the user. Required field.
Source code in backend/app/modules/health/health_steps/schema.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class HealthStepsRead(HealthStepsBase):
    """
    Schema for reading health steps records.

    Extends the base health steps schema with an identifier field for retrieving
    or referencing existing steps records in the database.

    Attributes:
        id (StrictInt): Unique identifier for the steps record to update. Required field.
        user_id (StrictInt): Foreign key reference to the user. Required field.
    """

    id: StrictInt = Field(..., description="Unique identifier for the steps record to update")
    user_id: StrictInt = Field(..., description="Foreign key reference to the user")

HealthStepsUpdate

Bases: HealthStepsRead

Schema for updating health steps records.

Inherits from HealthStepsRead to maintain consistency with read operations while allowing modifications to health steps data. This schema is used for PUT/PATCH requests to update existing health steps entries.

Source code in backend/app/modules/health/health_steps/schema.py 
 96
 97
 98
 99
100
101
102
103
class HealthStepsUpdate(HealthStepsRead):
    """
    Schema for updating health steps records.

    Inherits from HealthStepsRead to maintain consistency with read operations
    while allowing modifications to health steps data. This schema is used for
    PUT/PATCH requests to update existing health steps entries.
    """

Source

Bases: Enum

An enumeration representing supported sources for the application.

Members

GARMIN: Garmin health data source

Source code in backend/app/modules/health/health_steps/schema.py 
11
12
13
14
15
16
17
18
19
class Source(Enum):
    """
    An enumeration representing supported sources for the application.

    Members:
        GARMIN: Garmin health data source
    """

    GARMIN = "garmin"

create_health_steps 

create_health_steps(user_id, health_steps, db)

Create a new health steps record for a user.

Parameters:

Name Type Description Default
user_id int

User ID for the record owner.

 required
health_steps HealthStepsCreate

Health steps data to create.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthStepsRead

Created health steps record.

Raises:

Type Description
HTTPException

If duplicate entry or database error.
Source code in backend/app/modules/health/health_steps/crud.py 
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
@core_decorators.handle_db_errors
def create_health_steps(
    user_id: int,
    health_steps: health_steps_schema.HealthStepsCreate,
    db: Session,
) -> health_steps_schema.HealthStepsRead:
    """
    Create a new health steps record for a user.

    Args:
        user_id: User ID for the record owner.
        health_steps: Health steps data to create.
        db: Database session.

    Returns:
        Created health steps record.

    Raises:
        HTTPException: If duplicate entry or database error.
    """
    try:
        # Create a new health_steps
        db_health_steps = health_steps_models.HealthSteps(
            **health_steps.model_dump(exclude_none=False),
            user_id=user_id,
        )

        # Add the health_steps to the database
        db.add(db_health_steps)
        db.commit()
        db.refresh(db_health_steps)

        # Return the health_steps
        return _transform_health_steps(db_health_steps)
    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=(f"Duplicate entry error. Check if there is already a entry created for {health_steps.date}"),
        ) from integrity_error

delete_health_steps 

delete_health_steps(user_id, health_steps_id, db)

Delete a health steps record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the record.

 required
health_steps_id int

Health steps record ID to delete.

 required
db Session

Database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

If record not found or database error.
Source code in backend/app/modules/health/health_steps/crud.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
@core_decorators.handle_db_errors
def delete_health_steps(user_id: int, health_steps_id: int, db: Session) -> None:
    """
    Delete a health steps record for a user.

    Args:
        user_id: User ID who owns the record.
        health_steps_id: Health steps record ID to delete.
        db: Database session.

    Returns:
        None

    Raises:
        HTTPException: If record not found or database error.
    """
    # Get the record first to ensure it exists
    db_health_steps = _get_health_steps_model_by_id_and_user_id_or_404(health_steps_id, user_id, db)

    # Delete the health_steps
    db.delete(db_health_steps)
    # Commit the transaction
    db.commit()

edit_health_steps 

edit_health_steps(user_id, health_steps, db)

Edit health steps record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the record.

 required
health_steps HealthStepsUpdate

Health steps data to update.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthStepsRead

Updated HealthStepsRead schema.

Raises:

Type Description
HTTPException

403 if trying to edit other user record, 404 if not found, 500 if database error.
Source code in backend/app/modules/health/health_steps/crud.py 
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
@core_decorators.handle_db_errors
def edit_health_steps(
    user_id: int,
    health_steps: health_steps_schema.HealthStepsUpdate,
    db: Session,
) -> health_steps_schema.HealthStepsRead:
    """
    Edit health steps record for a user.

    Args:
        user_id: User ID who owns the record.
        health_steps: Health steps data to update.
        db: Database session.

    Returns:
        Updated HealthStepsRead schema.

    Raises:
        HTTPException: 403 if trying to edit other user record, 404 if not
            found, 500 if database error.
    """
    # Ensure the health_steps belongs to the user
    if health_steps.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Cannot edit health steps for another user.",
        )

    # Get the health_steps from the database
    db_health_steps = _get_health_steps_model_by_id_and_user_id_or_404(health_steps.id, user_id, db)

    # Dictionary of the fields to update if they are not None
    health_steps_data = health_steps.model_dump(exclude_unset=True)
    # Iterate over the fields and update the db_health_steps dynamically
    for key, value in health_steps_data.items():
        setattr(db_health_steps, key, value)

    # Commit the transaction
    db.commit()
    # Refresh the object to ensure it reflects database state
    db.refresh(db_health_steps)

    return _transform_health_steps(db_health_steps)

get_health_steps_by_date_and_user_id 

get_health_steps_by_date_and_user_id(user_id, date, db)

Retrieve health steps record for a user on a specific date.

Parameters:

Name Type Description Default
user_id int

User ID.

 required
date str

Date string for the steps record.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthStepsRead | None

HealthStepsRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_steps/crud.py 
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
@core_decorators.handle_db_errors
def get_health_steps_by_date_and_user_id(
    user_id: int, date: str, db: Session
) -> health_steps_schema.HealthStepsRead | None:
    """
    Retrieve health steps record for a user on a specific date.

    Args:
        user_id: User ID.
        date: Date string for the steps record.
        db: Database session.

    Returns:
        HealthStepsRead schema if found, None otherwise.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the health_steps from the database
    stmt = select(health_steps_models.HealthSteps).where(
        health_steps_models.HealthSteps.date == func.date(date),
        health_steps_models.HealthSteps.user_id == user_id,
    )
    db_health_steps = db.execute(stmt).scalar_one_or_none()
    return _transform_health_steps(db_health_steps) if db_health_steps else None

get_health_steps_by_user_id 

get_health_steps_by_user_id(user_id, db, page_number=None, num_records=None, interval=None)

Retrieve health steps records for a specific user with optional pagination and filtering.

Parameters:

Name Type Description Default
user_id int

The ID of the user whose health steps records are to be retrieved.

 required
db Session

The database session used to execute the query.

 required
page_number int | None

The page number for pagination (1-indexed). If provided, num_records must also be provided. Defaults to None.

 None
num_records int | None

The number of records per page. If provided, page_number must also be provided. Defaults to None.

 None
interval Interval | None

The time interval to filter records. If provided, only records from the start of the interval to present are returned. Defaults to None.

 None

Returns:

Type Description
list[HealthStepsRead]

list[health_steps_schema.HealthStepsRead]: A list of health steps records sorted by date in descending order, optionally paginated.
Source code in backend/app/modules/health/health_steps/crud.py 
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
@core_decorators.handle_db_errors
def get_health_steps_by_user_id(
    user_id: int,
    db: Session,
    page_number: int | None = None,
    num_records: int | None = None,
    interval: health_constants.Interval | None = None,
) -> list[health_steps_schema.HealthStepsRead]:
    """
    Retrieve health steps records for a specific user with optional pagination
        and filtering.

    Args:
        user_id (int): The ID of the user whose health steps records are to be
            retrieved.
        db (Session): The database session used to execute the query.
        page_number (int | None, optional): The page number for pagination
            (1-indexed).
            If provided, num_records must also be provided. Defaults to None.
        num_records (int | None, optional): The number of records per page.
            If provided, page_number must also be provided. Defaults to None.
        interval (health_constants.Interval | None, optional): The time
            interval to filter records.
            If provided, only records from the start of the interval to present
            are returned. Defaults to None.

    Returns:
        list[health_steps_schema.HealthStepsRead]: A list of health steps
            records sorted by date in descending order, optionally paginated.
    """
    # Get the health_steps from the database
    stmt = select(health_steps_models.HealthSteps).where(health_steps_models.HealthSteps.user_id == user_id)

    if interval is not None:
        stmt = stmt.where(
            health_steps_models.HealthSteps.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    stmt = stmt.order_by(desc(health_steps_models.HealthSteps.date))

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

    db_health_steps_list = db.execute(stmt).scalars().all()
    return _transform_health_steps(list(db_health_steps_list))

get_health_steps_number_by_user_id 

get_health_steps_number_by_user_id(user_id, db, interval=None)

Retrieve total count of health steps records for a user. If interval is provided, count only records starting from the calculated start date.

Parameters:

Name Type Description Default
user_id int

User ID to count records for.

 required
db Session

Database session.

 required
interval Interval | None

Optional filter by goal interval.

 None

Returns:

Type Description
int

Total number of health steps records.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_steps/crud.py 
 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
@core_decorators.handle_db_errors
def get_health_steps_number_by_user_id(
    user_id: int, db: Session, interval: health_constants.Interval | None = None
) -> int:
    """
    Retrieve total count of health steps records for a user. If interval is
    provided, count only records starting from the calculated start date.

    Args:
        user_id: User ID to count records for.
        db: Database session.
        interval: Optional filter by goal interval.

    Returns:
        Total number of health steps records.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the number of health_steps from the database
    stmt = (
        select(func.count())
        .select_from(health_steps_models.HealthSteps)
        .where(health_steps_models.HealthSteps.user_id == user_id)
    )

    if interval is not None:
        stmt = stmt.where(
            health_steps_models.HealthSteps.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    return db.execute(stmt).scalar_one()

Health targets

Health targets module for managing user health goals.

This module provides CRUD operations and data models for user health targets including weight, steps, and sleep goals.

Exports
  • CRUD: get_health_targets_by_user_id, create_health_targets, edit_health_target
  • Schemas: HealthTargetsBase, HealthTargetsUpdate, HealthTargetsRead

HealthTargetsBase

Bases: BaseModel

Base schema for health targets with shared fields.

Attributes:

Name Type Description
weight StrictFloat | None

Target weight in kg.
steps StrictInt | None

Target daily steps count.
sleep StrictInt | None

Target sleep duration in seconds.
fasting StrictInt | None

Target fasting duration in seconds.
water_ml StrictFloat | None

Target daily water intake in milliliters.
poop_count StrictInt | None

Target daily bowel movement count.
Source code in backend/app/modules/health/health_targets/schema.py 
 4
 5
 6
 7
 8
 9
10
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
class HealthTargetsBase(BaseModel):
    """
    Base schema for health targets with shared fields.

    Attributes:
        weight: Target weight in kg.
        steps: Target daily steps count.
        sleep: Target sleep duration in seconds.
        fasting: Target fasting duration in seconds.
        water_ml: Target daily water intake in milliliters.
        poop_count: Target daily bowel movement count.
    """

    weight: StrictFloat | None = Field(default=None, gt=0, le=500, description="Target weight in kg")
    steps: StrictInt | None = Field(default=None, ge=0, description="Target daily steps count")
    sleep: StrictInt | None = Field(default=None, ge=0, le=86400, description="Target sleep duration in seconds")
    fasting: StrictInt | None = Field(
        default=None,
        ge=0,
        le=259200,
        description="Target fasting duration in seconds (max 72h)",
    )
    water_ml: StrictFloat | None = Field(
        default=None,
        ge=0,
        le=20000,
        description=("Target daily water intake in milliliters"),
    )
    poop_count: StrictInt | None = Field(
        default=None,
        ge=0,
        le=20,
        description="Target daily bowel movement count",
    )

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

HealthTargetsRead

Bases: HealthTargetsBase

Schema for reading health targets.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the health target record.
user_id StrictInt

Foreign key reference to the user.
Source code in backend/app/modules/health/health_targets/schema.py 
46
47
48
49
50
51
52
53
54
55
56
class HealthTargetsRead(HealthTargetsBase):
    """
    Schema for reading health targets.

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

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

HealthTargetsUpdate

Bases: HealthTargetsRead

Schema for updating health targets.

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

Source code in backend/app/modules/health/health_targets/schema.py 
59
60
61
62
63
64
65
class HealthTargetsUpdate(HealthTargetsRead):
    """
    Schema for updating health targets.

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

create_health_targets 

create_health_targets(user_id, db)

Create new health targets for a user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to create targets for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
HealthTargetsRead

The created HealthTargetsRead schema.

Raises:

Type Description
HTTPException

409 error if targets already exist.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/health/health_targets/crud.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
@core_decorators.handle_db_errors
def create_health_targets(user_id: int, db: Session) -> health_targets_schema.HealthTargetsRead:
    """
    Create new health targets for a user.

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

    Returns:
        The created HealthTargetsRead schema.

    Raises:
        HTTPException: 409 error if targets already exist.
        HTTPException: 500 error if database operation fails.
    """
    try:
        # Create a new health_target
        db_health_targets = health_targets_models.HealthTargets(
            user_id=user_id,
        )

        # Add the health_targets to the database
        db.add(db_health_targets)
        db.commit()
        db.refresh(db_health_targets)

        # Return the health_targets
        return _transform_health_targets(db_health_targets)
    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 there is already an entry created for the user",
        ) from integrity_error

edit_health_target 

edit_health_target(health_target, user_id, db)

Update health targets for a specific user.

Parameters:

Name Type Description Default
health_target HealthTargetsUpdate

Schema with fields to update.

 required
user_id int

The ID of the user to update targets for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
HealthTargetsRead

The updated HealthTargetsRead schema.

Raises:

Type Description
HTTPException

404 error if targets not found.
HTTPException

500 error if database operation fails.
Source code in backend/app/modules/health/health_targets/crud.py 
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
@core_decorators.handle_db_errors
def edit_health_target(
    health_target: health_targets_schema.HealthTargetsUpdate,
    user_id: int,
    db: Session,
) -> health_targets_schema.HealthTargetsRead:
    """
    Update health targets for a specific user.

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

    Returns:
        The updated HealthTargetsRead schema.

    Raises:
        HTTPException: 404 error if targets not found.
        HTTPException: 500 error if database operation fails.
    """
    # Get the user health target from the database
    db_health_target = _get_health_targets_model_by_user_id_or_404(user_id=user_id, db=db)

    # Dictionary of the fields to update if they are not None
    health_target_data = health_target.model_dump(exclude_unset=True)
    # Iterate over the fields and update dynamically
    for key, value in health_target_data.items():
        setattr(db_health_target, key, value)

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

    return _transform_health_targets(db_health_target)

get_health_targets_by_user_id 

get_health_targets_by_user_id(user_id, db)

Retrieve health targets for a specific user.

Parameters:

Name Type Description Default
user_id int

The ID of the user to fetch targets for.

 required
db Session

SQLAlchemy database session.

 required

Returns:

Type Description
HealthTargetsRead | None

The HealthTargetsRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

500 error if database query fails.
Source code in backend/app/modules/health/health_targets/crud.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@core_decorators.handle_db_errors
def get_health_targets_by_user_id(user_id: int, db: Session) -> health_targets_schema.HealthTargetsRead | None:
    """
    Retrieve health targets for a specific user.

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

    Returns:
        The HealthTargetsRead schema if found, None otherwise.

    Raises:
        HTTPException: 500 error if database query fails.
    """
    # Get the health_targets from the database
    stmt = select(health_targets_models.HealthTargets).where(health_targets_models.HealthTargets.user_id == user_id)
    db_health_targets = db.execute(stmt).scalar_one_or_none()

    return _transform_health_targets(db_health_targets) if db_health_targets else None

Health water

Health water intake module for managing user hydration data.

This module provides CRUD operations and data models for user water intake tracking including daily consumption amounts and data sources.

Exports
  • CRUD: get_health_water_number_by_user_id, get_health_water_by_id_and_user_id, get_health_water_by_user_id, get_health_water_by_date_and_user_id, create_health_water, edit_health_water, delete_health_water
  • Schemas: HealthWaterBase, HealthWaterCreate, HealthWaterUpdate, HealthWaterRead, HealthWaterListResponse
  • Enums: Source

HealthWaterBase

Bases: BaseModel

Base model for health water intake data.

Attributes:

Name Type Description
date date | None

Calendar date of the water intake record.
amount_ml StrictFloat | None

Water consumed in milliliters.
source Source | None

Source of the water intake data.
Source code in backend/app/modules/health/health_water/schema.py 
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
class HealthWaterBase(BaseModel):
    """
    Base model for health water intake data.

    Attributes:
        date: Calendar date of the water intake record.
        amount_ml: Water consumed in milliliters.
        source: Source of the water intake data.
    """

    date: datetime_date | None = Field(
        default=None,
        description="Calendar date of the water intake",
    )
    amount_ml: StrictFloat | None = Field(
        default=None,
        ge=0,
        le=20000,
        description="Water consumed in milliliters",
    )
    source: Source | None = Field(
        default=None,
        description="Source of the water intake data",
    )

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

HealthWaterCreate

Bases: HealthWaterBase

Schema for creating a new water intake record.

Automatically sets the date to today if not provided.

Source code in backend/app/modules/health/health_water/schema.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class HealthWaterCreate(HealthWaterBase):
    """
    Schema for creating a new water intake record.

    Automatically sets the date to today if not provided.
    """

    @model_validator(mode="after")
    def set_default_date(self) -> "HealthWaterCreate":
        """
        Set date to today if not provided.

        Clients should always send their own local date — the request carries no
        timezone, so this fallback can only use the server's configured zone and
        will file an entry logged near local midnight under the wrong day for an
        athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
        than ``date.today()``, which silently depends on the container's clock.

        Returns:
            The validated model instance with date set.
        """
        if self.date is None:
            self.date = core_timezone.today_in(core_config.settings.TZ)
        return self

set_default_date 

set_default_date()

Set date to today if not provided.

Clients should always send their own local date — the request carries no timezone, so this fallback can only use the server's configured zone and will file an entry logged near local midnight under the wrong day for an athlete elsewhere. Resolved through settings.TZ explicitly rather than date.today(), which silently depends on the container's clock.

Returns:

Type Description
HealthWaterCreate

The validated model instance with date set.
Source code in backend/app/modules/health/health_water/schema.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@model_validator(mode="after")
def set_default_date(self) -> "HealthWaterCreate":
    """
    Set date to today if not provided.

    Clients should always send their own local date — the request carries no
    timezone, so this fallback can only use the server's configured zone and
    will file an entry logged near local midnight under the wrong day for an
    athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
    than ``date.today()``, which silently depends on the container's clock.

    Returns:
        The validated model instance with date set.
    """
    if self.date is None:
        self.date = core_timezone.today_in(core_config.settings.TZ)
    return self

HealthWaterListResponse

Bases: HealthListResponse

Response model for listing health water intake records.

Attributes:

Name Type Description
records list[HealthWaterRead]

A list of HealthWaterRead objects representing individual water intake records.
Source code in backend/app/modules/health/health_water/schema.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
class HealthWaterListResponse(health_schema.HealthListResponse):
    """
    Response model for listing health water intake records.

    Attributes:
        records: A list of HealthWaterRead objects representing
            individual water intake records.
    """

    records: list[HealthWaterRead] = Field(
        ...,
        description="List of health water intake records",
    )

HealthWaterRead

Bases: HealthWaterBase

Schema for reading a water intake record.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the water intake record.
user_id StrictInt

Foreign key reference to the user.
Source code in backend/app/modules/health/health_water/schema.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class HealthWaterRead(HealthWaterBase):
    """
    Schema for reading a water intake record.

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

    id: StrictInt = Field(
        ...,
        description=("Unique identifier for the water intake record"),
    )
    user_id: StrictInt = Field(
        ...,
        description="Foreign key reference to the user",
    )

HealthWaterUpdate

Bases: HealthWaterRead

Schema for updating a water intake record.

Inherits from HealthWaterRead for consistency with read operations while allowing modifications to water intake data.

Source code in backend/app/modules/health/health_water/schema.py 
116
117
118
119
120
121
122
class HealthWaterUpdate(HealthWaterRead):
    """
    Schema for updating a water intake record.

    Inherits from HealthWaterRead for consistency with read
    operations while allowing modifications to water intake data.
    """

Source

Bases: Enum

Enumeration of data sources for water intake records.

Attributes:

Name Type Description
MANUAL

Manually entered water intake data.
GARMIN

Garmin fitness tracking platform as a data source.
Source code in backend/app/modules/health/health_water/schema.py 
25
26
27
28
29
30
31
32
33
34
35
class Source(Enum):
    """
    Enumeration of data sources for water intake records.

    Attributes:
        MANUAL: Manually entered water intake data.
        GARMIN: Garmin fitness tracking platform as a data source.
    """

    MANUAL = "manual"
    GARMIN = "garmin"

create_health_water 

create_health_water(user_id, health_water, db)

Create a new water intake record for a user.

Parameters:

Name Type Description Default
user_id int

User ID for the record owner.

 required
health_water HealthWaterCreate

Water intake data to create.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthWaterRead

Created water intake record.

Raises:

Type Description
HTTPException

If duplicate entry or database error.
Source code in backend/app/modules/health/health_water/crud.py 
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
@core_decorators.handle_db_errors
def create_health_water(
    user_id: int,
    health_water: health_water_schema.HealthWaterCreate,
    db: Session,
) -> health_water_schema.HealthWaterRead:
    """
    Create a new water intake record for a user.

    Args:
        user_id: User ID for the record owner.
        health_water: Water intake data to create.
        db: Database session.

    Returns:
        Created water intake record.

    Raises:
        HTTPException: If duplicate entry or database error.
    """
    try:
        db_health_water = health_water_models.HealthWater(
            **health_water.model_dump(exclude_none=False),
            user_id=user_id,
        )

        db.add(db_health_water)
        db.commit()
        db.refresh(db_health_water)

        return _transform_health_water(db_health_water)
    except IntegrityError as integrity_error:
        db.rollback()

        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=(f"Duplicate entry error. Check if there is already an entry created for {health_water.date}"),
        ) from integrity_error

delete_health_water 

delete_health_water(user_id, health_water_id, db)

Delete a water intake record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the record.

 required
health_water_id int

Water intake record ID to delete.

 required
db Session

Database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

If record not found or database error.
Source code in backend/app/modules/health/health_water/crud.py 
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
@core_decorators.handle_db_errors
def delete_health_water(user_id: int, health_water_id: int, db: Session) -> None:
    """
    Delete a water intake record for a user.

    Args:
        user_id: User ID who owns the record.
        health_water_id: Water intake record ID to delete.
        db: Database session.

    Returns:
        None

    Raises:
        HTTPException: If record not found or database error.
    """
    db_health_water = _get_health_water_model_by_id_and_user_id_or_404(health_water_id, user_id, db)

    db.delete(db_health_water)
    db.commit()

edit_health_water 

edit_health_water(user_id, health_water, db)

Edit water intake record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the record.

 required
health_water HealthWaterUpdate

Water intake data to update.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthWaterRead

Updated HealthWaterRead schema.

Raises:

Type Description
HTTPException

403 if editing another user's record, 404 if not found, 500 if database error.
Source code in backend/app/modules/health/health_water/crud.py 
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
@core_decorators.handle_db_errors
def edit_health_water(
    user_id: int,
    health_water: health_water_schema.HealthWaterUpdate,
    db: Session,
) -> health_water_schema.HealthWaterRead:
    """
    Edit water intake record for a user.

    Args:
        user_id: User ID who owns the record.
        health_water: Water intake data to update.
        db: Database session.

    Returns:
        Updated HealthWaterRead schema.

    Raises:
        HTTPException: 403 if editing another user's record,
            404 if not found, 500 if database error.
    """
    if health_water.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=("Cannot edit health water intake for another user."),
        )

    db_health_water = _get_health_water_model_by_id_and_user_id_or_404(health_water.id, user_id, db)

    health_water_data = health_water.model_dump(exclude_unset=True)
    for key, value in health_water_data.items():
        setattr(db_health_water, key, value)

    db.commit()
    db.refresh(db_health_water)

    return _transform_health_water(db_health_water)

get_health_water_by_date_and_user_id 

get_health_water_by_date_and_user_id(user_id, date, db)

Retrieve water intake record for a user on a specific date.

Parameters:

Name Type Description Default
user_id int

User ID.

 required
date str

Date string for the water intake record.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthWaterRead | None

HealthWaterRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_water/crud.py 
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
@core_decorators.handle_db_errors
def get_health_water_by_date_and_user_id(
    user_id: int, date: str, db: Session
) -> health_water_schema.HealthWaterRead | None:
    """
    Retrieve water intake record for a user on a specific date.

    Args:
        user_id: User ID.
        date: Date string for the water intake record.
        db: Database session.

    Returns:
        HealthWaterRead schema if found, None otherwise.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = select(health_water_models.HealthWater).where(
        health_water_models.HealthWater.date == func.date(date),
        health_water_models.HealthWater.user_id == user_id,
    )
    db_health_water = db.execute(stmt).scalar_one_or_none()

    return _transform_health_water(db_health_water) if db_health_water else None

get_health_water_by_id_and_user_id 

get_health_water_by_id_and_user_id(health_water_id, user_id, db)

Retrieve water intake record by ID and user ID.

Parameters:

Name Type Description Default
health_water_id int

Water intake record ID to fetch.

 required
user_id int

User ID to fetch record for.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthWaterRead | None

HealthWaterRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_water/crud.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
@core_decorators.handle_db_errors
def get_health_water_by_id_and_user_id(
    health_water_id: int, user_id: int, db: Session
) -> health_water_schema.HealthWaterRead | None:
    """
    Retrieve water intake record by ID and user ID.

    Args:
        health_water_id: Water intake record ID to fetch.
        user_id: User ID to fetch record for.
        db: Database session.

    Returns:
        HealthWaterRead schema if found, None otherwise.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = select(health_water_models.HealthWater).where(
        health_water_models.HealthWater.id == health_water_id,
        health_water_models.HealthWater.user_id == user_id,
    )
    db_health_water = db.execute(stmt).scalar_one_or_none()
    return _transform_health_water(db_health_water) if db_health_water else None

get_health_water_by_user_id 

get_health_water_by_user_id(user_id, db, page_number=None, num_records=None, interval=None)

Retrieve water intake records for a user with optional pagination and filtering.

Parameters:

Name Type Description Default
user_id int

User ID whose records are to be retrieved.

 required
db Session

Database session used to execute the query.

 required
page_number int | None

Page number for pagination (1-indexed).

 None
num_records int | None

Number of records per page.

 None
interval Interval | None

Time interval to filter records.

 None

Returns:

Type Description
list[HealthWaterRead]

List of HealthWaterRead schema records sorted by date descending.
Source code in backend/app/modules/health/health_water/crud.py 
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
@core_decorators.handle_db_errors
def get_health_water_by_user_id(
    user_id: int,
    db: Session,
    page_number: int | None = None,
    num_records: int | None = None,
    interval: health_constants.Interval | None = None,
) -> list[health_water_schema.HealthWaterRead]:
    """
    Retrieve water intake records for a user with optional
    pagination and filtering.

    Args:
        user_id: User ID whose records are to be retrieved.
        db: Database session used to execute the query.
        page_number: Page number for pagination (1-indexed).
        num_records: Number of records per page.
        interval: Time interval to filter records.

    Returns:
        List of HealthWaterRead schema records sorted by date descending.
    """
    stmt = select(health_water_models.HealthWater).where(health_water_models.HealthWater.user_id == user_id)

    if interval is not None:
        stmt = stmt.where(
            health_water_models.HealthWater.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    stmt = stmt.order_by(desc(health_water_models.HealthWater.date))

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

    db_health_water = db.execute(stmt).scalars().all()

    return _transform_health_water(list(db_health_water))

get_health_water_number_by_user_id 

get_health_water_number_by_user_id(user_id, db, interval=None)

Retrieve total count of water intake records for a user.

If interval is provided, count only records starting from the calculated start date.

Parameters:

Name Type Description Default
user_id int

User ID to count records for.

 required
db Session

Database session.

 required
interval Interval | None

Optional filter by goal interval.

 None

Returns:

Type Description
int

Total number of health water intake records.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_water/crud.py 
 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
@core_decorators.handle_db_errors
def get_health_water_number_by_user_id(
    user_id: int,
    db: Session,
    interval: health_constants.Interval | None = None,
) -> int:
    """
    Retrieve total count of water intake records for a user.

    If interval is provided, count only records starting from
    the calculated start date.

    Args:
        user_id: User ID to count records for.
        db: Database session.
        interval: Optional filter by goal interval.

    Returns:
        Total number of health water intake records.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = (
        select(func.count())
        .select_from(health_water_models.HealthWater)
        .where(health_water_models.HealthWater.user_id == user_id)
    )

    if interval is not None:
        stmt = stmt.where(
            health_water_models.HealthWater.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    return db.execute(stmt).scalar_one()

Health weight

Health weight module for managing user weight and body composition.

This module provides CRUD operations and data models for user weight tracking including BMI, body composition metrics, and various health indicators.

Exports
  • CRUD: get_all_health_weight, get_health_weight_number_by_user_id, get_all_health_weight_by_user_id, get_health_weight_by_user_id, get_health_weight_by_date_and_user_id, create_health_weight, edit_health_weight, delete_health_weight
  • Schemas: HealthWeightBase, HealthWeightCreate, HealthWeightUpdate, HealthWeightRead, HealthWeightListResponse
  • Enums: Source
  • Utils: calculate_bmi, calculate_bmi_all_user_entries

HealthWeightBase

Bases: BaseModel

Pydantic model for health weight data.

This model defines the structure and validation rules for health weight information, including body composition metrics and related health indicators.

Attributes:

Name Type Description
date date | None

The date of the health weight measurement.
weight StrictFloat | None

Weight in kilograms. Must be between 0 and 500.
bmi StrictFloat | None

Body Mass Index value. Must be between 0 and 100.
body_fat StrictFloat | None

Body fat percentage. Must be between 0 and 100.
body_water StrictFloat | None

Body water percentage. Must be between 0 and 100.
bone_mass StrictFloat | None

Bone mass in kilograms. Must be between 0 and 500.
muscle_mass StrictFloat | None

Muscle mass in kilograms. Must be between 0 and 500.
physique_rating StrictInt | None

Physique rating score. Must be greater than or equal to 0.
visceral_fat StrictFloat | None

Visceral fat percentage. Must be between 0 and 100.
metabolic_age StrictInt | None

Metabolic age in years. Must be between 0 and 120.
source Source | None

The source or device from which the weight data was obtained.
Model Configuration
  • Populates from ORM attributes
  • Forbids extra fields
  • Validates assignments
  • Uses enum values for serialization
Source code in backend/app/modules/health/health_weight/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
class HealthWeightBase(BaseModel):
    """
    Pydantic model for health weight data.

    This model defines the structure and validation rules for health weight information,
    including body composition metrics and related health indicators.

    Attributes:
        date (datetime_date | None): The date of the health weight measurement.
        weight (StrictFloat | None): Weight in kilograms. Must be between 0 and 500.
        bmi (StrictFloat | None): Body Mass Index value. Must be between 0 and 100.
        body_fat (StrictFloat | None): Body fat percentage. Must be between 0 and 100.
        body_water (StrictFloat | None): Body water percentage. Must be between 0 and 100.
        bone_mass (StrictFloat | None): Bone mass in kilograms. Must be between 0 and 500.
        muscle_mass (StrictFloat | None): Muscle mass in kilograms. Must be between 0 and 500.
        physique_rating (StrictInt | None): Physique rating score. Must be greater than or equal to 0.
        visceral_fat (StrictFloat | None): Visceral fat percentage. Must be between 0 and 100.
        metabolic_age (StrictInt | None): Metabolic age in years. Must be between 0 and 120.
        source (Source | None): The source or device from which the weight data was obtained.

    Model Configuration:
        - Populates from ORM attributes
        - Forbids extra fields
        - Validates assignments
        - Uses enum values for serialization
    """

    date: datetime_date | None = Field(default=None, description="Health weight date (date)")
    weight: StrictFloat | None = Field(default=None, ge=0, le=500, description="Weight in kilograms")
    bmi: StrictFloat | None = Field(default=None, ge=0, le=100, description="Body Mass Index")
    body_fat: StrictFloat | None = Field(default=None, ge=0, le=100, description="Body fat percentage")
    body_water: StrictFloat | None = Field(default=None, ge=0, le=100, description="Body water percentage")
    bone_mass: StrictFloat | None = Field(default=None, ge=0, le=500, description="Bone mass in kilograms")
    muscle_mass: StrictFloat | None = Field(default=None, ge=0, le=500, description="Muscle mass in kilograms")
    physique_rating: StrictInt | None = Field(default=None, ge=0, description="Physique rating")
    visceral_fat: StrictFloat | None = Field(default=None, ge=0, le=100, description="Visceral fat")
    metabolic_age: StrictInt | None = Field(default=None, ge=0, le=120, description="Metabolic age")
    source: Source | None = Field(default=None, description="Source of the weight data")

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

HealthWeightCreate

Bases: HealthWeightBase

Validator for HealthWeightCreate model that automatically sets the date field.

This validator runs after model initialization to ensure that if no date is provided, it defaults to today's date.

Source code in backend/app/modules/health/health_weight/schema.py 
 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
class HealthWeightCreate(HealthWeightBase):
    """
    Validator for HealthWeightCreate model that automatically sets the date field.

    This validator runs after model initialization to ensure that if no date
    is provided, it defaults to today's date.
    """

    @model_validator(mode="after")
    def set_default_date(self) -> "HealthWeightCreate":
        """
        Set date to today if not provided.

        Clients should always send their own local date — the request carries no
        timezone, so this fallback can only use the server's configured zone and
        will file an entry logged near local midnight under the wrong day for an
        athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
        than ``date.today()``, which silently depends on the container's clock.

        Returns:
            The validated model instance with date set.
        """
        if self.date is None:
            self.date = core_timezone.today_in(core_config.settings.TZ)
        return self

set_default_date 

set_default_date()

Set date to today if not provided.

Clients should always send their own local date — the request carries no timezone, so this fallback can only use the server's configured zone and will file an entry logged near local midnight under the wrong day for an athlete elsewhere. Resolved through settings.TZ explicitly rather than date.today(), which silently depends on the container's clock.

Returns:

Type Description
HealthWeightCreate

The validated model instance with date set.
Source code in backend/app/modules/health/health_weight/schema.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@model_validator(mode="after")
def set_default_date(self) -> "HealthWeightCreate":
    """
    Set date to today if not provided.

    Clients should always send their own local date — the request carries no
    timezone, so this fallback can only use the server's configured zone and
    will file an entry logged near local midnight under the wrong day for an
    athlete elsewhere. Resolved through ``settings.TZ`` explicitly rather
    than ``date.today()``, which silently depends on the container's clock.

    Returns:
        The validated model instance with date set.
    """
    if self.date is None:
        self.date = core_timezone.today_in(core_config.settings.TZ)
    return self

HealthWeightListResponse

Bases: HealthListResponse

Response model for listing health weight records.

Extends the base HealthListResponse to provide a typed response containing a list of weight records. This model is used when returning multiple weight entries from API endpoints.

Attributes:

Name Type Description
records list[HealthWeightRead]

A list of HealthWeightRead objects representing individual weight records.
Source code in backend/app/modules/health/health_weight/schema.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class HealthWeightListResponse(health_schema.HealthListResponse):
    """
    Response model for listing health weight records.

    Extends the base HealthListResponse to provide a typed response containing
    a list of weight records. This model is used when returning multiple weight
    entries from API endpoints.

    Attributes:
        records: A list of HealthWeightRead objects representing individual
            weight records.
    """

    records: list[HealthWeightRead] = Field(..., description="List of health weight records")

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

HealthWeightRead

Bases: HealthWeightBase

Schema for reading a health weight record.

Extends HealthWeightBase with identifier fields required for retrieving and referencing weight records in the system.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the weight record to update.
user_id StrictInt

Foreign key reference to the user.
Source code in backend/app/modules/health/health_weight/schema.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class HealthWeightRead(HealthWeightBase):
    """
    Schema for reading a health weight record.

    Extends HealthWeightBase with identifier fields required for retrieving
    and referencing weight records in the system.

    Attributes:
        id (StrictInt): Unique identifier for the weight record to update.
        user_id (StrictInt): Foreign key reference to the user.
    """

    id: StrictInt = Field(..., description="Unique identifier for the weight record to update")
    user_id: StrictInt = Field(..., description="Foreign key reference to the user")

HealthWeightUpdate

Bases: HealthWeightRead

Schema for updating health weight records.

Inherits from HealthWeightRead to maintain consistency with read operations while allowing modifications to health weight data. This schema is used for PUT/PATCH requests to update existing health weight entries.

Source code in backend/app/modules/health/health_weight/schema.py 
119
120
121
122
123
124
125
126
class HealthWeightUpdate(HealthWeightRead):
    """
    Schema for updating health weight records.

    Inherits from HealthWeightRead to maintain consistency with read operations
    while allowing modifications to health weight data. This schema is used for
    PUT/PATCH requests to update existing health weight entries.
    """

Source

Bases: Enum

Enumeration of data sources for health weight records.

Attributes:

Name Type Description
GARMIN

Garmin fitness tracking platform as a data source.
Source code in backend/app/modules/health/health_weight/schema.py 
18
19
20
21
22
23
24
25
26
class Source(Enum):
    """
    Enumeration of data sources for health weight records.

    Attributes:
        GARMIN: Garmin fitness tracking platform as a data source.
    """

    GARMIN = "garmin"

calculate_bmi 

calculate_bmi(health_weight, user_id, db)

Calculate the Body Mass Index (BMI) for a health weight record.

Parameters:

Name Type Description Default
health_weight HealthWeightCreate | HealthWeightUpdate

Health weight record with weight value.

 required
user_id int

Unique identifier of the user.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthWeightCreate | HealthWeightUpdate

Updated health weight record with calculated BMI.
Source code in backend/app/modules/health/health_weight/utils.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def calculate_bmi(
    health_weight: (health_weight_schema.HealthWeightCreate | health_weight_schema.HealthWeightUpdate),
    user_id: int,
    db: Session,
) -> health_weight_schema.HealthWeightCreate | health_weight_schema.HealthWeightUpdate:
    """
    Calculate the Body Mass Index (BMI) for a health weight record.

    Args:
        health_weight: Health weight record with weight value.
        user_id: Unique identifier of the user.
        db: Database session.

    Returns:
        Updated health weight record with calculated BMI.
    """
    # Get the user from the database
    user = users_crud.get_user_by_id(user_id, db)

    # Calculate BMI if user and required data exist
    calculated_bmi = None
    if user is not None and user.height is not None and health_weight.weight is not None:
        # Calculate the bmi: weight (kg) / (height (m))^2
        calculated_bmi = float(health_weight.weight) / ((user.height / 100) ** 2)

    # Return updated model with BMI
    return health_weight.model_copy(update={"bmi": calculated_bmi})

calculate_bmi_all_user_entries 

calculate_bmi_all_user_entries(user_id, db)

Recalculate and persist BMI for all of a user's weight entries.

Fetches the user once and delegates to a single bulk update, avoiding a per-entry query and commit cycle.

Parameters:

Name Type Description Default
user_id int

User ID whose entries should be processed.

 required
db Session

Database session.

 required

Returns:

Type Description
None

None
Source code in backend/app/modules/health/health_weight/utils.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def calculate_bmi_all_user_entries(user_id: int, db: Session) -> None:
    """
    Recalculate and persist BMI for all of a user's weight entries.

    Fetches the user once and delegates to a single bulk update,
    avoiding a per-entry query and commit cycle.

    Args:
        user_id: User ID whose entries should be processed.
        db: Database session.

    Returns:
        None
    """
    # Fetch the user once to read the height used for every entry.
    user = users_crud.get_user_by_id(user_id, db)
    height_cm = float(user.height) if user is not None and user.height is not None else None

    # Recalculate BMI for every entry in a single bulk statement.
    health_weight_crud.recalculate_bmi_for_user(user_id, height_cm, db)

create_health_weight 

create_health_weight(user_id, health_weight, db)

Create a new health weight entry for a user.

This function creates a new health weight record in the database. If the date is not provided, it defaults to the current date. If BMI is not provided, it is automatically calculated using the user's height and the provided weight.

Parameters:

Name Type Description Default
user_id int

The ID of the user for whom the health weight entry is being created.

 required
health_weight HealthWeightCreate

The health weight data to be created, containing fields such as weight, date, and optionally BMI.

 required
db Session

The database session used for database operations.

 required

Returns:

Type Description
HealthWeightRead

health_weight_schema.HealthWeightRead: The created health weight schema instance.

Raises:

Type Description
HTTPException
  • 409 Conflict: If a duplicate entry exists for the same date.
  • 500 Internal Server Error: If any other unexpected error occurs during creation.
Note
  • The function automatically sets the date to current timestamp if not provided.
  • BMI is calculated automatically if not provided in the input.
  • The database transaction is rolled back in case of any errors.
Source code in backend/app/modules/health/health_weight/crud.py 
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
@core_decorators.handle_db_errors
def create_health_weight(
    user_id: int, health_weight: health_weight_schema.HealthWeightCreate, db: Session
) -> health_weight_schema.HealthWeightRead:
    """
    Create a new health weight entry for a user.

    This function creates a new health weight record in the database. If the date is not provided,
    it defaults to the current date. If BMI is not provided, it is automatically calculated
    using the user's height and the provided weight.

    Args:
        user_id (int): The ID of the user for whom the health weight entry is being created.
        health_weight (health_weight_schema.HealthWeightCreate): The health weight data to be created,
            containing fields such as weight, date, and optionally BMI.
        db (Session): The database session used for database operations.

    Returns:
        health_weight_schema.HealthWeightRead: The created health weight schema instance.

    Raises:
        HTTPException:
            - 409 Conflict: If a duplicate entry exists for the same date.
            - 500 Internal Server Error: If any other unexpected error occurs during creation.

    Note:
        - The function automatically sets the date to current timestamp if not provided.
        - BMI is calculated automatically if not provided in the input.
        - The database transaction is rolled back in case of any errors.
    """
    try:
        # Check if bmi is None
        if health_weight.bmi is None:
            health_weight = cast(
                health_weight_schema.HealthWeightCreate,
                health_weight_utils.calculate_bmi(health_weight, user_id, db),
            )

        # Create a new health_weight
        db_health_weight = health_weight_models.HealthWeight(
            **health_weight.model_dump(exclude_none=False),
            user_id=user_id,
        )

        # Add the health_weight to the database
        db.add(db_health_weight)
        db.commit()
        db.refresh(db_health_weight)

        # Return the health_weight
        return _transform_health_weight(db_health_weight)
    except IntegrityError as integrity_error:
        # Rollback the transaction
        db.rollback()

        # Raise an HTTPException with a 409 Internal Server Error status code
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Duplicate entry error. Check if there is already a entry created for {health_weight.date}",
        ) from integrity_error

delete_health_weight 

delete_health_weight(user_id, health_weight_id, db)

Delete a health weight record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the health weight record.

 required
health_weight_id int

Health weight record ID to delete.

 required
db Session

Database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

If record not found or database error.
Source code in backend/app/modules/health/health_weight/crud.py 
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 delete_health_weight(user_id: int, health_weight_id: int, db: Session) -> None:
    """
    Delete a health weight record for a user.

    Args:
        user_id: User ID who owns the health weight record.
        health_weight_id: Health weight record ID to delete.
        db: Database session.

    Returns:
        None

    Raises:
        HTTPException: If record not found or database error.
    """
    # Get and delete the health_weight
    db_health_weight = _get_health_weight_model_by_id_and_user_id_or_404(health_weight_id, user_id, db)

    # Delete the record
    db.delete(db_health_weight)
    db.commit()

edit_health_weight 

edit_health_weight(user_id, health_weight, db)

Edit an existing health weight record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the health weight record.

 required
health_weight HealthWeightUpdate

Health weight data to update.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthWeightRead

Updated health weight schema instance.

Raises:

Type Description
HTTPException

If record not found or database error.
Source code in backend/app/modules/health/health_weight/crud.py 
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
@core_decorators.handle_db_errors
def edit_health_weight(
    user_id: int,
    health_weight: health_weight_schema.HealthWeightUpdate,
    db: Session,
) -> health_weight_schema.HealthWeightRead:
    """
    Edit an existing health weight record for a user.

    Args:
        user_id: User ID who owns the health weight record.
        health_weight: Health weight data to update.
        db: Database session.

    Returns:
        Updated health weight schema instance.

    Raises:
        HTTPException: If record not found or database error.
    """
    # Ensure the health_weight belongs to the user
    if health_weight.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Cannot edit health weight for another user.",
        )

    # Get the health_weight from the database
    db_health_weight = _get_health_weight_model_by_id_and_user_id_or_404(health_weight.id, user_id, db)

    # Check if bmi is None
    if health_weight.bmi is None and health_weight.weight is not None:
        health_weight = cast(
            health_weight_schema.HealthWeightUpdate,
            health_weight_utils.calculate_bmi(health_weight, user_id, db),
        )

    # Dictionary of fields to update if they are not None
    health_weight_data = health_weight.model_dump(exclude_unset=True)
    # Iterate over the fields and update dynamically
    for key, value in health_weight_data.items():
        setattr(db_health_weight, key, value)

    # Commit the transaction and refresh
    db.commit()
    db.refresh(db_health_weight)

    return _transform_health_weight(db_health_weight)

get_all_health_weight 

get_all_health_weight(db)

Retrieve all health weight records from the database.

Parameters:

Name Type Description Default
db Session

Database session.

 required

Returns:

Type Description
list[HealthWeightRead]

List of HealthWeightRead schemas ordered by date descending.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_weight/crud.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@core_decorators.handle_db_errors
def get_all_health_weight(
    db: Session,
) -> list[health_weight_schema.HealthWeightRead]:
    """
    Retrieve all health weight records from the database.

    Args:
        db: Database session.

    Returns:
        List of HealthWeightRead schemas ordered by date descending.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the health_weight from the database
    stmt = select(health_weight_models.HealthWeight).order_by(desc(health_weight_models.HealthWeight.date))
    db_health_weights = db.execute(stmt).scalars().all()
    return _transform_health_weight(list(db_health_weights))

get_all_health_weight_by_user_id 

get_all_health_weight_by_user_id(user_id, db)

Retrieve all health weight records for a user.

Parameters:

Name Type Description Default
user_id int

User ID to fetch records for.

 required
db Session

Database session.

 required

Returns:

Type Description
list[HealthWeightRead]

List of HealthWeightRead schemas ordered by date descending.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_weight/crud.py 
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 get_all_health_weight_by_user_id(user_id: int, db: Session) -> list[health_weight_schema.HealthWeightRead]:
    """
    Retrieve all health weight records for a user.

    Args:
        user_id: User ID to fetch records for.
        db: Database session.

    Returns:
        List of HealthWeightRead schemas ordered by date descending.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the health_weight from the database
    stmt = (
        select(health_weight_models.HealthWeight)
        .where(health_weight_models.HealthWeight.user_id == user_id)
        .order_by(desc(health_weight_models.HealthWeight.date))
    )

    db_health_weights = db.execute(stmt).scalars().all()

    return _transform_health_weight(list(db_health_weights))

get_health_weight_by_date_and_user_id 

get_health_weight_by_date_and_user_id(user_id, date, db)

Retrieve health weight record for a user on a specific date.

Parameters:

Name Type Description Default
user_id int

User ID.

 required
date str

Date string for the weight record.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthWeightRead | None

HealthWeightRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_weight/crud.py 
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
@core_decorators.handle_db_errors
def get_health_weight_by_date_and_user_id(
    user_id: int, date: str, db: Session
) -> health_weight_schema.HealthWeightRead | None:
    """
    Retrieve health weight record for a user on a specific date.

    Args:
        user_id: User ID.
        date: Date string for the weight record.
        db: Database session.

    Returns:
        HealthWeightRead schema if found, None otherwise.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the health_weight from the database
    stmt = select(health_weight_models.HealthWeight).where(
        health_weight_models.HealthWeight.date == func.date(date),
        health_weight_models.HealthWeight.user_id == user_id,
    )

    db_health_weight = db.execute(stmt).scalar_one_or_none()

    return _transform_health_weight(db_health_weight) if db_health_weight else None

get_health_weight_by_user_id 

get_health_weight_by_user_id(user_id, db, page_number=None, num_records=None, interval=None)

Retrieve health weight records for a specific user with optional pagination and filtering.

Parameters:

Name Type Description Default
user_id int

The ID of the user whose health weight records are to be retrieved.

 required
db Session

The database session used to execute the query.

 required
page_number int | None

The page number for pagination (1-indexed). If provided, num_records must also be provided. Defaults to None.

 None
num_records int | None

The number of records per page. If provided, page_number must also be provided. Defaults to None.

 None
interval Interval | None

The time interval to filter records. If provided, only records from the start of the interval to present are returned. Defaults to None.

 None

Returns:

Type Description
list[HealthWeightRead]

list[health_weight_models.HealthWeight]: A list of health weight records sorted by date in descending order, optionally paginated.
Source code in backend/app/modules/health/health_weight/crud.py 
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
@core_decorators.handle_db_errors
def get_health_weight_by_user_id(
    user_id: int,
    db: Session,
    page_number: int | None = None,
    num_records: int | None = None,
    interval: health_constants.Interval | None = None,
) -> list[health_weight_schema.HealthWeightRead]:
    """
    Retrieve health weight records for a specific user with optional pagination
        and filtering.

    Args:
        user_id (int): The ID of the user whose health weight records are to be
            retrieved.
        db (Session): The database session used to execute the query.
        page_number (int | None, optional): The page number for pagination
            (1-indexed).
            If provided, num_records must also be provided. Defaults to None.
        num_records (int | None, optional): The number of records per page.
            If provided, page_number must also be provided. Defaults to None.
        interval (health_constants.Interval | None, optional): The time
            interval to filter records.
            If provided, only records from the start of the interval to present
            are returned. Defaults to None.

    Returns:
        list[health_weight_models.HealthWeight]: A list of health weight
            records sorted by date in descending order, optionally paginated.
    """
    # Get the health_weight from the database
    stmt = select(health_weight_models.HealthWeight).where(health_weight_models.HealthWeight.user_id == user_id)

    if interval is not None:
        stmt = stmt.where(
            health_weight_models.HealthWeight.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    stmt = stmt.order_by(desc(health_weight_models.HealthWeight.date))

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

    db_health_weights = db.execute(stmt).scalars().all()

    return _transform_health_weight(list(db_health_weights))

get_health_weight_number_by_user_id 

get_health_weight_number_by_user_id(user_id, db, interval=None)

Retrieve total count of health weight records for a user. If interval is provided, count only records starting from the calculated start date.

Parameters:

Name Type Description Default
user_id int

User ID to count records for.

 required
db Session

Database session.

 required
interval Interval | None

Optional filter by goal interval.

 None

Returns:

Type Description
int

Total number of health weight records.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_weight/crud.py 
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
@core_decorators.handle_db_errors
def get_health_weight_number_by_user_id(
    user_id: int, db: Session, interval: health_constants.Interval | None = None
) -> int:
    """
    Retrieve total count of health weight records for a user. If interval is
    provided, count only records starting from the calculated start date.

    Args:
        user_id: User ID to count records for.
        db: Database session.
        interval: Optional filter by goal interval.

    Returns:
        Total number of health weight records.

    Raises:
        HTTPException: If database error occurs.
    """
    # Get the number of health_weight from the database
    stmt = (
        select(func.count())
        .select_from(health_weight_models.HealthWeight)
        .where(health_weight_models.HealthWeight.user_id == user_id)
    )

    if interval is not None:
        stmt = stmt.where(
            health_weight_models.HealthWeight.date
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    return db.execute(stmt).scalar_one()

Health poop

Health poop module for managing user bowel movement data.

This module provides CRUD operations and data models for user bowel movement tracking including daily records, Bristol stool scale type, and optional notes.

Exports
  • CRUD: get_health_poop_number_by_user_id, get_health_poop_by_id_and_user_id, get_health_poop_by_user_id, get_health_poop_by_date_and_user_id, create_health_poop, edit_health_poop, delete_health_poop
  • Schemas: HealthPoopBase, HealthPoopCreate, HealthPoopUpdate, HealthPoopRead, HealthPoopListResponse
  • Enums: Source, BristolType
  • Models: HealthPoop (ORM model)

BristolType

Bases: Enum

Bristol Stool Scale classification types.

Attributes:

Name Type Description
TYPE_1

Separate hard lumps (severe constipation).
TYPE_2

Lumpy and sausage-like (mild constipation).
TYPE_3

Sausage shape with cracks (normal).
TYPE_4

Smooth, soft sausage (normal, ideal).
TYPE_5

Soft blobs with clear edges (lacking fiber).
TYPE_6

Mushy with ragged edges (mild diarrhea).
TYPE_7

Liquid, no solid pieces (severe diarrhea).
Source code in backend/app/modules/health/health_poop/schema.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class BristolType(Enum):
    """
    Bristol Stool Scale classification types.

    Attributes:
        TYPE_1: Separate hard lumps (severe constipation).
        TYPE_2: Lumpy and sausage-like (mild constipation).
        TYPE_3: Sausage shape with cracks (normal).
        TYPE_4: Smooth, soft sausage (normal, ideal).
        TYPE_5: Soft blobs with clear edges (lacking fiber).
        TYPE_6: Mushy with ragged edges (mild diarrhea).
        TYPE_7: Liquid, no solid pieces (severe diarrhea).
    """

    TYPE_1 = 1
    TYPE_2 = 2
    TYPE_3 = 3
    TYPE_4 = 4
    TYPE_5 = 5
    TYPE_6 = 6
    TYPE_7 = 7

HealthPoopBase

Bases: BaseModel

Base model for health poop data.

Attributes:

Name Type Description
date_time datetime | None

Date and time of the bowel movement.
bristol_type BristolType | None

Bristol Stool Scale type (1-7).
color Color | None

Stool color description.
notes StrictStr | None

Optional notes about the bowel movement.
source Source | None

Source of the data.
Source code in backend/app/modules/health/health_poop/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
class HealthPoopBase(BaseModel):
    """
    Base model for health poop data.

    Attributes:
        date_time: Date and time of the bowel movement.
        bristol_type: Bristol Stool Scale type (1-7).
        color: Stool color description.
        notes: Optional notes about the bowel movement.
        source: Source of the data.
    """

    date_time: datetime | None = Field(
        default=None,
        description="Date and time of the bowel movement",
    )
    bristol_type: BristolType | None = Field(
        default=None,
        description="Bristol Stool Scale type (1-7)",
    )
    color: Color | None = Field(
        default=None,
        description="Stool color description",
    )
    notes: StrictStr | None = Field(
        default=None,
        max_length=500,
        description=("Optional notes about the bowel movement"),
    )
    source: Source | None = Field(
        default=None,
        description="Source of the data",
    )

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

HealthPoopCreate

Bases: HealthPoopBase

Schema for creating a new poop record.

Requires date_time and sets defaults for source.

Source code in backend/app/modules/health/health_poop/schema.py 
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
class HealthPoopCreate(HealthPoopBase):
    """
    Schema for creating a new poop record.

    Requires date_time and sets defaults for source.
    """

    @model_validator(mode="after")
    def set_defaults_and_validate(
        self,
    ) -> "HealthPoopCreate":
        """
        Validate required fields and set defaults.

        Returns:
            The validated model instance with defaults set.

        Raises:
            ValueError: If date_time is not provided.
        """
        if self.date_time is None:
            raise ValueError("date_time is required")
        if self.source is None:
            self.source = Source.MANUAL
        return self

set_defaults_and_validate 

set_defaults_and_validate()

Validate required fields and set defaults.

Returns:

Type Description
HealthPoopCreate

The validated model instance with defaults set.

Raises:

Type Description
ValueError

If date_time is not provided.
Source code in backend/app/modules/health/health_poop/schema.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@model_validator(mode="after")
def set_defaults_and_validate(
    self,
) -> "HealthPoopCreate":
    """
    Validate required fields and set defaults.

    Returns:
        The validated model instance with defaults set.

    Raises:
        ValueError: If date_time is not provided.
    """
    if self.date_time is None:
        raise ValueError("date_time is required")
    if self.source is None:
        self.source = Source.MANUAL
    return self

HealthPoopListResponse

Bases: HealthListResponse

Response model for listing health poop records.

Attributes:

Name Type Description
records list[HealthPoopRead]

A list of HealthPoopRead objects representing individual bowel movement records.
Source code in backend/app/modules/health/health_poop/schema.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
class HealthPoopListResponse(health_schema.HealthListResponse):
    """
    Response model for listing health poop records.

    Attributes:
        records: A list of HealthPoopRead objects representing
            individual bowel movement records.
    """

    records: list[HealthPoopRead] = Field(
        ...,
        description="List of health poop records",
    )

HealthPoopModel

Bases: Base

User bowel movement tracking data.

Attributes:

Name Type Description
id Mapped[int]

Primary key, auto-incremented unique identifier.
user_id Mapped[int]

Foreign key referencing users.id.
date_time Mapped[datetime]

Date and time of the bowel movement.
bristol_type Mapped[int | None]

Bristol Stool Scale type (1-7).
color Mapped[str | None]

Stool color description.
notes Mapped[str | None]

Optional notes about the bowel movement.
source Mapped[str | None]

Source of the data.
user Mapped[str | None]

Relationship to the Users model.
Table

health_poop

Relationships
  • Many-to-One with Users model through user_id
Source code in backend/app/modules/health/health_poop/models.py 
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
class HealthPoop(Base):
    """
    User bowel movement tracking data.

    Attributes:
        id: Primary key, auto-incremented unique identifier.
        user_id: Foreign key referencing users.id.
        date_time: Date and time of the bowel movement.
        bristol_type: Bristol Stool Scale type (1-7).
        color: Stool color description.
        notes: Optional notes about the bowel movement.
        source: Source of the data.
        user: Relationship to the Users model.

    Table:
        health_poop

    Relationships:
        - Many-to-One with Users model through user_id
    """

    __tablename__ = "health_poop"

    id: Mapped[int] = mapped_column(
        primary_key=True,
        autoincrement=True,
    )
    user_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"),
        nullable=False,
        index=True,
        comment="User ID that the health_poop belongs",
    )
    date_time: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        nullable=False,
        index=True,
        comment="Date and time of bowel movement",
    )
    bristol_type: Mapped[int | None] = mapped_column(
        nullable=True,
        comment="Bristol Stool Scale type (1-7)",
    )
    color: Mapped[str | None] = mapped_column(
        String(50),
        nullable=True,
        comment="Stool color description",
    )
    notes: Mapped[str | None] = mapped_column(
        String(500),
        nullable=True,
        comment="Optional notes about the bowel movement",
    )
    source: Mapped[str | None] = mapped_column(
        String(250),
        nullable=True,
        comment="Source of the health poop data",
    )

    # Define a relationship to the Users model
    users: Mapped["Users"] = relationship(back_populates="health_poop")

HealthPoopRead

Bases: HealthPoopBase

Schema for reading a poop record.

Attributes:

Name Type Description
id StrictInt

Unique identifier for the poop record.
user_id StrictInt

Foreign key reference to the user.
Source code in backend/app/modules/health/health_poop/schema.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class HealthPoopRead(HealthPoopBase):
    """
    Schema for reading a poop record.

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

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

HealthPoopUpdate

Bases: HealthPoopRead

Schema for updating a poop record.

Inherits from HealthPoopRead for consistency with read operations while allowing modifications to poop data.

Source code in backend/app/modules/health/health_poop/schema.py 
171
172
173
174
175
176
177
class HealthPoopUpdate(HealthPoopRead):
    """
    Schema for updating a poop record.

    Inherits from HealthPoopRead for consistency with read
    operations while allowing modifications to poop data.
    """

Source

Bases: Enum

Enumeration of data sources for poop records.

Attributes:

Name Type Description
MANUAL

Manually entered data.
Source code in backend/app/modules/health/health_poop/schema.py 
24
25
26
27
28
29
30
31
32
class Source(Enum):
    """
    Enumeration of data sources for poop records.

    Attributes:
        MANUAL: Manually entered data.
    """

    MANUAL = "manual"

create_health_poop 

create_health_poop(user_id, health_poop, db)

Create a new poop record for a user.

Parameters:

Name Type Description Default
user_id int

User ID for the record owner.

 required
health_poop HealthPoopCreate

Poop data to create.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthPoopRead

Created poop record.

Raises:

Type Description
HTTPException

If database integrity error.
Source code in backend/app/modules/health/health_poop/crud.py 
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
@core_decorators.handle_db_errors
def create_health_poop(
    user_id: int,
    health_poop: health_poop_schema.HealthPoopCreate,
    db: Session,
) -> health_poop_schema.HealthPoopRead:
    """
    Create a new poop record for a user.

    Args:
        user_id: User ID for the record owner.
        health_poop: Poop data to create.
        db: Database session.

    Returns:
        Created poop record.

    Raises:
        HTTPException: If database integrity error.
    """
    try:
        db_health_poop = health_poop_models.HealthPoop(
            **health_poop.model_dump(exclude_none=False),
            user_id=user_id,
        )

        db.add(db_health_poop)
        db.commit()
        db.refresh(db_health_poop)

        return _transform_health_poop(db_health_poop)
    except IntegrityError as integrity_error:
        db.rollback()

        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=("Failed to create poop record. Database integrity error."),
        ) from integrity_error

delete_health_poop 

delete_health_poop(user_id, health_poop_id, db)

Delete a poop record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the record.

 required
health_poop_id int

Poop record ID to delete.

 required
db Session

Database session.

 required

Returns:

Type Description
None

None

Raises:

Type Description
HTTPException

If record not found or database error.
Source code in backend/app/modules/health/health_poop/crud.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
@core_decorators.handle_db_errors
def delete_health_poop(user_id: int, health_poop_id: int, db: Session) -> None:
    """
    Delete a poop record for a user.

    Args:
        user_id: User ID who owns the record.
        health_poop_id: Poop record ID to delete.
        db: Database session.

    Returns:
        None

    Raises:
        HTTPException: If record not found or database error.
    """
    db_health_poop = _get_health_poop_model_by_id_and_user_id_or_404(health_poop_id, user_id, db)

    db.delete(db_health_poop)
    db.commit()

edit_health_poop 

edit_health_poop(user_id, health_poop, db)

Edit poop record for a user.

Parameters:

Name Type Description Default
user_id int

User ID who owns the record.

 required
health_poop HealthPoopUpdate

Poop data to update.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthPoopRead

Updated HealthPoopRead schema.

Raises:

Type Description
HTTPException

403 if editing another user's record, 404 if not found, 500 if database error.
Source code in backend/app/modules/health/health_poop/crud.py 
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
@core_decorators.handle_db_errors
def edit_health_poop(
    user_id: int,
    health_poop: health_poop_schema.HealthPoopUpdate,
    db: Session,
) -> health_poop_schema.HealthPoopRead:
    """
    Edit poop record for a user.

    Args:
        user_id: User ID who owns the record.
        health_poop: Poop data to update.
        db: Database session.

    Returns:
        Updated HealthPoopRead schema.

    Raises:
        HTTPException: 403 if editing another user's record,
            404 if not found, 500 if database error.
    """
    if health_poop.user_id != user_id:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=("Cannot edit health poop record for another user."),
        )

    db_health_poop = _get_health_poop_model_by_id_and_user_id_or_404(health_poop.id, user_id, db)

    health_poop_data = health_poop.model_dump(exclude_unset=True)
    for key, value in health_poop_data.items():
        setattr(db_health_poop, key, value)

    db.commit()
    db.refresh(db_health_poop)

    return _transform_health_poop(db_health_poop)

get_health_poop_by_date_and_user_id 

get_health_poop_by_date_and_user_id(user_id, date, db)

Retrieve poop records for a user on a specific date.

Unlike other health modules, poop can have multiple entries per day, so this returns a list.

Parameters:

Name Type Description Default
user_id int

User ID.

 required
date str

Date string for the poop records.

 required
db Session

Database session.

 required

Returns:

Type Description
list[HealthPoopRead]

List of HealthPoopRead schemas for the given date.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_poop/crud.py 
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
@core_decorators.handle_db_errors
def get_health_poop_by_date_and_user_id(
    user_id: int, date: str, db: Session
) -> list[health_poop_schema.HealthPoopRead]:
    """
    Retrieve poop records for a user on a specific date.

    Unlike other health modules, poop can have multiple entries
    per day, so this returns a list.

    Args:
        user_id: User ID.
        date: Date string for the poop records.
        db: Database session.

    Returns:
        List of HealthPoopRead schemas for the given date.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = (
        select(health_poop_models.HealthPoop)
        .where(
            health_utils.local_date_expression(
                health_poop_models.HealthPoop.date_time,
                users_utils.resolve_user_timezone(user_id, db),
            )
            == func.date(date),
            health_poop_models.HealthPoop.user_id == user_id,
        )
        .order_by(desc(health_poop_models.HealthPoop.date_time))
    )
    db_health_poops = db.execute(stmt).scalars().all()
    return _transform_health_poop(list(db_health_poops))

get_health_poop_by_id_and_user_id 

get_health_poop_by_id_and_user_id(health_poop_id, user_id, db)

Retrieve poop record by ID and user ID.

Parameters:

Name Type Description Default
health_poop_id int

Poop record ID to fetch.

 required
user_id int

User ID to fetch record for.

 required
db Session

Database session.

 required

Returns:

Type Description
HealthPoopRead | None

HealthPoopRead schema if found, None otherwise.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_poop/crud.py 
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
@core_decorators.handle_db_errors
def get_health_poop_by_id_and_user_id(
    health_poop_id: int, user_id: int, db: Session
) -> health_poop_schema.HealthPoopRead | None:
    """
    Retrieve poop record by ID and user ID.

    Args:
        health_poop_id: Poop record ID to fetch.
        user_id: User ID to fetch record for.
        db: Database session.

    Returns:
        HealthPoopRead schema if found, None otherwise.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = select(health_poop_models.HealthPoop).where(
        health_poop_models.HealthPoop.id == health_poop_id,
        health_poop_models.HealthPoop.user_id == user_id,
    )
    db_health_poop = db.execute(stmt).scalar_one_or_none()

    return _transform_health_poop(db_health_poop) if db_health_poop else None

get_health_poop_by_user_id 

get_health_poop_by_user_id(user_id, db, page_number=None, num_records=None, interval=None)

Retrieve poop records for a user with optional pagination and filtering.

Parameters:

Name Type Description Default
user_id int

User ID whose records are to be retrieved.

 required
db Session

Database session used to execute the query.

 required
page_number int | None

Page number for pagination (1-indexed).

 None
num_records int | None

Number of records per page.

 None
interval Interval | None

Time interval to filter records.

 None

Returns:

Type Description
list[HealthPoopRead]

List of HealthPoopRead records sorted by date_time descending.
Source code in backend/app/modules/health/health_poop/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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@core_decorators.handle_db_errors
def get_health_poop_by_user_id(
    user_id: int,
    db: Session,
    page_number: int | None = None,
    num_records: int | None = None,
    interval: health_constants.Interval | None = None,
) -> list[health_poop_schema.HealthPoopRead]:
    """
    Retrieve poop records for a user with optional pagination
    and filtering.

    Args:
        user_id: User ID whose records are to be retrieved.
        db: Database session used to execute the query.
        page_number: Page number for pagination (1-indexed).
        num_records: Number of records per page.
        interval: Time interval to filter records.

    Returns:
        List of HealthPoopRead records sorted by date_time
            descending.
    """
    stmt = select(health_poop_models.HealthPoop).where(health_poop_models.HealthPoop.user_id == user_id)

    if interval is not None:
        stmt = stmt.where(
            health_poop_models.HealthPoop.date_time
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    stmt = stmt.order_by(desc(health_poop_models.HealthPoop.date_time))

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

    db_health_poops = db.execute(stmt).scalars().all()

    return _transform_health_poop(list(db_health_poops))

get_health_poop_number_by_user_id 

get_health_poop_number_by_user_id(user_id, db, interval=None)

Retrieve total count of poop records for a user.

If interval is provided, count only records starting from the calculated start date.

Parameters:

Name Type Description Default
user_id int

User ID to count records for.

 required
db Session

Database session.

 required
interval Interval | None

Optional filter by goal interval.

 None

Returns:

Type Description
int

Total number of health poop records.

Raises:

Type Description
HTTPException

If database error occurs.
Source code in backend/app/modules/health/health_poop/crud.py 
 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
@core_decorators.handle_db_errors
def get_health_poop_number_by_user_id(
    user_id: int,
    db: Session,
    interval: health_constants.Interval | None = None,
) -> int:
    """
    Retrieve total count of poop records for a user.

    If interval is provided, count only records starting from
    the calculated start date.

    Args:
        user_id: User ID to count records for.
        db: Database session.
        interval: Optional filter by goal interval.

    Returns:
        Total number of health poop records.

    Raises:
        HTTPException: If database error occurs.
    """
    stmt = (
        select(func.count())
        .select_from(health_poop_models.HealthPoop)
        .where(health_poop_models.HealthPoop.user_id == user_id)
    )

    if interval is not None:
        stmt = stmt.where(
            health_poop_models.HealthPoop.date_time
            >= health_utils.get_start_date_for_interval(interval.value, users_utils.user_local_today(user_id, db))
        )

    return db.execute(stmt).scalar_one()