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¶

Reference Tables¶

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

Connection Management¶

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from config import settings

engine = create_async_engine(
    settings.database_url,
    pool_pre_ping=True,
    pool_recycle=3600,
    echo=settings.debug,
)

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:

# 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.location import EVLocation
from db.models.reference import EVChargingNetwork, EVChargerStall, EVLocationLookup, EVStatistics, AppSettings
from db.models.trip_metrics import EVTripMetrics
from db.models.vehicle_status import EVVehicleStatus

__all__ = [
    "Base",
    "EVChargingSession",
    "EVBatteryStatus",
    "EVTripMetrics",
    "EVLocation",
    "EVVehicleStatus",
    "EVChargingNetwork",
    "EVChargerStall",
    "EVLocationLookup",
    "EVStatistics",
    "AppSettings",
]

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:

from typing 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.