app.schemas module#

Pydantic models for request/response payloads and internal validation.

This module defines the data transfer objects (DTOs) used by the FastAPI application and background jobs to validate inputs and serialize outputs. Schemas mirror the structure of the SQLAlchemy ORM models where appropriate and are designed to be safe for JSON encoding in API responses and logs.

See Also#

app.models

SQLAlchemy models mirrored by these schemas.

app.main

FastAPI endpoints that emit and consume these schemas.

app.database

Session/engine configuration used to produce ORM objects.

Notes#

  • Primary role: provide Pydantic v2 models for API endpoints and service boundaries, enabling strict, well-typed validation and serialization.

  • Key dependencies: relies on Pydantic v2. The from_attributes=True configuration allows converting ORM objects (e.g., app.models) into schemas via model_validate.

  • Invariants: field names intentionally match ORM attributes to simplify conversions. Timestamps are naive datetime objects assumed to be in UTC.

Examples#

>>> from datetime import datetime
>>> from app.schemas import CoordinateSchema, WeatherObservationSchema
>>> center = CoordinateSchema(latitude=59.33, longitude=18.06, label="STHLM")
>>> center.model_dump()["label"]
'STHLM'
>>> obs = WeatherObservationSchema(
...     timestamp=datetime(2024, 1, 1, 12, 0, 0),
...     latitude=59.33,
...     longitude=18.06,
...     air_temperature=2.3,
...     is_imputed=False,
... )
>>> isinstance(obs.timestamp, datetime)
True
class app.schemas.CoordinateListResponse(*, coordinates: List[CoordinateSchema])[source]#

Bases: BaseModel

Response payload carrying a list of coordinates.

Parameters:
coordinateslist[CoordinateSchema]

Collection of coordinates to return to the client.

Attributes:
coordinateslist[CoordinateSchema]

The returned coordinates.

Examples

>>> resp = CoordinateListResponse(
...     coordinates=[CoordinateSchema(latitude=0, longitude=0)]
... )
>>> len(resp.coordinates)
1
coordinates: List[CoordinateSchema]#
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'coordinates': FieldInfo(annotation=List[CoordinateSchema], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class app.schemas.CoordinateSchema(*, id: int | None = None, latitude: float, longitude: float, label: str | None = None, is_central: bool | None = None)[source]#

Bases: BaseModel

Geographic coordinate used for data collection and grouping.

This schema represents a point on Earth expressed in decimal degrees. It mirrors app.models.Coordinate and is commonly returned by the API for map/grid operations and stored as part of training metadata.

Parameters:
idint, optional

Surrogate primary key if available; not required for requests.

latitudefloat

Coordinate latitude in decimal degrees.

longitudefloat

Coordinate longitude in decimal degrees.

labelstr, optional

Optional human‑readable label for the coordinate.

is_centralbool, optional

Marks the central coordinate used as a reference point.

Attributes:
idint | None

Primary key, when present.

latitudefloat

Latitude in decimal degrees.

longitudefloat

Longitude in decimal degrees.

labelstr | None

Optional label for UI or logs.

is_centralbool | None

Whether this coordinate is the designated center.

See also

app.models.Coordinate

ORM model providing the database mapping.

Examples

>>> CoordinateSchema(latitude=59.33, longitude=18.06).model_dump()["latitude"]
59.33
id: int | None#
is_central: bool | None#
label: str | None#
latitude: float#
longitude: float#
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'id': FieldInfo(annotation=Union[int, NoneType], required=False, default=None, description='Primary key'), 'is_central': FieldInfo(annotation=Union[bool, NoneType], required=False, default=None, description='True if this is the central coordinate'), 'label': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, description='Optional label for the point'), 'latitude': FieldInfo(annotation=float, required=True, description='Latitude of the point'), 'longitude': FieldInfo(annotation=float, required=True, description='Longitude of the point')}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class app.schemas.GenericStatusResponse(*, status: str, message: str | None = None)[source]#

Bases: BaseModel

Simple status envelope for human‑readable messages.

Parameters:
statusstr

Machine‑friendly status value (e.g., "ok", "error").

messagestr, optional

Human‑readable description or context for the status.

Attributes:
Same as Parameters.

Examples

>>> GenericStatusResponse(status="ok").model_dump()["status"]
'ok'
message: str | None#
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'message': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'status': FieldInfo(annotation=str, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

status: str#
class app.schemas.PredictionDataResponse(*, history: Dict[str, Dict[str, Any]])[source]#

Bases: BaseModel

Historical model performance keyed by horizon identifier.

Parameters:
historydict[str, dict[str, Any]]

Mapping from horizon key (e.g., "latX_lonY_5min") to arbitrary value dictionaries suitable for charting or tabular display.

Attributes:
historydict[str, dict[str, Any]]

The historical data structure for visualization.

Examples

>>> payload = PredictionDataResponse(history={"h1": {"r2": 0.9}})
>>> float(payload.history["h1"]["r2"]) == 0.9
True
history: Dict[str, Dict[str, Any]]#
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'history': FieldInfo(annotation=Dict[str, Dict[str, Any]], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class app.schemas.TrainingLogSchema(*, id: str, timestamp: datetime, horizon: str, sklearn_score: float, pytorch_score: float, data_count: int, coord_latitude: float | None = None, coord_longitude: float | None = None, horizon_label: str | None = None)[source]#

Bases: BaseModel

Append‑only log entry with training scores and metadata.

Mirrors app.models.TrainingLog. Each entry captures model scores for a given horizon and, optionally, coordinate context.

Parameters:
idstr

Unique identifier (UUID string) for the run.

timestampdatetime

Completion time of the training run in UTC.

horizonstr

Non‑empty key identifying the grouping (often "<coord>_<horizon_label>").

sklearn_scorefloat

R^2 score from the Scikit‑learn model.

pytorch_scorefloat

R^2 score from the PyTorch model.

data_countint

Number of data points used for the run.

coord_latitudefloat, optional

Coordinate latitude associated with the run.

coord_longitudefloat, optional

Coordinate longitude associated with the run.

horizon_labelstr, optional

Human‑friendly label for the horizon (e.g., "5min").

Attributes:
Same as Parameters.

See also

app.models.TrainingLog

ORM model persisted by the training jobs.

Examples

>>> TrainingLogSchema(
...     id="00000000-0000-0000-0000-000000000000",
...     timestamp=datetime(2024, 1, 1, 12, 0, 0),
...     horizon="59.33_18.06_5min",
...     sklearn_score=0.92,
...     pytorch_score=0.93,
...     data_count=1000,
... ).model_dump()["horizon"]
'59.33_18.06_5min'
coord_latitude: float | None#
coord_longitude: float | None#
data_count: int#
horizon: str#
horizon_label: str | None#
id: str#
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'coord_latitude': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'coord_longitude': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'data_count': FieldInfo(annotation=int, required=True), 'horizon': FieldInfo(annotation=str, required=True), 'horizon_label': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'id': FieldInfo(annotation=str, required=True), 'pytorch_score': FieldInfo(annotation=float, required=True), 'sklearn_score': FieldInfo(annotation=float, required=True), 'timestamp': FieldInfo(annotation=datetime, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

pytorch_score: float#
sklearn_score: float#
timestamp: datetime#
class app.schemas.TrainingStatusSchema(*, id: int, is_training: bool, last_trained_at: datetime | None, train_count: int, current_horizon: str | None)[source]#

Bases: BaseModel

Snapshot of the current ML training state.

Represents the single logical row tracked by the application to indicate whether a training job is running and metadata about recent runs.

Parameters:
idint

Primary key, conventionally 1 for the singleton row.

is_trainingbool

Flag indicating whether a training job is currently running.

last_trained_atdatetime, optional

Timestamp of the last completed training job in UTC.

train_countint

Monotonic counter of completed training runs.

current_horizonstr, optional

Human‑readable horizon label (e.g., "5min") or status text.

Attributes:
Same as Parameters.

See also

app.models.TrainingStatus

ORM model used by the backend.

Examples

>>> TrainingStatusSchema(id=1, is_training=False, train_count=3).train_count
3
current_horizon: str | None#
id: int#
is_training: bool#
last_trained_at: datetime | None#
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'current_horizon': FieldInfo(annotation=Union[str, NoneType], required=True), 'id': FieldInfo(annotation=int, required=True), 'is_training': FieldInfo(annotation=bool, required=True), 'last_trained_at': FieldInfo(annotation=Union[datetime, NoneType], required=True), 'train_count': FieldInfo(annotation=int, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

train_count: int#
class app.schemas.WeatherObservationSchema(*, timestamp: datetime, latitude: float, longitude: float, air_temperature: float | None = None, wind_speed: float | None = None, wind_direction: float | None = None, cloud_area_fraction: float | None = None, precipitation_amount: float | None = None, is_imputed: bool)[source]#

Bases: BaseModel

Point‑in‑time weather values at a geographic location.

Mirrors app.models.WeatherObservation. The composite identity of an observation is (timestamp, latitude, longitude). Optional fields may be missing in raw feeds but can be provided after imputation.

Parameters:
timestampdatetime

Observation timestamp in UTC (naive datetime assumed to be UTC).

latitudefloat

Coordinate latitude in decimal degrees.

longitudefloat

Coordinate longitude in decimal degrees.

air_temperaturefloat, optional

Air temperature in degrees Celsius.

wind_speedfloat, optional

Wind speed in meters per second.

wind_directionfloat, optional

Wind direction in degrees (meteorological convention).

cloud_area_fractionfloat, optional

Fraction of the sky covered by clouds (0–1).

precipitation_amountfloat, optional

Precipitation amount in millimeters over the interval.

is_imputedbool

Whether this record was generated by an imputation process.

Attributes:
Same as Parameters.

See also

app.models.WeatherObservation

ORM model with the canonical schema.

Examples

>>> WeatherObservationSchema(
...     timestamp=datetime(2024, 1, 1, 0, 0, 0),
...     latitude=59.33,
...     longitude=18.06,
...     is_imputed=False,
... ).model_dump()["is_imputed"]
False
air_temperature: float | None#
cloud_area_fraction: float | None#
is_imputed: bool#
latitude: float#
longitude: float#
model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'air_temperature': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'cloud_area_fraction': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'is_imputed': FieldInfo(annotation=bool, required=True), 'latitude': FieldInfo(annotation=float, required=True), 'longitude': FieldInfo(annotation=float, required=True), 'precipitation_amount': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'timestamp': FieldInfo(annotation=datetime, required=True), 'wind_direction': FieldInfo(annotation=Union[float, NoneType], required=False, default=None), 'wind_speed': FieldInfo(annotation=Union[float, NoneType], required=False, default=None)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

precipitation_amount: float | None#
timestamp: datetime#
wind_direction: float | None#
wind_speed: float | None#