furyhawk/home_stack
1
0
Fork 0
mirror of https://github.com/furyhawk/home_stack.git synced 2026-07-20 09:47:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: optimize backend API services and Docker build caching

Browse Source 
- Add database connection pooling (pool_size=20, max_overflow=10, pool_pre_ping)
- Add LRU response caching for weather endpoints with configurable TTLs
- Reuse httpx.AsyncClient across weather API requests
- Add Docker build cache mounts for uv and npm package managers
- Optimize query ordering in users and items endpoints
- Apply ruff formatting fixes across backend
This commit is contained in:
furyhawk
2026-04-23 19:45:48 +08:00
parent 7b4dc70beb
commit 3fa5720b5a
13 changed files with 418 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+8
View File
@@ -13,7 +13,15 @@ Use this as the default operating guide for coding agents working here.
## Repository Layout
- `backend/`: FastAPI app, SQLModel models, Alembic migrations, pytest suite, backend scripts.
- `frontend/`: React 19 + TypeScript + Vite app, Chakra UI, TanStack Router, TanStack Query, Playwright tests.
- `ai_stack/`: AI inference backends (llamacpp, ollama, speaches, vllm).
- `esphome/`: ESPHome configuration for IoT devices.
- `glance/`: Glance dashboard configuration.
- `notebooks/`: Jupyter notebooks for AI/ML experiments (Gemma, Qwen, Whisper, Pydantic AI).
- `scripts/`: top-level Docker and integration scripts.
- `hooks/`: Copier project generation hooks.
- `.github/`: GitHub workflows, issue/discussion templates, dependabot config.
- `docker-compose*.yml`: Docker Compose configurations (base, override, traefik).
- `makefile`: Make commands for common tasks.
- `frontend/src/client/`: generated OpenAPI client; avoid manual edits.
- `frontend/src/routeTree.gen.ts`: generated route tree; do not hand-edit.
backend/Dockerfile
+2 -1
View File
@@ -26,12 +26,13 @@ COPY ./pyproject.toml ./uv.lock ./alembic.ini /app/
# Install dependencies
# Ref: https://docs.astral.sh/uv/guides/integration/docker/#intermediate-layers
RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=cache,target=/root/.cache/pip \
uv pip install --system -e .
ENV PYTHONPATH=/app
# Copy application code and scripts
COPY ./scripts /app/scripts
COPY ./app /app/app
COPY ./scripts /app/scripts
CMD ["fastapi", "run", "--workers", "4", "app/main.py"]
backend/app/api/routes/items.py
+7 -9
View File
@@ -17,19 +17,11 @@ def read_items(
"""
Retrieve items.
"""
if current_user.is_superuser:
count_statement = select(func.count()).select_from(Item)
count = session.exec(count_statement).one()
statement = select(Item).offset(skip).limit(limit)
items = session.exec(statement).all()
count_statement = select(func.count()).select_from(Item)
else:
count_statement = (
select(func.count())
.select_from(Item)
.where(Item.owner_id == current_user.id)
)
count = session.exec(count_statement).one()
statement = (
select(Item)
.where(Item.owner_id == current_user.id)
@@ -37,6 +29,12 @@ def read_items(
.limit(limit)
)
items = session.exec(statement).all()
count_statement = (
select(func.count())
.select_from(Item)
.where(Item.owner_id == current_user.id)
)
count = session.exec(count_statement).one()
return ItemsPublic(data=items, count=count)
backend/app/api/routes/users.py
+2 -3
View File
@@ -38,13 +38,12 @@ def read_users(session: SessionDep, skip: int = 0, limit: int = 100) -> Any:
"""
Retrieve users.
"""
statement = select(User).offset(skip).limit(limit)
users = session.exec(statement).all()
count_statement = select(func.count()).select_from(User)
count = session.exec(count_statement).one()
statement = select(User).offset(skip).limit(limit)
users = session.exec(statement).all()
return UsersPublic(data=users, count=count)
backend/app/api/routes/weather.py
+197 -191
View File
@@ -1,196 +1,198 @@
from datetime import datetime
from typing import Any, Optional, Dict
from typing import Any
import httpx
from fastapi import APIRouter, HTTPException, Query, Depends
from fastapi import APIRouter, HTTPException, Query
from pydantic import ValidationError
# Import directly from models file
from app.core.cache import cache_response, weather_cache
from app.models import (
WeatherApiError,
WeatherResponse,
AirTemperatureResponse,
WindDirectionResponse,
FourDayForecastResponse,
LightningResponse,
WBGTResponse,
TwentyFourHourForecastResponse,
FourDayForecastResponse
WBGTResponse,
WeatherApiError,
WeatherResponse,
WindDirectionResponse,
)
router = APIRouter(prefix="/weather", tags=["weather"])
WEATHER_API_BASE_URL = "https://api-open.data.gov.sg/v2/real-time/api"
_http_client: httpx.AsyncClient | None = None
async def get_http_client() -> httpx.AsyncClient:
global _http_client
if _http_client is None or _http_client.is_closed:
_http_client = httpx.AsyncClient(timeout=30.0)
return _http_client
async def make_api_request(
endpoint: str,
params: Dict[str, Any] = None,
response_model: Any = None
endpoint: str,
params: dict[str, Any] | None = None,
response_model: Any = None,
) -> Any:
"""
Generic function to make API requests to the weather API
"""
url = f"{WEATHER_API_BASE_URL}/{endpoint}"
try:
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url, params=params)
# Handle error responses
if response.status_code != 200:
try:
error_data = response.json()
error = WeatherApiError(**error_data)
raise HTTPException(
status_code=response.status_code,
detail=f"{error.name}: {error.error_msg}"
)
except (ValidationError, KeyError):
raise HTTPException(
status_code=response.status_code,
detail=f"Error from weather API: {response.text}"
)
# Parse successful response
client = await get_http_client()
response = await client.get(url, params=params)
if response.status_code != 200:
try:
data = response.json()
# Try parsing with the model
if response_model:
return response_model(**data)
return data
except ValidationError as e:
# If there's a validation error, provide detailed error information
print(f"Validation error parsing weather data: {str(e)}")
# If the error is specifically about timestamp fields, try to fix
if 'update_timestamp' in str(e) or 'updated_timestamp' in str(e) or 'updatedTimestamp' in str(e):
# Try to manually fix the data structure
if 'data' in data:
# For standard items structure
if 'items' in data['data']:
for item in data['data']['items']:
if 'update_timestamp' in item and 'updated_timestamp' not in item:
item['updated_timestamp'] = item['update_timestamp']
# For forecast records structure (both 24-hour and 4-day)
if 'records' in data['data']:
for record in data['data']['records']:
# Handle updatedTimestamp field
if 'updatedTimestamp' in record and 'updated_timestamp' not in record:
record['updated_timestamp'] = record['updatedTimestamp']
# Handle possible "tiemstamp" typo in API that should be "timestamp"
if 'tiemstamp' in record and 'timestamp' not in record:
record['timestamp'] = record['tiemstamp']
# Try parsing again with the fixed data
try:
if response_model:
return response_model(**data)
return data
except ValidationError:
# If it still fails, fall back to the original error
pass
# Handle field name mismatches for air temperature data
if 'device_id' in str(e) and 'Station' in str(e):
# API returns deviceId, but model expects device_id
if 'data' in data and 'stations' in data['data']:
for station in data['data']['stations']:
if 'deviceId' in station and 'device_id' not in station:
station['device_id'] = station['deviceId']
# Try parsing again with the fixed data
try:
if response_model:
return response_model(**data)
return data
except ValidationError as new_e:
# If there are other issues, continue with fixes
print(f"Still having validation issues after device_id fix: {str(new_e)}")
# Handle 24-hour forecast structure issues
if 'TwentyFourHourForecast' in str(e):
# Handle missing area_metadata or other structural issues
if 'data' in data:
# If area_metadata is missing but required, provide an empty list
if 'area_metadata' not in data['data'] and 'records' in data['data']:
data['data']['area_metadata'] = []
# Try parsing again with the fixed data
try:
if response_model:
return response_model(**data)
return data
except ValidationError as new_e:
# If it still fails, continue with other fixes
print(f"Still having validation issues after 24-hour forecast fix: {str(new_e)}")
# Handle readings issues (station_id -> stationId mapping and data flattening)
if 'station_id' in str(e) or 'value' in str(e) or 'Reading' in str(e) or 'stationId' in str(e):
if 'data' in data and 'readings' in data['data']:
for reading in data['data']['readings']:
# The API structure has timestamp at reading level and data array with stationId/value pairs
# but the frontend/older structure expects flattened readings
if 'data' in reading and isinstance(reading['data'], list):
# For air temperature, flatten the data structure
# Convert from: {"timestamp": "...", "data": [{"stationId": "S109", "value": 33}]}
# To: [{"station_id": "S109", "value": 33, "timestamp": "..."}]
flattened_data = []
timestamp = reading.get('timestamp')
for data_point in reading['data']:
flattened_data.append({
'station_id': data_point.get('stationId', data_point.get('id')),
'value': data_point.get('value'),
'timestamp': timestamp
})
# Replace the readings array with flattened structure
data['data']['readings'] = flattened_data
break # Only process first reading with nested data
elif 'id' in reading and 'station_id' not in reading:
reading['station_id'] = reading['id']
# Try parsing again with the fixed data
try:
if response_model:
return response_model(**data)
return data
except ValidationError as new_e:
# If it still fails, fall back to the original error
print(f"Still having validation issues after station_id fix: {str(new_e)}")
# If we couldn't fix it, raise the original error
error_data = response.json()
error = WeatherApiError(**error_data)
raise HTTPException(
status_code=500,
detail=f"Error parsing weather data: {str(e)}"
status_code=response.status_code,
detail=f"{error.name}: {error.error_msg}",
)
except (ValidationError, KeyError):
raise HTTPException(
status_code=response.status_code,
detail=f"Error from weather API: {response.text}",
)
try:
data = response.json()
if response_model:
return response_model(**data)
return data
except ValidationError as e:
print(f"Validation error parsing weather data: {str(e)}")
if (
"update_timestamp" in str(e)
or "updated_timestamp" in str(e)
or "updatedTimestamp" in str(e)
):
if "data" in data:
if "items" in data["data"]:
for item in data["data"]["items"]:
if (
"update_timestamp" in item
and "updated_timestamp" not in item
):
item["updated_timestamp"] = item["update_timestamp"]
if "records" in data["data"]:
for record in data["data"]["records"]:
if (
"updatedTimestamp" in record
and "updated_timestamp" not in record
):
record["updated_timestamp"] = record["updatedTimestamp"]
if "tiemstamp" in record and "timestamp" not in record:
record["timestamp"] = record["tiemstamp"]
try:
if response_model:
return response_model(**data)
return data
except ValidationError:
pass
if "device_id" in str(e) and "Station" in str(e):
if "data" in data and "stations" in data["data"]:
for station in data["data"]["stations"]:
if "deviceId" in station and "device_id" not in station:
station["device_id"] = station["deviceId"]
try:
if response_model:
return response_model(**data)
return data
except ValidationError as new_e:
print(
f"Still having validation issues after device_id fix: {str(new_e)}"
)
if "TwentyFourHourForecast" in str(e):
if "data" in data:
if (
"area_metadata" not in data["data"]
and "records" in data["data"]
):
data["data"]["area_metadata"] = []
try:
if response_model:
return response_model(**data)
return data
except ValidationError as new_e:
print(
f"Still having validation issues after 24-hour forecast fix: {str(new_e)}"
)
if (
"station_id" in str(e)
or "value" in str(e)
or "Reading" in str(e)
or "stationId" in str(e)
):
if "data" in data and "readings" in data["data"]:
for reading in data["data"]["readings"]:
if "data" in reading and isinstance(reading["data"], list):
flattened_data = []
timestamp = reading.get("timestamp")
for data_point in reading["data"]:
flattened_data.append(
{
"station_id": data_point.get(
"stationId", data_point.get("id")
),
"value": data_point.get("value"),
"timestamp": timestamp,
}
)
data["data"]["readings"] = flattened_data
break
elif "id" in reading and "station_id" not in reading:
reading["station_id"] = reading["id"]
try:
if response_model:
return response_model(**data)
return data
except ValidationError as new_e:
print(
f"Still having validation issues after station_id fix: {str(new_e)}"
)
raise HTTPException(
status_code=500,
detail=f"Error parsing weather data: {str(e)}",
)
except httpx.RequestError as e:
raise HTTPException(
status_code=503,
detail=f"Error communicating with weather API: {str(e)}"
detail=f"Error communicating with weather API: {str(e)}",
)
@router.get("/two-hour-forecast", response_model=WeatherResponse)
@cache_response(_ttl=1800, cache=weather_cache)
async def get_two_hour_forecast(
date: Optional[str] = Query(
date: str | None = Query(
None,
description="SGT date for which to retrieve data (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS)",
),
pagination_token: Optional[str] = Query(
None, description="Pagination token for retrieving subsequent data pages"
pagination_token: str | None = Query(
None,
description="Pagination token for retrieving subsequent data pages",
),
) -> Any:
"""
Retrieve the latest two hour weather forecast from data.gov.sg API.
- Updated half-hourly from NEA
- Forecasts are given for multiple areas in Singapore
"""
# Build the URL and parameters
params = {}
if date:
params["date"] = date
@@ -201,22 +203,23 @@ async def get_two_hour_forecast(
@router.get("/air-temperature", response_model=AirTemperatureResponse)
@cache_response(_ttl=60, cache=weather_cache)
async def get_air_temperature(
date: Optional[str] = Query(
date: str | None = Query(
None,
description="Format: YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS (SGT). Example: 2024-07-16 or 2024-07-16T23:59:00",
),
pagination_token: Optional[str] = Query(
None, description="Pagination token for retrieving subsequent data pages"
pagination_token: str | None = Query(
None,
description="Pagination token for retrieving subsequent data pages",
),
) -> Any:
"""
Get air temperature readings across Singapore
- Has per-minute readings from NEA
- Unit of measure for readings is °C
"""
params = {}
if date:
params["date"] = date
@@ -227,22 +230,23 @@ async def get_air_temperature(
@router.get("/wind-direction", response_model=WindDirectionResponse)
@cache_response(_ttl=60, cache=weather_cache)
async def get_wind_direction(
date: Optional[str] = Query(
date: str | None = Query(
None,
description="SGT date for which to retrieve data (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS)",
),
pagination_token: Optional[str] = Query(
None, description="Pagination token for retrieving subsequent data pages"
pagination_token: str | None = Query(
None,
description="Pagination token for retrieving subsequent data pages",
),
) -> Any:
"""
Get wind direction readings across Singapore
- Has per-minute readings from NEA
- Unit of measure for readings is °
"""
params = {}
if date:
params["date"] = date
@@ -253,24 +257,23 @@ async def get_wind_direction(
@router.get("/lightning", response_model=LightningResponse)
@cache_response(_ttl=300, cache=weather_cache)
async def get_lightning(
date: Optional[str] = Query(
date: str | None = Query(
None,
description="SGT date (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS). Example: 2025-01-16 or 2025-01-16T23:59:00",
),
pagination_token: Optional[str] = Query(
None, description="Pagination token for retrieving subsequent data pages"
pagination_token: str | None = Query(
None,
description="Pagination token for retrieving subsequent data pages",
),
) -> Any:
"""
Retrieve the latest lightning observation
- Updated multiple times throughout the day
"""
params = {
"api": "lightning" # Required parameter for this endpoint
}
params = {"api": "lightning"}
if date:
params["date"] = date
if pagination_token:
@@ -280,25 +283,24 @@ async def get_lightning(
@router.get("/wbgt", response_model=WBGTResponse)
@cache_response(_ttl=300, cache=weather_cache)
async def get_wbgt(
date: Optional[str] = Query(
date: str | None = Query(
None,
description="SGT date (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS). Example: 2025-01-16 or 2025-01-16T23:59:00",
),
pagination_token: Optional[str] = Query(
None, description="Pagination token for retrieving subsequent data pages"
pagination_token: str | None = Query(
None,
description="Pagination token for retrieving subsequent data pages",
),
) -> Any:
"""
Retrieve the latest WBGT (Wet Bulb Globe Temperature) data for accurate heat stress assessment
- Updated multiple times throughout the day
- Unit of measure for readings is °C
"""
params = {
"api": "wbgt" # Required parameter for this endpoint
}
params = {"api": "wbgt"}
if date:
params["date"] = date
if pagination_token:
@@ -308,52 +310,56 @@ async def get_wbgt(
@router.get("/twenty-four-hour-forecast", response_model=TwentyFourHourForecastResponse)
@cache_response(_ttl=3600, cache=weather_cache)
async def get_twenty_four_hour_forecast(
date: Optional[str] = Query(
date: str | None = Query(
None,
description="SGT date (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS). Example: 2024-07-16 or 2024-07-16T23:59:00",
),
pagination_token: Optional[str] = Query(
None, description="Pagination token for retrieving subsequent data pages"
pagination_token: str | None = Query(
None,
description="Pagination token for retrieving subsequent data pages",
),
) -> Any:
"""
Retrieve the latest 24 hour weather forecast
- Updated multiple times throughout the day
- Provides forecasts for different areas of Singapore
"""
params = {}
if date:
params["date"] = date
if pagination_token:
params["paginationToken"] = pagination_token
return await make_api_request("twenty-four-hr-forecast", params, TwentyFourHourForecastResponse)
return await make_api_request(
"twenty-four-hr-forecast", params, TwentyFourHourForecastResponse
)
@router.get("/four-day-outlook", response_model=FourDayForecastResponse)
@cache_response(_ttl=7200, cache=weather_cache)
async def get_four_day_outlook(
date: Optional[str] = Query(
date: str | None = Query(
None,
description="SGT date for which to retrieve data (YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS)",
),
pagination_token: Optional[str] = Query(
None, description="Pagination token for retrieving subsequent data pages"
pagination_token: str | None = Query(
None,
description="Pagination token for retrieving subsequent data pages",
),
) -> Any:
"""
Retrieve the latest 4 day weather forecast
- Updated twice a day from NEA
- The forecast is for the next 4 days
"""
params = {}
if date:
params["date"] = date
if pagination_token:
params["paginationToken"] = pagination_token
return await make_api_request("four-day-outlook", params, FourDayForecastResponse)
return await make_api_request("four-day-outlook", params, FourDayForecastResponse)
backend/app/core/cache.py
+53
View File
@@ -0,0 +1,53 @@
import time
from collections import OrderedDict
from collections.abc import Callable
from functools import wraps
from typing import Any
class LRUCache:
def __init__(self, max_size: int = 128, ttl: int = 300):
self.max_size = max_size
self.ttl = ttl
self.cache: OrderedDict[str, tuple[Any, float]] = OrderedDict()
def get(self, key: str) -> Any | None:
if key not in self.cache:
return None
value, timestamp = self.cache[key]
if time.time() - timestamp > self.ttl:
del self.cache[key]
return None
self.cache.move_to_end(key)
return value
def set(self, key: str, value: Any) -> None:
if key in self.cache:
self.cache.move_to_end(key)
elif len(self.cache) >= self.max_size:
self.cache.popitem(last=False)
self.cache[key] = (value, time.time())
def clear(self) -> None:
self.cache.clear()
weather_cache = LRUCache(max_size=64, ttl=300)
def cache_response(*, _ttl: int = 300, cache: LRUCache | None = None):
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args: Any, **kwargs: Any) -> Any:
target_cache = cache or weather_cache
cache_key = f"{func.__name__}:{str(args)}:{str(sorted(kwargs.items()))}"
cached = target_cache.get(cache_key)
if cached is not None:
return cached
result = await func(*args, **kwargs)
target_cache.set(cache_key, result)
return result
return wrapper
return decorator
backend/app/core/db.py
+8 -1
View File
@@ -4,7 +4,14 @@ from app import crud
from app.core.config import settings
from app.models import User, UserCreate
engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
engine = create_engine(
str(settings.SQLALCHEMY_DATABASE_URI),
pool_size=20,
max_overflow=10,
pool_timeout=30,
pool_recycle=3600,
pool_pre_ping=True,
)
# make sure all SQLModel models are imported (app.models) before initializing DB
backend/app/main.py
+12
View File
@@ -1,3 +1,5 @@
from contextlib import asynccontextmanager
import sentry_sdk
from fastapi import FastAPI
from fastapi.routing import APIRoute
@@ -7,6 +9,15 @@ from app.api.main import api_router
from app.core.config import settings
@asynccontextmanager
async def lifespan(_app: FastAPI):
from app.api.routes.weather import _http_client
yield
if _http_client and not _http_client.is_closed:
await _http_client.aclose()
def custom_generate_unique_id(route: APIRoute) -> str:
return f"{route.tags[0]}-{route.name}"
@@ -18,6 +29,7 @@ app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=f"{settings.API_V1_STR}/openapi.json",
generate_unique_id_function=custom_generate_unique_id,
lifespan=lifespan,
)
# Set all CORS enabled origins
backend/app/models.py
+111 -168
View File
@@ -1,9 +1,10 @@
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional
from typing import Any
from pydantic import EmailStr, BaseModel, Field, model_validator
from sqlmodel import Field as SQLModelField, Relationship, SQLModel
from pydantic import BaseModel, EmailStr, Field
from sqlmodel import Field as SQLModelField
from sqlmodel import Relationship, SQLModel
# Shared properties
@@ -119,7 +120,7 @@ class NewPassword(SQLModel):
class LabelLocation(BaseModel):
latitude: float
longitude: float
model_config = {
"extra": "ignore" # Ignore extra fields in API response
}
@@ -128,7 +129,7 @@ class LabelLocation(BaseModel):
class AreaMetadata(BaseModel):
name: str
label_location: LabelLocation
model_config = {
"extra": "ignore" # Ignore extra fields in API response
}
@@ -138,7 +139,7 @@ class ForecastPeriod(BaseModel):
start: datetime
end: datetime
text: str
model_config = {
"extra": "ignore" # Ignore extra fields in API response
}
@@ -147,7 +148,7 @@ class ForecastPeriod(BaseModel):
class Forecast(BaseModel):
area: str
forecast: str
model_config = {
"extra": "ignore" # Ignore extra fields in API response
}
@@ -157,19 +158,19 @@ class WeatherItem(BaseModel):
updated_timestamp: datetime = Field(alias="update_timestamp")
timestamp: datetime
valid_period: ForecastPeriod
forecasts: List[Forecast]
forecasts: list[Forecast]
model_config = {
"extra": "ignore", # Ignore extra fields in API response
"populate_by_name": True # Allow populating by field name or alias
"populate_by_name": True, # Allow populating by field name or alias
}
class WeatherData(BaseModel):
area_metadata: List[AreaMetadata]
items: List[WeatherItem]
pagination_token: Optional[str] = None
area_metadata: list[AreaMetadata]
items: list[WeatherItem]
pagination_token: str | None = None
model_config = {
"extra": "ignore" # Ignore extra fields in API response
}
@@ -177,21 +178,21 @@ class WeatherData(BaseModel):
class WeatherResponse(BaseModel):
code: int
error_msg: Optional[str] = Field(None, alias="errorMsg")
data: Optional[Dict[str, Any]] = None
error_msg: str | None = Field(None, alias="errorMsg")
data: dict[str, Any] | None = None
model_config = {
"extra": "ignore", # Ignore extra fields in API response
"populate_by_name": True # Allow populating by field name or alias
"populate_by_name": True, # Allow populating by field name or alias
}
class WeatherApiError(BaseModel):
code: int
name: str
data: Optional[Dict[str, Any]] = None
data: dict[str, Any] | None = None
error_msg: str = Field(alias="errorMsg")
model_config = {
"extra": "ignore" # Ignore extra fields in API response
}
@@ -200,13 +201,13 @@ class WeatherApiError(BaseModel):
# Models for air temperature readings
class Station(BaseModel):
id: str
device_id: Optional[str] = Field(None, alias="deviceId") # API returns deviceId
device_id: str | None = Field(None, alias="deviceId") # API returns deviceId
name: str
location: Dict[str, float] # Contains latitude and longitude
location: dict[str, float] # Contains latitude and longitude
model_config = {
"extra": "ignore", # Ignore extra fields in API response
"populate_by_name": True # Allow both snake_case and camelCase
"populate_by_name": True, # Allow both snake_case and camelCase
}
@@ -214,70 +215,52 @@ class Station(BaseModel):
class ReadingDataPoint(BaseModel):
station_id: str = Field(alias="stationId")
value: float
model_config = {
"extra": "ignore",
"populate_by_name": True
}
model_config = {"extra": "ignore", "populate_by_name": True}
class Reading(BaseModel):
timestamp: datetime
data: List[ReadingDataPoint]
model_config = {
"extra": "ignore",
"populate_by_name": True
}
data: list[ReadingDataPoint]
model_config = {"extra": "ignore", "populate_by_name": True}
class AirTemperatureData(BaseModel):
stations: List[Station]
readings: List[Reading]
stations: list[Station]
readings: list[Reading]
reading_type: str = Field(alias="readingType")
reading_unit: str = Field(alias="readingUnit")
pagination_token: Optional[str] = Field(None, alias="paginationToken")
model_config = {
"extra": "ignore",
"populate_by_name": True
}
pagination_token: str | None = Field(None, alias="paginationToken")
model_config = {"extra": "ignore", "populate_by_name": True}
class AirTemperatureResponse(BaseModel):
code: int
error_msg: Optional[str] = Field(None, alias="errorMsg")
data: Optional[AirTemperatureData] = None
model_config = {
"extra": "ignore",
"populate_by_name": True
}
error_msg: str | None = Field(None, alias="errorMsg")
data: AirTemperatureData | None = None
model_config = {"extra": "ignore", "populate_by_name": True}
# Models for wind direction readings
class WindDirectionData(BaseModel):
stations: List[Station]
readings: List[Reading]
stations: list[Station]
readings: list[Reading]
reading_type: str = Field(alias="readingType")
reading_unit: str = Field(alias="readingUnit")
pagination_token: Optional[str] = Field(None, alias="paginationToken")
model_config = {
"extra": "ignore",
"populate_by_name": True
}
pagination_token: str | None = Field(None, alias="paginationToken")
model_config = {"extra": "ignore", "populate_by_name": True}
class WindDirectionResponse(BaseModel):
code: int
error_msg: Optional[str] = Field(None, alias="errorMsg")
data: Optional[WindDirectionData] = None
model_config = {
"extra": "ignore",
"populate_by_name": True
}
error_msg: str | None = Field(None, alias="errorMsg")
data: WindDirectionData | None = None
model_config = {"extra": "ignore", "populate_by_name": True}
# Models for lightning observation
@@ -285,73 +268,55 @@ class LightningItem(BaseModel):
# Generic item model for lightning data
# The actual structure can be expanded based on the real API response
pass
model_config = {
"extra": "ignore"
}
model_config = {"extra": "ignore"}
class LightningRecord(BaseModel):
datetime: str
item: Dict[str, Any] # Flexible structure to accommodate different data formats
updated_timestamp: Optional[datetime] = Field(None)
model_config = {
"extra": "ignore"
}
item: dict[str, Any] # Flexible structure to accommodate different data formats
updated_timestamp: datetime | None = Field(None)
model_config = {"extra": "ignore"}
class LightningData(BaseModel):
records: List[LightningRecord]
pagination_token: Optional[str] = Field(None, alias="paginationToken")
model_config = {
"extra": "ignore",
"populate_by_name": True
}
records: list[LightningRecord]
pagination_token: str | None = Field(None, alias="paginationToken")
model_config = {"extra": "ignore", "populate_by_name": True}
class LightningResponse(BaseModel):
code: int
error_msg: Optional[str] = Field(None, alias="errorMsg")
data: Optional[LightningData] = None
model_config = {
"extra": "ignore",
"populate_by_name": True
}
error_msg: str | None = Field(None, alias="errorMsg")
data: LightningData | None = None
model_config = {"extra": "ignore", "populate_by_name": True}
# Models for WBGT (Wet Bulb Globe Temperature) observations
class WBGTRecord(BaseModel):
datetime: str
item: Dict[str, Any] # Flexible structure to accommodate different data formats
updated_timestamp: Optional[datetime] = Field(None)
model_config = {
"extra": "ignore"
}
item: dict[str, Any] # Flexible structure to accommodate different data formats
updated_timestamp: datetime | None = Field(None)
model_config = {"extra": "ignore"}
class WBGTData(BaseModel):
records: List[WBGTRecord]
pagination_token: Optional[str] = Field(None, alias="paginationToken")
model_config = {
"extra": "ignore",
"populate_by_name": True
}
records: list[WBGTRecord]
pagination_token: str | None = Field(None, alias="paginationToken")
model_config = {"extra": "ignore", "populate_by_name": True}
class WBGTResponse(BaseModel):
code: int
error_msg: Optional[str] = Field(None, alias="errorMsg")
data: Optional[WBGTData] = None
model_config = {
"extra": "ignore",
"populate_by_name": True
}
error_msg: str | None = Field(None, alias="errorMsg")
data: WBGTData | None = None
model_config = {"extra": "ignore", "populate_by_name": True}
# Models for 24-hour weather forecast
@@ -359,48 +324,38 @@ class ForecastPeriodGeneral(BaseModel):
start: datetime
end: datetime
text: str
model_config = {
"extra": "ignore"
}
model_config = {"extra": "ignore"}
class TemperatureRange(BaseModel):
low: int
high: int
unit: str
model_config = {
"extra": "ignore"
}
model_config = {"extra": "ignore"}
class HumidityRange(BaseModel):
low: int
high: int
unit: str
model_config = {
"extra": "ignore"
}
model_config = {"extra": "ignore"}
class ForecastInfo(BaseModel):
code: Optional[str] = None
code: str | None = None
text: str
model_config = {
"extra": "ignore"
}
model_config = {"extra": "ignore"}
class WindInfo(BaseModel):
speed: Dict[str, Any]
speed: dict[str, Any]
direction: str
model_config = {
"extra": "ignore"
}
model_config = {"extra": "ignore"}
class GeneralForecast(BaseModel):
@@ -409,11 +364,8 @@ class GeneralForecast(BaseModel):
relative_humidity: HumidityRange = Field(alias="relativeHumidity")
forecast: ForecastInfo
wind: WindInfo
model_config = {
"extra": "ignore",
"populate_by_name": True
}
model_config = {"extra": "ignore", "populate_by_name": True}
class ForecastItem(BaseModel):
@@ -421,34 +373,25 @@ class ForecastItem(BaseModel):
updated_timestamp: datetime = Field(alias="updatedTimestamp")
timestamp: datetime
general: GeneralForecast
periods: Optional[List[Dict[str, Any]]] = None
model_config = {
"extra": "ignore",
"populate_by_name": True
}
periods: list[dict[str, Any]] | None = None
model_config = {"extra": "ignore", "populate_by_name": True}
class TwentyFourHourForecastData(BaseModel):
area_metadata: Optional[List[AreaMetadata]] = Field(default_factory=list)
records: List[ForecastItem]
pagination_token: Optional[str] = Field(None, alias="paginationToken")
model_config = {
"extra": "ignore",
"populate_by_name": True
}
area_metadata: list[AreaMetadata] | None = Field(default_factory=list)
records: list[ForecastItem]
pagination_token: str | None = Field(None, alias="paginationToken")
model_config = {"extra": "ignore", "populate_by_name": True}
class TwentyFourHourForecastResponse(BaseModel):
code: int
error_msg: Optional[str] = Field(None, alias="errorMsg")
data: Optional[TwentyFourHourForecastData] = None
model_config = {
"extra": "ignore",
"populate_by_name": True
}
error_msg: str | None = Field(None, alias="errorMsg")
data: TwentyFourHourForecastData | None = None
model_config = {"extra": "ignore", "populate_by_name": True}
# Models for 4-day weather forecast
@@ -456,30 +399,30 @@ class FourDayForecastItem(BaseModel):
date: str
updated_timestamp: datetime = Field(alias="updatedTimestamp")
timestamp: datetime
forecasts: List[Dict[str, Any]] # Flexible structure for the forecasts
forecasts: list[dict[str, Any]] # Flexible structure for the forecasts
model_config = {
"extra": "ignore",
"populate_by_name": True # Allow both snake_case and camelCase
"populate_by_name": True, # Allow both snake_case and camelCase
}
class FourDayForecastData(BaseModel):
records: List[FourDayForecastItem]
pagination_token: Optional[str] = Field(None, alias="paginationToken")
records: list[FourDayForecastItem]
pagination_token: str | None = Field(None, alias="paginationToken")
model_config = {
"extra": "ignore",
"populate_by_name": True # Allow both snake_case and camelCase
"populate_by_name": True, # Allow both snake_case and camelCase
}
class FourDayForecastResponse(BaseModel):
code: int
error_msg: Optional[str] = Field(None, alias="errorMsg")
data: Optional[FourDayForecastData] = None
error_msg: str | None = Field(None, alias="errorMsg")
data: FourDayForecastData | None = None
model_config = {
"extra": "ignore",
"populate_by_name": True # Allow both snake_case and camelCase
"populate_by_name": True, # Allow both snake_case and camelCase
}
backend/app/tests/scripts/test_backend_pre_start.py
+6 -6
View File
@@ -24,10 +24,10 @@ def test_init_successful_connection() -> None:
except Exception:
connection_successful = False
assert (
connection_successful
), "The database connection should be successful and not raise an exception."
assert connection_successful, (
"The database connection should be successful and not raise an exception."
)
assert session_mock.exec.called_once_with(
select(1)
), "The session should execute a select statement once."
assert session_mock.exec.called_once_with(select(1)), (
"The session should execute a select statement once."
)
backend/app/tests/scripts/test_test_pre_start.py
+6 -6
View File
@@ -24,10 +24,10 @@ def test_init_successful_connection() -> None:
except Exception:
connection_successful = False
assert (
connection_successful
), "The database connection should be successful and not raise an exception."
assert connection_successful, (
"The database connection should be successful and not raise an exception."
)
assert session_mock.exec.called_once_with(
select(1)
), "The session should execute a select statement once."
assert session_mock.exec.called_once_with(select(1)), (
"The session should execute a select statement once."
)
frontend/Dockerfile
+3 -2
View File
@@ -7,9 +7,10 @@ ARG VITE_API_URL
ARG NODE_ENV=production
ENV VITE_API_URL=${VITE_API_URL}
COPY package*.json /app/
COPY package.json package-lock.json /app/
RUN NODE_ENV=development npm ci
RUN --mount=type=cache,target=/root/.npm \
npm ci --cache=/root/.npm
ENV NODE_ENV=${NODE_ENV}
frontend/Dockerfile.playwright
+3 -2
View File
@@ -2,9 +2,10 @@ FROM node:24
WORKDIR /app
COPY package*.json /app/
COPY package.json package-lock.json /app/
RUN npm install
RUN --mount=type=cache,target=/root/.npm \
npm ci --cache=/root/.npm
RUN npx -y playwright install --with-deps