Skip to content

Database

LightningROD uses PostgreSQL 16 with SQLAlchemy 2.0 in async mode. The schema covers the full ha-fordpass data model plus reference tables for networks, locations, stalls, and settings.

Schema Overview

Core Tables

Table Purpose
ev_charging_session Charging events: timing, energy, cost, location, SOC, EVSE data
ev_battery_status HV/12V battery snapshots: SOC, voltage, current, temperature
ev_vehicle_status Drivetrain, temperatures, tire pressure, door locks
ev_trip_metrics Per-trip energy, distance, efficiency, driving scores
ev_location GPS snapshots with optional reverse geocoding

Reference Tables

Table Purpose
ev_charging_networks Network definitions with cost_per_kwh and color
ev_location_lookup Known locations with network FK and optional cost override
ev_charger_stalls Charger configurations per location (type, rated kW, connector)
app_settings Key-value store for user preferences and toggles
ev_statistics Aggregate statistics summary (single row, recomputed)

Key Relationships

ev_charging_networks
    ├── ev_location_lookup (network_id FK)
    │       └── ev_charger_stalls (location_id FK)
    └── ev_charging_session (network_id FK, stall_id FK)

Charging Session Fields

The ev_charging_session table includes:

  • Identity: id, session_id (UUID), device_id
  • Type/Location: charge_type, location_name, location_type, network_id, location_id, is_free
  • Power: charging_voltage, charging_amperage, charging_kw, max_power, min_power
  • Timestamps: session_start_utc, session_end_utc, estimated_end_utc, recorded_at
  • Duration: charge_duration_seconds, plugged_in_duration_seconds
  • Energy/SOC: start_soc, end_soc, energy_kwh, miles_added
  • Cost: cost, cost_source, estimated_cost, cost_without_overrides
  • EVSE: evse_voltage, evse_amperage, evse_kw, evse_energy_kwh, evse_max_power_kw, charger_rated_kw, evse_source, stall_id
  • Location data: address, latitude, longitude
  • Metadata: source_system, ingested_at, original_timestamp, ingest_schema_version

Connection Management

db/engine.py
"""Async SQLAlchemy engine and session factory."""
from urllib.parse import urlparse

from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from config import settings


def _is_sqlite(url: str) -> bool:
    return urlparse(url).scheme.startswith("sqlite")


_kwargs: dict = {"echo": settings.debug}
if _is_sqlite(settings.database_url):
    _kwargs["poolclass"] = NullPool
else:
    _kwargs["pool_pre_ping"] = True
    _kwargs["pool_recycle"] = 3600

engine = create_async_engine(settings.database_url, **_kwargs)


# Dialect captured at engine construction so the listener can no-op cleanly
# on PostgreSQL without inspecting the engine on every connection.
_DIALECT_IS_SQLITE = engine.sync_engine.dialect.name == "sqlite"


@event.listens_for(engine.sync_engine, "connect")
def _set_sqlite_pragmas(dbapi_connection, connection_record):
    """Enable SQLite safety/performance PRAGMAs on each DBAPI connection."""
    if not _DIALECT_IS_SQLITE:
        return
    cursor = dbapi_connection.cursor()
    try:
        cursor.execute("PRAGMA foreign_keys = ON")
        cursor.execute("PRAGMA journal_mode = WAL")
        cursor.execute("PRAGMA synchronous = NORMAL")
        cursor.execute("PRAGMA busy_timeout = 5000")
    finally:
        cursor.close()


AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

Key settings:

  • pool_pre_ping=True -- validates connections before use
  • pool_recycle=3600 -- refreshes connections every hour
  • asyncpg driver for native async PostgreSQL access

Models

Models live in db/models/ with one file per domain area. The __init__.py imports all model classes so Alembic's autogenerate discovers every table:

db/models/__init__.py
"""SQLAlchemy model package."""

# Import Base and ALL model classes so Alembic's autogenerate sees all tables.
# CRITICAL: Every model module must be imported here. If a module is missing,
# alembic revision --autogenerate will produce an empty migration.
from db.models.base import Base
from db.models.battery_status import EVBatteryStatus
from db.models.charging_session import EVChargingSession
from db.models.data_source_config import DataSourceConfig
from db.models.ice_vehicle import IceVehicle
from db.models.location import EVLocation
from db.models.reference import (
    AppSettings,
    EVChargerStall,
    EVChargingNetwork,
    EVLocationLookup,
    EVStatistics,
    GasPriceHistory,
    GasPriceReading,
)
from db.models.trip_metrics import EVTripMetrics
from db.models.vehicle import EVVehicle
from db.models.vehicle_status import EVVehicleStatus

__all__ = [
    "Base",
    "DataSourceConfig",
    "EVChargingSession",
    "EVBatteryStatus",
    "EVTripMetrics",
    "EVLocation",
    "EVVehicleStatus",
    "EVChargingNetwork",
    "EVChargerStall",
    "EVLocationLookup",
    "EVStatistics",
    "EVVehicle",
    "IceVehicle",
    "AppSettings",
    "GasPriceHistory",
    "GasPriceReading",
]

Migrations

Alembic manages schema versioning with an async-compatible env.py.

Creating a New Migration

uv run alembic revision --autogenerate -m "add new column"
uv run alembic upgrade head

uv run alembic revision -m "add new column"
# Edit the generated file in db/migrations/versions/
uv run alembic upgrade head

Dependency Injection

Each request gets its own database session via FastAPI's dependency system:

web/dependencies.py
"""FastAPI dependency providers."""

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession

from db.engine import AsyncSessionLocal


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

Route handlers declare the dependency:

@router.get("/sessions")
async def sessions(request: Request, db: AsyncSession = Depends(get_db)):
    ...

The session is automatically committed on success and closed after the request.