mirror of
https://github.com/ipnet-mesh/meshcore-hub.git
synced 2026-08-07 17:33:20 +02:00
Merge pull request #243 from ipnet-mesh/feat/postgres-support
Add optional PostgreSQL backend (v0.14.0)
This commit is contained in:
JingleManSweep
@@ -66,6 +66,29 @@ LOG_LEVEL=INFO
|
||||
# └── meshcore.db # SQLite database
|
||||
DATA_HOME=./data
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Database
|
||||
# -----------------------------------------------------------------------------
|
||||
# SQLite is the zero-config default and needs nothing here — it lives at
|
||||
# ${DATA_HOME}/collector/meshcore.db.
|
||||
#
|
||||
# To use PostgreSQL instead, set DATABASE_BACKEND=postgres and fill in the
|
||||
# DATABASE_* values below (the bundled postgres container derives its
|
||||
# POSTGRES_USER/PASSWORD/DB from DATABASE_USER/PASSWORD/NAME). You must also
|
||||
# activate the compose 'postgres' profile, e.g. `docker compose --profile postgres up`.
|
||||
#
|
||||
# DATABASE_BACKEND=postgres
|
||||
# DATABASE_HOST=postgres
|
||||
# DATABASE_PORT=5432
|
||||
# DATABASE_NAME=meshcorehub
|
||||
# DATABASE_SCHEMA=meshcorehub # override per instance (e.g. prod, stg) on a shared cluster
|
||||
# DATABASE_USER=meshcorehub
|
||||
# DATABASE_PASSWORD= # required for postgres; e.g. `openssl rand -base64 32`
|
||||
#
|
||||
# Advanced: set DATABASE_URL to a full SQLAlchemy URL to override all of the above
|
||||
# (e.g. a managed/external Postgres). Takes precedence over DATABASE_BACKEND.
|
||||
# DATABASE_URL=postgresql+psycopg2://user:pass@host:5432/dbname
|
||||
|
||||
# Directory containing seed data files for import
|
||||
# Default: ./seed (relative to docker-compose.yml location)
|
||||
# Inside containers this is mapped to /seed
|
||||
|
||||
+1
-1
@@ -55,7 +55,7 @@ ARG BUILD_VERSION=dev
|
||||
# Set version in _version.py and install the package
|
||||
RUN sed -i "s|__version__ = \"dev\"|__version__ = \"${BUILD_VERSION}\"|" src/meshcore_hub/_version.py && \
|
||||
pip install --upgrade pip && \
|
||||
pip install .
|
||||
pip install ".[postgres]"
|
||||
|
||||
# =============================================================================
|
||||
# Stage 3: Runtime - Final production image
|
||||
|
||||
@@ -244,7 +244,7 @@ Each worker is an independent process sharing one listening socket, so the kerne
|
||||
|
||||
Pick a worker count around the number of CPU cores available to the container; start with `2`–`4` and measure under realistic load.
|
||||
|
||||
**SQLite caveat:** all workers share the same SQLite file on the same host. WAL mode (enabled automatically) allows concurrent readers alongside the single writer (the collector), so reads scale — but **writes do not**, and this does not extend across multiple hosts (a network filesystem breaks SQLite locking). To scale the API across hosts, switch `DATABASE_URL` to PostgreSQL; the API requires no code changes for this.
|
||||
**SQLite caveat:** all workers share the same SQLite file on the same host. WAL mode (enabled automatically) allows concurrent readers alongside the single writer (the collector), so reads scale — but **writes do not**, and this does not extend across multiple hosts (a network filesystem breaks SQLite locking). To scale the API across hosts, switch to PostgreSQL (`DATABASE_BACKEND=postgres`); the API requires no code changes for this. See [Database Backend](#database-backend).
|
||||
|
||||
> Prefer `API_WORKERS` over running multiple `api` containers (`--scale api=N`): the `api` service uses a fixed `container_name`, and one process-managed container per stack keeps logs, health checks, and monitoring simple.
|
||||
|
||||
@@ -346,6 +346,32 @@ All components are configured via environment variables. Create a `.env` file or
|
||||
|
||||
> **Note:** `MQTT_PREFIX` also accepts the legacy alias `MQTT_TOPIC_PREFIX` for backward compatibility.
|
||||
|
||||
### Database Backend
|
||||
|
||||
MeshCore Hub defaults to **SQLite** (zero-config, single host). Set `DATABASE_BACKEND=postgres` to switch to **PostgreSQL** for write scaling and multi-host deployments. Postgres is opt-in — leave these unset to keep using SQLite.
|
||||
|
||||
| Variable | Default | Description |
|
||||
| ------------------- | ------------- | --------------------------------------------------------------------------------------- |
|
||||
| `DATABASE_BACKEND` | `sqlite` | `sqlite` or `postgres`. Explicit switch — Postgres is never selected implicitly. |
|
||||
| `DATABASE_HOST` | `postgres` | Postgres hostname (`postgres` = bundled container service name) |
|
||||
| `DATABASE_PORT` | `5432` | Postgres port |
|
||||
| `DATABASE_NAME` | `meshcorehub` | Database name |
|
||||
| `DATABASE_SCHEMA` | `meshcorehub` | Schema (search_path). Set a distinct value per instance on a shared cluster |
|
||||
| `DATABASE_USER` | `meshcorehub` | Role name |
|
||||
| `DATABASE_PASSWORD` | _(none)_ | **Required** for Postgres |
|
||||
| `DATABASE_URL` | _(none)_ | Advanced: full SQLAlchemy URL; overrides all of the above |
|
||||
|
||||
**Docker:** Postgres is bundled behind the `postgres` profile. The container's credentials/name are derived from the `DATABASE_*` values (single source of truth).
|
||||
|
||||
```bash
|
||||
docker compose --profile postgres --profile core up # Start on Postgres
|
||||
docker compose --profile core up # Start on SQLite (default)
|
||||
```
|
||||
|
||||
**Schema-per-instance:** several instances (e.g. `prod`, `stg`) can share one Postgres cluster, each isolated to its own schema via `search_path` — give each a distinct `DATABASE_SCHEMA`. The schema is created automatically on `db upgrade`.
|
||||
|
||||
See [docs/upgrading.md](docs/upgrading.md#optional-postgresql-backend) for the setup reference and the SQLite → Postgres data-migration runbook.
|
||||
|
||||
### Collector Settings
|
||||
|
||||
| Variable | Default | Description |
|
||||
|
||||
+46
-20
@@ -4,7 +4,7 @@ import os
|
||||
from logging.config import fileConfig
|
||||
|
||||
from alembic import context
|
||||
from sqlalchemy import engine_from_config, pool
|
||||
from sqlalchemy import engine_from_config, pool, text
|
||||
|
||||
from meshcore_hub.common.models import Base
|
||||
|
||||
@@ -20,25 +20,34 @@ target_metadata = Base.metadata
|
||||
|
||||
|
||||
def get_database_url() -> str:
|
||||
"""Get database URL from environment or config."""
|
||||
"""Get the database URL using the same resolution as the app.
|
||||
|
||||
Delegates to CommonSettings.effective_database_url so DATABASE_BACKEND=postgres
|
||||
(+ DATABASE_* components) and an explicit DATABASE_URL are honoured identically to
|
||||
the running services — otherwise migrations would silently target SQLite.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
# First try explicit DATABASE_URL environment variable
|
||||
url = os.environ.get("DATABASE_URL")
|
||||
if url:
|
||||
# Ensure directory exists for sqlite URLs
|
||||
if url.startswith("sqlite:///"):
|
||||
db_path = Path(url.replace("sqlite:///", ""))
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
return url
|
||||
# Try DATA_HOME environment variable
|
||||
data_home = os.environ.get("DATA_HOME")
|
||||
if data_home:
|
||||
db_path = Path(data_home) / "collector" / "meshcore.db"
|
||||
from meshcore_hub.common.config import CommonSettings
|
||||
|
||||
url = CommonSettings().effective_database_url
|
||||
# Ensure the parent directory exists for SQLite file URLs.
|
||||
if url.startswith("sqlite:///"):
|
||||
db_path = Path(url.replace("sqlite:///", ""))
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
return f"sqlite:///{db_path}"
|
||||
# Fall back to alembic.ini
|
||||
return config.get_main_option("sqlalchemy.url", "sqlite:///./meshcore.db")
|
||||
return url
|
||||
|
||||
|
||||
def get_schema(url: str) -> str | None:
|
||||
"""Postgres schema to migrate into, or None for SQLite.
|
||||
|
||||
Each Hub instance keeps its tables and alembic_version in its own schema so
|
||||
multiple instances (prod, stg, ...) can share one Postgres database with
|
||||
independent migration state.
|
||||
"""
|
||||
if url.startswith(("postgresql", "postgres")):
|
||||
return os.environ.get("DATABASE_SCHEMA", "meshcorehub")
|
||||
return None
|
||||
|
||||
|
||||
def run_migrations_offline() -> None:
|
||||
@@ -53,12 +62,17 @@ def run_migrations_offline() -> None:
|
||||
script output.
|
||||
"""
|
||||
url = get_database_url()
|
||||
schema = get_schema(url)
|
||||
context.configure(
|
||||
url=url,
|
||||
target_metadata=target_metadata,
|
||||
literal_binds=True,
|
||||
dialect_opts={"paramstyle": "named"},
|
||||
render_as_batch=True, # SQLite batch mode for ALTER TABLE
|
||||
# Batch mode is a SQLite-only workaround for its limited ALTER TABLE;
|
||||
# Postgres performs ALTERs directly.
|
||||
render_as_batch=url.startswith("sqlite"),
|
||||
version_table_schema=schema,
|
||||
include_schemas=schema is not None,
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
@@ -72,7 +86,9 @@ def run_migrations_online() -> None:
|
||||
and associate a connection with the context.
|
||||
"""
|
||||
configuration = config.get_section(config.config_ini_section, {})
|
||||
configuration["sqlalchemy.url"] = get_database_url()
|
||||
url = get_database_url()
|
||||
configuration["sqlalchemy.url"] = url
|
||||
schema = get_schema(url)
|
||||
|
||||
connectable = engine_from_config(
|
||||
configuration,
|
||||
@@ -81,10 +97,20 @@ def run_migrations_online() -> None:
|
||||
)
|
||||
|
||||
with connectable.connect() as connection:
|
||||
# Ensure the instance's schema exists and scope this connection to it so
|
||||
# tables (and alembic_version) are created there. No-op for SQLite.
|
||||
if schema is not None:
|
||||
connection.execute(text(f'CREATE SCHEMA IF NOT EXISTS "{schema}"'))
|
||||
connection.execute(text(f'SET search_path TO "{schema}"'))
|
||||
connection.commit()
|
||||
|
||||
context.configure(
|
||||
connection=connection,
|
||||
target_metadata=target_metadata,
|
||||
render_as_batch=True, # SQLite batch mode for ALTER TABLE
|
||||
# Batch mode is a SQLite-only workaround for its limited ALTER TABLE.
|
||||
render_as_batch=url.startswith("sqlite"),
|
||||
version_table_schema=schema,
|
||||
include_schemas=schema is not None,
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
|
||||
@@ -32,14 +32,22 @@ depends_on = None
|
||||
def upgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
|
||||
# Aggregate the duplicate ids per lowercased key. SQLite uses GROUP_CONCAT;
|
||||
# Postgres uses STRING_AGG. HAVING references COUNT(*) directly (not the alias)
|
||||
# since Postgres does not allow SELECT aliases in HAVING.
|
||||
if conn.dialect.name == "postgresql":
|
||||
id_agg = "STRING_AGG(id, ',')"
|
||||
else:
|
||||
id_agg = "GROUP_CONCAT(id)"
|
||||
|
||||
# Find groups of duplicate nodes (same lowercase public_key, different actual case)
|
||||
duplicates = conn.execute(text("""
|
||||
duplicates = conn.execute(text(f"""
|
||||
SELECT LOWER(public_key) AS lower_pk,
|
||||
GROUP_CONCAT(id) AS ids,
|
||||
{id_agg} AS ids,
|
||||
COUNT(*) AS cnt
|
||||
FROM nodes
|
||||
GROUP BY LOWER(public_key)
|
||||
HAVING cnt > 1
|
||||
HAVING COUNT(*) > 1
|
||||
""")).fetchall()
|
||||
|
||||
for row in duplicates:
|
||||
|
||||
@@ -10,7 +10,6 @@ from typing import Sequence, Union
|
||||
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy.dialects import sqlite
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "e9f0c4079540"
|
||||
@@ -34,7 +33,7 @@ def upgrade() -> None:
|
||||
sa.Column("route_type", sa.String(length=20), nullable=True),
|
||||
sa.Column("path_len", sa.Integer(), nullable=True),
|
||||
sa.Column("snr", sa.Float(), nullable=True),
|
||||
sa.Column("decoded", sqlite.JSON(), nullable=True),
|
||||
sa.Column("decoded", sa.JSON(), nullable=True),
|
||||
sa.Column("received_at", sa.DateTime(timezone=True), nullable=False),
|
||||
sa.Column("id", sa.String(), nullable=False),
|
||||
sa.Column(
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
"""widen messages.signature to 32 chars
|
||||
|
||||
Revision ID: d4e5f6a7b8c9
|
||||
Revises: c3d4e5f6a7b8
|
||||
Create Date: 2026-06-14 09:15:00.000000+00:00
|
||||
|
||||
The column was declared String(8) but actually stores 16-char hex signatures
|
||||
(and can hold the up-to-32-char packet_hash fallback). SQLite never enforced the
|
||||
length, so the undersized definition went unnoticed until a Postgres migration
|
||||
rejected the data (varchar(8)). Widen it to String(32).
|
||||
|
||||
"""
|
||||
|
||||
from typing import Sequence, Union
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision: str = "d4e5f6a7b8c9"
|
||||
down_revision: Union[str, None] = "c3d4e5f6a7b8"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
with op.batch_alter_table("messages", schema=None) as batch_op:
|
||||
batch_op.alter_column(
|
||||
"signature",
|
||||
existing_type=sa.String(length=8),
|
||||
type_=sa.String(length=32),
|
||||
existing_nullable=True,
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
with op.batch_alter_table("messages", schema=None) as batch_op:
|
||||
batch_op.alter_column(
|
||||
"signature",
|
||||
existing_type=sa.String(length=32),
|
||||
type_=sa.String(length=8),
|
||||
existing_nullable=True,
|
||||
)
|
||||
@@ -147,6 +147,36 @@ services:
|
||||
timeout: 5s
|
||||
retries: 3
|
||||
|
||||
# ==========================================================================
|
||||
# PostgreSQL - optional database backend (alternative to default SQLite)
|
||||
# Enable with: DATABASE_BACKEND=postgres (+ DATABASE_* vars) AND the
|
||||
# 'postgres' profile, e.g. `docker compose --profile core --profile postgres up`.
|
||||
# POSTGRES_* init vars derive from the same DATABASE_* values (single source
|
||||
# of truth); the schema itself is created by `db upgrade` (alembic).
|
||||
# ==========================================================================
|
||||
postgres:
|
||||
image: postgres:17-alpine
|
||||
container_name: ${COMPOSE_PROJECT_NAME:-hub}-postgres
|
||||
profiles:
|
||||
- postgres
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
- POSTGRES_USER=${DATABASE_USER:-meshcorehub}
|
||||
- POSTGRES_PASSWORD=${DATABASE_PASSWORD:-}
|
||||
- POSTGRES_DB=${DATABASE_NAME:-meshcorehub}
|
||||
volumes:
|
||||
- postgres_data:/var/lib/postgresql/data
|
||||
healthcheck:
|
||||
test:
|
||||
[
|
||||
"CMD-SHELL",
|
||||
"pg_isready -U ${DATABASE_USER:-meshcorehub} -d ${DATABASE_NAME:-meshcorehub}",
|
||||
]
|
||||
interval: 10s
|
||||
timeout: 5s
|
||||
retries: 5
|
||||
start_period: 10s
|
||||
|
||||
# ==========================================================================
|
||||
# Collector - MQTT subscriber and database storage
|
||||
# ==========================================================================
|
||||
@@ -179,6 +209,14 @@ services:
|
||||
- CHANNEL_REFRESH_INTERVAL_SECONDS=${CHANNEL_REFRESH_INTERVAL_SECONDS:-300}
|
||||
- DATA_HOME=/data
|
||||
- SEED_HOME=/seed
|
||||
# Database backend (SQLite default; set DATABASE_BACKEND=postgres to switch)
|
||||
- DATABASE_BACKEND=${DATABASE_BACKEND:-sqlite}
|
||||
- DATABASE_HOST=${DATABASE_HOST:-postgres}
|
||||
- DATABASE_PORT=${DATABASE_PORT:-5432}
|
||||
- DATABASE_NAME=${DATABASE_NAME:-meshcorehub}
|
||||
- DATABASE_SCHEMA=${DATABASE_SCHEMA:-meshcorehub}
|
||||
- DATABASE_USER=${DATABASE_USER:-meshcorehub}
|
||||
- DATABASE_PASSWORD=${DATABASE_PASSWORD:-}
|
||||
# Webhook configuration
|
||||
- WEBHOOK_ADVERTISEMENT_URL=${WEBHOOK_ADVERTISEMENT_URL:-}
|
||||
- WEBHOOK_ADVERTISEMENT_SECRET=${WEBHOOK_ADVERTISEMENT_SECRET:-}
|
||||
@@ -240,6 +278,14 @@ services:
|
||||
- MQTT_TRANSPORT=${MQTT_TRANSPORT:-websockets}
|
||||
- MQTT_WS_PATH=${MQTT_WS_PATH:-/}
|
||||
- DATA_HOME=/data
|
||||
# Database backend (SQLite default; set DATABASE_BACKEND=postgres to switch)
|
||||
- DATABASE_BACKEND=${DATABASE_BACKEND:-sqlite}
|
||||
- DATABASE_HOST=${DATABASE_HOST:-postgres}
|
||||
- DATABASE_PORT=${DATABASE_PORT:-5432}
|
||||
- DATABASE_NAME=${DATABASE_NAME:-meshcorehub}
|
||||
- DATABASE_SCHEMA=${DATABASE_SCHEMA:-meshcorehub}
|
||||
- DATABASE_USER=${DATABASE_USER:-meshcorehub}
|
||||
- DATABASE_PASSWORD=${DATABASE_PASSWORD:-}
|
||||
- API_HOST=0.0.0.0
|
||||
- API_PORT=8000
|
||||
# Worker processes for multi-core concurrency (1 = single process).
|
||||
@@ -376,10 +422,23 @@ services:
|
||||
- core
|
||||
- migrate
|
||||
restart: "no"
|
||||
# Wait for postgres only when it's running (postgres profile active);
|
||||
# required:false means SQLite deployments don't pull it in or block on it.
|
||||
depends_on:
|
||||
postgres:
|
||||
condition: service_healthy
|
||||
required: false
|
||||
volumes:
|
||||
- data:/data
|
||||
environment:
|
||||
- DATA_HOME=/data
|
||||
- DATABASE_BACKEND=${DATABASE_BACKEND:-sqlite}
|
||||
- DATABASE_HOST=${DATABASE_HOST:-postgres}
|
||||
- DATABASE_PORT=${DATABASE_PORT:-5432}
|
||||
- DATABASE_NAME=${DATABASE_NAME:-meshcorehub}
|
||||
- DATABASE_SCHEMA=${DATABASE_SCHEMA:-meshcorehub}
|
||||
- DATABASE_USER=${DATABASE_USER:-meshcorehub}
|
||||
- DATABASE_PASSWORD=${DATABASE_PASSWORD:-}
|
||||
command: ["db", "upgrade"]
|
||||
|
||||
# ==========================================================================
|
||||
@@ -419,3 +478,5 @@ volumes:
|
||||
name: ${COMPOSE_PROJECT_NAME:-hub}_observer_data
|
||||
redis_data:
|
||||
name: ${COMPOSE_PROJECT_NAME:-hub}_redis_data
|
||||
postgres_data:
|
||||
name: ${COMPOSE_PROJECT_NAME:-hub}_postgres_data
|
||||
|
||||
@@ -0,0 +1,405 @@
|
||||
# Plan: Add PostgreSQL support and migrate existing SQLite databases
|
||||
|
||||
## Context
|
||||
|
||||
`meshcore-hub` currently runs on SQLite (`sqlite:///{DATA_HOME}/collector/meshcore.db`).
|
||||
SQLite WAL does not work over network filesystems and limits concurrent writers, so it
|
||||
caps the project at a single host — the README already flags switching to Postgres for
|
||||
multi-host scaling. The goal is to (1) make the codebase genuinely Postgres-compatible,
|
||||
(2) add a Postgres container and component-based connection config, and (3) give existing
|
||||
community operators a one-command path to migrate their live SQLite data into Postgres
|
||||
(downtime is acceptable).
|
||||
|
||||
The stack is already mostly ready: SQLAlchemy 2.0 + Alembic, `asyncpg`/`psycopg2-binary`
|
||||
declared as the `[postgres]` optional dependency in `pyproject.toml`, and `DATABASE_URL`
|
||||
threaded through `config.py` and `alembic/env.py`. The work is closing the SQLite-specific
|
||||
gaps and adding the container + migration tooling.
|
||||
|
||||
**Decisions made:** data migration uses a **SQLAlchemy ORM copy script** (type-safe, no
|
||||
extra system dependency for operators); connection config uses **component env vars
|
||||
assembled into a URL**; and the backend is selected by an **explicit `DATABASE_BACKEND`
|
||||
switch** (`sqlite` default | `postgres`) that fails fast when set to `postgres` without the
|
||||
required component vars — rather than silently falling back to SQLite. An explicit
|
||||
`DATABASE_URL`, if set, still overrides everything (managed/external PG, tests). pgloader is
|
||||
*not* used — see "Why not pgloader" below.
|
||||
|
||||
SQLite remains the **zero-config default**: a deployment with no DB env vars behaves exactly
|
||||
as today. Postgres is opt-in and requires agreement across **two layers** — the app
|
||||
(`DATABASE_BACKEND=postgres` + component vars) and the infra (the compose `postgres`
|
||||
profile must be activated). The explicit switch makes a half-configured state fail loudly at
|
||||
startup instead of silently using the wrong database.
|
||||
|
||||
Two further decisions (detailed in Parts B and C):
|
||||
- **Multi-instance isolation via Postgres schemas.** Production runs several Hub instances
|
||||
(prod, stg, …) against one shared Postgres; each is scoped to its own schema
|
||||
(`DATABASE_SCHEMA`, default `meshcorehub`) via `search_path`, with its own
|
||||
`alembic_version` for independent migration state. Defaults for database/schema/role are
|
||||
all `meshcorehub`.
|
||||
- **No admin/bootstrap credentials.** The app never holds cluster-admin creds; provisioning
|
||||
is automatic on the bundled container (image entrypoint) and out-of-band for managed
|
||||
Postgres (IaC/console/DBA). No `POSTGRES_ADMIN_*`.
|
||||
|
||||
### Why not pgloader
|
||||
pgloader would infer the target schema from SQLite's *dynamic* typing and produce wrong
|
||||
Postgres types: `is_observer` (stored `0/1`) → `bigint` not `boolean`; `decoded` JSON
|
||||
(stored as `TEXT`) → `text` not `json`; `DateTime(timezone=True)` values (stored as text)
|
||||
→ no `timestamptz`; `String(64)` length constraints lost; and no `alembic_version`
|
||||
consistent with our migration history. The ORM copy script reuses the existing models, so
|
||||
SQLAlchemy performs every type conversion correctly and the schema is created by
|
||||
`alembic upgrade head`.
|
||||
|
||||
---
|
||||
|
||||
## Implementation order
|
||||
|
||||
Parts map to phases. The throughline: **only Phases 1–2 touch shared code and can regress
|
||||
SQLite, so each ends in a SQLite gate; Phases 3–5 are additive and Postgres-only.** Reach
|
||||
"Postgres-ready code + verified SQLite" at the end of Phase 2 before committing to the
|
||||
heavier container/migration work.
|
||||
|
||||
- **Phase 1 — Code compatibility (Part A).** The four dialect-neutral fixes (upsert, async
|
||||
URL mapping, generic `JSON`, conditional `render_as_batch`).
|
||||
- **Gate 1 (SQLite):** full existing test suite green on SQLite + fresh `db upgrade` from
|
||||
scratch on a new SQLite file. These changes are behaviour-neutral, so any breakage is
|
||||
isolated to these four edits — easy to bisect.
|
||||
- **Phase 2 — Backend switch + config (Part B).** `DATABASE_BACKEND`, the `DATABASE_*`
|
||||
vars, `search_path` wiring, `version_table_schema`.
|
||||
- **Gate 2 (SQLite — the key checkpoint):** re-run full suite; add a test asserting the
|
||||
**default no-env path resolves to the exact same SQLite URL/behaviour as before** (no
|
||||
schema/search_path logic engaged when `DATABASE_BACKEND` unset). *Milestone: code is
|
||||
Postgres-compatible and SQLite is proven untouched.*
|
||||
- **Phase 3 — Postgres container (Part C).** Service, `DATABASE_* → POSTGRES_*` derivation,
|
||||
`initdb.d` schema script, profile, healthcheck. Pure infra; cannot regress SQLite.
|
||||
- **Gate 3 (Postgres):** bring container up, `db upgrade` against it (tables land in the
|
||||
schema, `alembic_version` stamped), run the existing suite against Postgres. Wire a
|
||||
**SQLite + Postgres test matrix** here so both run going forward.
|
||||
- **Phase 4 — Data migration command (Part D).** New Postgres-only code path; zero SQLite risk.
|
||||
- **Gate 4:** round-trip the real dev DB (`data/collector/meshcore.db`) → Postgres,
|
||||
reconcile per-table row counts, spot-check a JSON/bool/timestamp value.
|
||||
- **Phase 5 — Verification + docs (Part E).** End-to-end stack run on the `postgres`
|
||||
profile; `docs/upgrading.md` runbook + `search_path`/provisioning docs.
|
||||
|
||||
---
|
||||
|
||||
## Part A — Make the code Postgres-compatible (required regardless of migration tool)
|
||||
|
||||
These are real runtime bugs on Postgres, not cosmetics.
|
||||
|
||||
1. **Dialect-aware upsert** — `src/meshcore_hub/common/models/event_observer.py:17,125-139`
|
||||
`add_event_observer()` is live collector code and currently uses
|
||||
`from sqlalchemy.dialects.sqlite import insert as sqlite_insert` +
|
||||
`.on_conflict_do_nothing(...)`. On Postgres this emits invalid SQL.
|
||||
Fix: pick the insert construct by bind dialect, e.g.
|
||||
```python
|
||||
if session.bind.dialect.name == "postgresql":
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
stmt = pg_insert(EventObserver).values(...).on_conflict_do_nothing(
|
||||
index_elements=["event_hash", "observer_node_id"])
|
||||
else:
|
||||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||||
stmt = sqlite_insert(EventObserver).values(...).on_conflict_do_nothing(...)
|
||||
```
|
||||
Both dialects expose the same `.on_conflict_do_nothing(index_elements=...)` API, so only
|
||||
the import/constructor differs. Grep for other `dialects.sqlite import insert` usages.
|
||||
|
||||
2. **Async driver mapping** — `src/meshcore_hub/common/database.py:145`
|
||||
`_ensure_async_engine()` only rewrites `sqlite://` → `sqlite+aiosqlite://`. A
|
||||
`postgresql://` URL keeps the sync `psycopg2` driver and async API sessions fail.
|
||||
Fix: map `postgresql://` / `postgres://` → `postgresql+asyncpg://` (leave an already
|
||||
`+driver`-qualified URL untouched). Add a small helper (e.g. `_to_async_url(url)`) used
|
||||
here.
|
||||
|
||||
3. **Generic JSON type** — 4 models import `from sqlalchemy.dialects.sqlite import JSON`:
|
||||
`models/raw_packet.py:7`, `models/telemetry.py`, `models/trace_path.py`,
|
||||
`models/event_log.py`. Switch to generic `from sqlalchemy import JSON`. Generic `JSON`
|
||||
maps to SQLite JSON and Postgres `JSON` automatically. (Optional: use
|
||||
`postgresql.JSONB` via `.with_variant()` for indexability — not required for parity.)
|
||||
|
||||
4. **Conditional batch migrations** — `alembic/env.py:61,87`
|
||||
`render_as_batch=True` is unconditional (it's a SQLite ALTER-TABLE workaround). Make it
|
||||
`render_as_batch = get_database_url().startswith("sqlite")` in both
|
||||
`run_migrations_offline()` and `run_migrations_online()`. Existing migrations that call
|
||||
`op.batch_alter_table(...)` still run correctly on Postgres (Alembic emits direct
|
||||
`ALTER` there), and a fresh Postgres DB runs the whole history from scratch.
|
||||
|
||||
> The SQLite `PRAGMA` block in `database.py:52-65,150-161` is already guarded by
|
||||
> `startswith("sqlite")` — no change needed.
|
||||
|
||||
**Verification for Part A:** run the existing test suite against Postgres (see Part E).
|
||||
|
||||
---
|
||||
|
||||
## Part B — Component-based connection config
|
||||
|
||||
Centralize config in `src/meshcore_hub/common/config.py`. `CollectorSettings` and
|
||||
`APISettings` currently each carry `database_url` + a duplicated `effective_database_url`
|
||||
property (`config.py:72-75,174-182` and the matching block in `APISettings`).
|
||||
|
||||
- Add a `database_backend` field to **`CommonSettings`** (env `DATABASE_BACKEND`,
|
||||
default `"sqlite"`, validated choice `sqlite | postgres`).
|
||||
- Add component fields to **`CommonSettings`** (so both inherit), all defaulting to
|
||||
`"meshcorehub"` where named: `database_host`, `database_port` (default `5432`),
|
||||
`database_name` (the **database**, default `"meshcorehub"`), `database_schema` (the
|
||||
**Postgres schema/namespace**, default `"meshcorehub"`), `database_user`
|
||||
(default `"meshcorehub"`), `database_password`.
|
||||
- `DATABASE_NAME` and `DATABASE_SCHEMA` are **distinct**: the former is the database, the
|
||||
latter is the namespace within it. They differ in the shared-cluster case (see
|
||||
"Multi-instance isolation" below); for a single instance both are `meshcorehub`.
|
||||
- Add a shared resolution helper (method on `CommonSettings`, or module function) used by
|
||||
both `effective_database_url` properties, with this precedence:
|
||||
1. explicit `database_url` if set → use verbatim (escape hatch: managed/external PG, tests);
|
||||
2. else if `database_backend == "postgres"` → require `database_host`, `database_name`,
|
||||
`database_user`, `database_password` (raise a clear startup error naming any missing
|
||||
var — do **not** fall back to SQLite), then assemble
|
||||
`postgresql+psycopg2://{user}:{password}@{host}:{port}/{name}` (URL-encode the password);
|
||||
3. else (`database_backend == "sqlite"`) → existing SQLite default under `DATA_HOME`.
|
||||
- Collapse the duplicated `effective_database_url` into the shared helper.
|
||||
|
||||
### Multi-instance isolation via Postgres schemas
|
||||
Production will run **multiple Hub instances (prod, stg, …) against one shared Postgres**.
|
||||
Isolate them with **schema-per-instance**, not database-per-instance: a shared database with
|
||||
each instance scoped to its own schema (`DATABASE_SCHEMA`). This reuses the schema var and is
|
||||
lighter to provision on a shared/managed cluster than many databases. (Database-per-instance
|
||||
is the heavier alternative — stronger isolation, but more provisioning; not chosen.)
|
||||
|
||||
Mechanics (Postgres only — SQLite ignores all of this):
|
||||
- **Scope every connection to the schema via `search_path`** rather than hardcoding
|
||||
`schema=` on the models/metadata (hardcoding would pin the ORM to one schema and break
|
||||
SQLite). Set it in `create_database_engine` for Postgres, e.g. psycopg2
|
||||
`connect_args={"options": f"-csearch_path={schema}"}` (and the asyncpg equivalent for the
|
||||
async engine — a `SET search_path` on connect via an event listener). Models stay
|
||||
schema-agnostic; the active schema is chosen entirely by config.
|
||||
- **Per-instance migration state**: Alembic must place its bookkeeping in the instance's
|
||||
schema — pass `version_table_schema=<schema>` in `alembic/env.py` (Postgres only) so each
|
||||
instance has its own `alembic_version` and prod/stg can sit on **different revisions
|
||||
independently**. `include_schemas=True` for autogenerate. The `search_path` set on the
|
||||
connection means `upgrade` creates the tables in the right schema automatically.
|
||||
- **Schema must exist first**: the app role needs its schema present. Either pre-provisioned
|
||||
out-of-band (managed; see Part C) or `CREATE SCHEMA IF NOT EXISTS` on the bundled
|
||||
container. The `db migrate-to-postgres` command inherits the same `search_path`, so it
|
||||
loads into the instance's schema with no extra flags.
|
||||
|
||||
Update `.env.example` with a `DATABASE_BACKEND` line (default `sqlite`) plus the
|
||||
`DATABASE_HOST` / `DATABASE_PORT` / `DATABASE_NAME` / `DATABASE_SCHEMA` / `DATABASE_USER` /
|
||||
`DATABASE_PASSWORD` block (defaults `meshcorehub`), documented as "set
|
||||
`DATABASE_BACKEND=postgres` and fill these in to use Postgres; override `DATABASE_SCHEMA`
|
||||
per instance (e.g. `prod`, `stg`) when sharing one cluster; leave defaults for SQLite."
|
||||
|
||||
---
|
||||
|
||||
## Part C — Postgres container
|
||||
|
||||
In `docker-compose.yml` (mirror the existing `redis` service style, named volume pattern
|
||||
at lines ~411-419):
|
||||
|
||||
- Add a `postgres` service (`postgres:17-alpine`), a named
|
||||
`postgres_data:/var/lib/postgresql/data` volume, and a `pg_isready` healthcheck.
|
||||
Put it behind a `postgres` compose profile (SQLite stays the zero-config default; the
|
||||
container only runs when the profile is activated, e.g. `docker compose --profile postgres up`).
|
||||
- **Single source of truth for credentials** — derive the container's init vars from the
|
||||
app's `DATABASE_*` rather than maintaining a duplicate set:
|
||||
```yaml
|
||||
environment:
|
||||
POSTGRES_USER: ${DATABASE_USER:-meshcorehub}
|
||||
POSTGRES_PASSWORD: ${DATABASE_PASSWORD}
|
||||
POSTGRES_DB: ${DATABASE_NAME:-meshcorehub}
|
||||
```
|
||||
The image entrypoint auto-creates the role + database from these on first init (empty
|
||||
volume only). Add a tiny `/docker-entrypoint-initdb.d/` script to also
|
||||
`CREATE SCHEMA IF NOT EXISTS "${DATABASE_SCHEMA}" AUTHORIZATION "${DATABASE_USER}"` so the
|
||||
bundled path needs no manual SQL.
|
||||
- Add `DATABASE_*` (including `DATABASE_BACKEND`) to the env passed to `migrate`,
|
||||
`collector`, and `api` services.
|
||||
- **Document the two-layer requirement explicitly**: enabling Postgres = set
|
||||
`DATABASE_BACKEND=postgres` + component vars **and** activate the `postgres` profile.
|
||||
Compose can't read the app env var to auto-activate its profile, so these stay two
|
||||
switches — but the app's fail-fast validation (Part B) turns a half-configured state into
|
||||
a clear startup error rather than a silent SQLite fallback.
|
||||
- Make `migrate` (and therefore `collector`/`api`) `depends_on` postgres `service_healthy`
|
||||
when the Postgres profile is active.
|
||||
- Add the `postgres_data` named volume to the `volumes:` block.
|
||||
- Mirror into `docker-compose.prod.yml` networking if Postgres should sit on `proxy-net`
|
||||
(usually it should stay internal-only — confirm during implementation).
|
||||
|
||||
### Provisioning decision: no admin/bootstrap credentials
|
||||
The app and tooling **never hold cluster-admin credentials**. The only privileged
|
||||
operations are the one-time `CREATE ROLE` / `CREATE DATABASE` / `CREATE SCHEMA` / `GRANT`;
|
||||
everything afterwards (Alembic `db upgrade` DDL, runtime DML, the data-migration command)
|
||||
runs as the least-privileged app role, which only needs to **own its schema/database**, not
|
||||
superuser. Provisioning is therefore split by environment, and we do **not** add
|
||||
`POSTGRES_ADMIN_USER/PASSWORD` (handing the app permanent admin creds for a one-time task is
|
||||
a security regression and not the production pattern):
|
||||
|
||||
| Environment | Provisioning of role/db/schema/grants | App holds |
|
||||
|-------------|----------------------------------------|-----------|
|
||||
| **Bundled container** (dev / small prod) | Automatic: image entrypoint creates role+db from `POSTGRES_*` (= `DATABASE_*`), init script creates the schema. No admin creds exist or are needed. | `DATABASE_*` |
|
||||
| **Managed / shared Postgres** (prod, stg) | **Out-of-band** — Terraform/IaC, cloud console, or a DBA runs the SQL once. The platform's master user stays in the infra layer, never in the app. | `DATABASE_*` only |
|
||||
|
||||
Managed/shared-cluster provisioning SQL to document in `docs/` (one schema per instance in a
|
||||
shared database; role must own its schema so `db upgrade` can create tables; **need not be
|
||||
superuser** — which is why the migration command's `session_replication_role` step has a
|
||||
documented fallback, Part D):
|
||||
```sql
|
||||
-- once per cluster
|
||||
CREATE ROLE meshcorehub LOGIN PASSWORD '…';
|
||||
CREATE DATABASE meshcorehub OWNER meshcorehub;
|
||||
-- connected to the meshcorehub database, once per Hub instance:
|
||||
CREATE SCHEMA IF NOT EXISTS prod AUTHORIZATION meshcorehub; -- and stg, etc.
|
||||
GRANT ALL ON SCHEMA prod TO meshcorehub;
|
||||
```
|
||||
Each instance then sets `DATABASE_SCHEMA=prod` / `stg`. A Terraform example is a nice-to-have
|
||||
follow-up, not required for v1. (Optional future convenience: a `meshcore-hub db bootstrap`
|
||||
command that accepts admin creds **only as one-shot CLI flags**, never persisted env — out of
|
||||
scope for v1.)
|
||||
|
||||
---
|
||||
|
||||
## Part D — Data migration command (`meshcore-hub db migrate-to-postgres`)
|
||||
|
||||
Add a new Click command in the `db` group in `src/meshcore_hub/__main__.py` (after
|
||||
`db_upgrade`, ~line 86), backed by a helper module (e.g.
|
||||
`src/meshcore_hub/common/db_migrate.py`).
|
||||
|
||||
### Command behaviour
|
||||
- Opens a source `DatabaseManager` (SQLite) and target `DatabaseManager` (Postgres) using
|
||||
the existing `create_database_engine` (`database.py:14`).
|
||||
- **Sensible defaults make the common case zero-flag:** `--source` defaults to the legacy
|
||||
SQLite path under `DATA_HOME` (`sqlite:////data/collector/meshcore.db`) regardless of the
|
||||
configured backend; `--target` defaults to the configured `effective_database_url`
|
||||
(Postgres). Both can be overridden explicitly.
|
||||
- Verifies the target schema is at `head` and tables are empty (refuse otherwise unless
|
||||
`--truncate`) — guards against accidental double-runs. **Non-destructive to the source.**
|
||||
- **Copies at the SQLAlchemy Core level, table-by-table** — *not* through ORM objects.
|
||||
Iterate `Base.metadata.sorted_tables` and for each `Table`, `select(table)` from SQLite
|
||||
and `insert(table)` into Postgres reusing the same `Table` object. Type conversion is a
|
||||
free side-effect of the dialects' type processors: reading applies the SQLite *result
|
||||
processor* per column (`0/1`→`bool`, JSON `TEXT`→`dict`, ISO string→`datetime`), writing
|
||||
applies the Postgres *bind processor* (`bool`→`boolean`, `dict`→`json`,
|
||||
`datetime`→`timestamptz`, UUID-string→`varchar`). No per-model code, works generically
|
||||
for every table. (Core, not ORM, because instances can't be shared across two sessions.)
|
||||
- `Base.metadata.sorted_tables` gives a **parent-first FK order** (`nodes` → `node_tags`,
|
||||
`user_profiles` → `user_profile_nodes`, `event_observers`, event tables, `channels`,
|
||||
`events_log`) and naturally **excludes `alembic_version`** (not a mapped model — it was
|
||||
already created/stamped by `db upgrade`, so we must not touch it).
|
||||
- **Streams large tables**: `select(...).execution_options(stream_results=True)` +
|
||||
`result.partitions(batch_size)` (~2k) so `raw_packets` doesn't materialize in memory;
|
||||
inserts go out via the executemany path.
|
||||
- **Timezone normalization (the one explicit conversion):** SQLite doesn't persist tz, so
|
||||
`DateTime(timezone=True)` values read back as *naive* datetimes; inserting naive values
|
||||
into Postgres `timestamptz` would assume the session tz. Since the app always writes UTC
|
||||
(`utc_now()`), a small normalize step attaches `tzinfo=UTC` to naive datetimes for
|
||||
tz-aware columns before insert.
|
||||
- **Single target transaction** wrapping the whole copy → all-or-nothing; any failure
|
||||
leaves Postgres empty and the operator re-runs. Acceptable given the downtime window.
|
||||
- **FK enforcement during load** — `sorted_tables` order is normally sufficient (the schema
|
||||
has no FK cycles). For robustness the load issues `SET session_replication_role = replica`
|
||||
to disable FK triggers, then restores `DEFAULT`. **Caveat: this requires DB-superuser
|
||||
privilege**, which the bundled compose `postgres` container has but a *managed* Postgres
|
||||
(RDS / Cloud SQL) may not grant. Fallback for managed targets: skip the
|
||||
`session_replication_role` toggle and rely solely on `sorted_tables` order (safe here as
|
||||
there are no FK cycles). The command should detect the missing privilege and fall back
|
||||
gracefully (or expose a `--no-replication-role` flag), and the docs should call this out.
|
||||
- After load, reconcile — no sequences to fix (all PKs are app-generated UUID strings), but
|
||||
log a **per-table source-vs-target row-count comparison** as a built-in check.
|
||||
- `--dry-run` prints the per-table row counts without writing.
|
||||
|
||||
> Reuse `Base.metadata.sorted_tables` and the existing models in
|
||||
> `src/meshcore_hub/common/models/` — do not redefine schema in the script.
|
||||
|
||||
> **Why run it inside the `migrate` service:** the command needs both databases reachable
|
||||
> at once — the `migrate` service mounts the `data` volume (SQLite source) *and* sits on the
|
||||
> compose network with `postgres` (target). Hence `docker compose run --rm migrate ...`
|
||||
> rather than a bare host invocation.
|
||||
|
||||
### Operator runbook (docker-compose deployment, downtime acceptable)
|
||||
|
||||
Starting state: running on SQLite, data in the `data` volume at `/data/collector/meshcore.db`.
|
||||
|
||||
1. **Back up the SQLite database** — this is the rollback path:
|
||||
```bash
|
||||
docker compose cp collector:/data/collector/meshcore.db ./meshcore-backup-$(date +%F).db
|
||||
```
|
||||
2. **Stop the writers** (downtime starts; quiesces the SQLite file):
|
||||
```bash
|
||||
docker compose stop collector api web
|
||||
```
|
||||
3. **Configure Postgres in `.env`** — only the `DATABASE_*` block (the bundled container's
|
||||
`POSTGRES_*` init vars derive from these, see Part C):
|
||||
```ini
|
||||
DATABASE_BACKEND=postgres
|
||||
DATABASE_HOST=postgres
|
||||
DATABASE_PORT=5432
|
||||
DATABASE_NAME=meshcorehub
|
||||
DATABASE_SCHEMA=meshcorehub # override per instance (prod/stg) on a shared cluster
|
||||
DATABASE_USER=meshcorehub
|
||||
DATABASE_PASSWORD=<strong-password> # e.g. openssl rand -base64 32
|
||||
```
|
||||
4. **Start the Postgres container** and wait for its `pg_isready` healthcheck:
|
||||
```bash
|
||||
docker compose --profile postgres up -d postgres
|
||||
```
|
||||
5. **Create the schema in Postgres** (builds correctly-typed tables + stamps `alembic_version`):
|
||||
```bash
|
||||
docker compose --profile postgres run --rm migrate meshcore-hub db upgrade
|
||||
```
|
||||
6. **Copy the data across** (zero-flag common case; optionally `--dry-run` first):
|
||||
```bash
|
||||
docker compose --profile postgres run --rm migrate \
|
||||
meshcore-hub db migrate-to-postgres
|
||||
```
|
||||
Confirm the per-table reconciliation counts all match.
|
||||
7. **Bring the stack up against Postgres and verify**:
|
||||
```bash
|
||||
docker compose --profile postgres up -d
|
||||
docker compose run --rm api meshcore-hub api health
|
||||
```
|
||||
Spot-check the web dashboard shows nodes/events. Downtime ends here.
|
||||
8. **Decommission SQLite (later)** — once confident (a few days), remove the old
|
||||
`meshcore.db` from the `data` volume; keep the step-1 backup archived.
|
||||
|
||||
**Rollback:** stop the stack, set `DATABASE_BACKEND=sqlite` (or remove it) in `.env`, and
|
||||
`docker compose up -d` without the `postgres` profile. You're back on the untouched SQLite
|
||||
file — the migration never mutates the source.
|
||||
|
||||
> The implementation should mirror this runbook into `docs/upgrading.md` (Part D
|
||||
> deliverable).
|
||||
|
||||
---
|
||||
|
||||
## Part E — Verification
|
||||
|
||||
1. **Unit/integration tests on Postgres.** Spin up a throwaway Postgres
|
||||
(`docker run --rm -e POSTGRES_PASSWORD=test -p 55432:5432 postgres:17-alpine`), export
|
||||
the matching `DATABASE_*`, run `meshcore-hub db upgrade`, then the existing test suite
|
||||
pointed at Postgres. Confirms Part A (esp. the `event_observer` upsert and async API
|
||||
sessions) and the migration chain build cleanly on Postgres.
|
||||
2. **Round-trip data migration.** Use the real dev DB at
|
||||
`data/collector/meshcore.db` (or `backup/meshcore.db`) as source. Run `db upgrade` +
|
||||
`db migrate-to-postgres`, then assert per-table row counts match (the command's built-in
|
||||
reconciliation), and spot-check a `raw_packets.decoded` JSON value, an `is_observer`
|
||||
boolean, and a `received_at` timestamp survived with correct types.
|
||||
3. **End-to-end app run.** Bring up the stack with the `postgres` profile + `DATABASE_*`
|
||||
set, confirm `migrate` completes, `collector` ingests an event (exercises
|
||||
`add_event_observer` upsert on Postgres), and the `api`/`web` serve data
|
||||
(`meshcore-hub api health`).
|
||||
|
||||
---
|
||||
|
||||
## Critical files
|
||||
|
||||
| File | Change |
|
||||
|------|--------|
|
||||
| `src/meshcore_hub/common/models/event_observer.py` | Dialect-aware upsert (A1) |
|
||||
| `src/meshcore_hub/common/database.py` | Async driver mapping for Postgres (A2) |
|
||||
| `src/meshcore_hub/common/models/{raw_packet,telemetry,trace_path,event_log}.py` | Generic `JSON` import (A3) |
|
||||
| `alembic/env.py` | Conditional `render_as_batch` (A4) |
|
||||
| `src/meshcore_hub/common/config.py` | Component DATABASE_* vars + shared URL assembly (B) |
|
||||
| `.env.example` | Document new DATABASE_* vars (B) |
|
||||
| `docker-compose.yml` (+ `.prod.yml`) | `postgres` service, volume, depends_on, env (C) |
|
||||
| `src/meshcore_hub/__main__.py` + new `common/db_migrate.py` | `db migrate-to-postgres` command (D) |
|
||||
| `README.md` / `docs/upgrading.md` | Document Postgres setup + migration procedure |
|
||||
|
||||
## Out of scope / notes
|
||||
- Keep SQLite as the zero-config default; Postgres is opt-in. No forced migration.
|
||||
- No schema redesign — all column types already map cleanly once A1–A4 land.
|
||||
- Consider gating CI to run the suite against both SQLite and Postgres (follow-up).
|
||||
@@ -2,6 +2,77 @@
|
||||
|
||||
This guide covers upgrading from a previous MeshCore Hub release to the current version. Check the relevant version section below before upgrading.
|
||||
|
||||
## v0.14.0
|
||||
|
||||
### Optional PostgreSQL Backend
|
||||
|
||||
MeshCore Hub can now run on **PostgreSQL** as an alternative to the default SQLite database. SQLite remains the zero-config default — Postgres is entirely opt-in and **no action is required** to keep using SQLite. Switch to Postgres to scale writes and run the stack across multiple hosts (SQLite's file locking does not work over network filesystems and caps you at a single host). Existing operators can migrate their live SQLite data into Postgres with a single command (downtime required while writers are stopped).
|
||||
|
||||
#### Enabling Postgres
|
||||
|
||||
Set `DATABASE_BACKEND=postgres` and the `DATABASE_*` connection variables, then activate the compose `postgres` profile:
|
||||
|
||||
| Variable | Default | Description |
|
||||
| ------------------- | ------------- | -------------------------------------------------------------------------------------------- |
|
||||
| `DATABASE_BACKEND` | `sqlite` | `sqlite` (default) or `postgres`. Explicit switch — Postgres is never used implicitly. |
|
||||
| `DATABASE_HOST` | `postgres` | Postgres hostname (`postgres` is the bundled container's service name). |
|
||||
| `DATABASE_PORT` | `5432` | Postgres port. |
|
||||
| `DATABASE_NAME` | `meshcorehub` | Database name. The bundled container is initialised with this name. |
|
||||
| `DATABASE_SCHEMA` | `meshcorehub` | Postgres schema (search_path). **Set a distinct value per instance** on a shared cluster. |
|
||||
| `DATABASE_USER` | `meshcorehub` | Role name. The bundled container is initialised with this user. |
|
||||
| `DATABASE_PASSWORD` | _(none)_ | **Required** for Postgres. Generate one, e.g. `openssl rand -base64 32`. |
|
||||
|
||||
```bash
|
||||
# Start the stack on Postgres (bundled container)
|
||||
docker compose -f docker-compose.yml -f docker-compose.dev.yml \
|
||||
--profile postgres --profile core up -d
|
||||
```
|
||||
|
||||
The bundled `postgres` container derives its `POSTGRES_USER` / `POSTGRES_PASSWORD` / `POSTGRES_DB` from the same `DATABASE_USER` / `DATABASE_PASSWORD` / `DATABASE_NAME` values — one source of truth. For a **managed/external** Postgres, point `DATABASE_HOST` at it (and skip the `postgres` profile). Advanced users can instead set a full `DATABASE_URL` (e.g. `postgresql+psycopg2://user:pass@host:5432/db`), which takes precedence over the component variables.
|
||||
|
||||
#### Schema-per-instance (`search_path`)
|
||||
|
||||
Each Hub instance is isolated to its own Postgres **schema** via the connection's `search_path`, not its own database. This lets several instances (e.g. `prod`, `stg`) share **one** Postgres cluster without colliding — each gets its own tables and its own `alembic_version`. Give every instance a distinct `DATABASE_SCHEMA` (e.g. `meshcorehub_prod`, `meshcorehub_stg`). The schema is created automatically on `db upgrade` if it does not exist.
|
||||
|
||||
#### Provisioning the role and database
|
||||
|
||||
The bundled container provisions the role and database for you on first start. For a managed/external Postgres, create them once before pointing Hub at it:
|
||||
|
||||
```sql
|
||||
CREATE ROLE meshcorehub LOGIN PASSWORD 'your-password';
|
||||
CREATE DATABASE meshcorehub OWNER meshcorehub;
|
||||
-- The schema is created by `db upgrade`; the role just needs CREATE on the database.
|
||||
```
|
||||
|
||||
No admin/bootstrap credentials are needed at runtime — Hub only ever connects as `DATABASE_USER`.
|
||||
|
||||
#### Migrating an existing SQLite database to Postgres
|
||||
|
||||
Downtime is required while writers are stopped; the source SQLite file is never modified.
|
||||
|
||||
1. **Back up first.** Copy your `meshcore.db` (or back up the `hub_data` volume — see [Backup & Restore](../README.md#backup--restore)).
|
||||
2. **Stop the writers** (collector and api):
|
||||
```bash
|
||||
docker compose -f docker-compose.yml -f docker-compose.dev.yml stop collector api
|
||||
```
|
||||
3. **Bring up Postgres** and create the schema:
|
||||
```bash
|
||||
docker compose -f docker-compose.yml -f docker-compose.dev.yml --profile postgres up -d postgres
|
||||
docker compose -f docker-compose.yml -f docker-compose.dev.yml --profile postgres run --rm migrate
|
||||
```
|
||||
`migrate` runs `db upgrade` against Postgres, creating the schema, all tables (with correct native types — `boolean`, `json`, `timestamptz`), and stamping `alembic_version`.
|
||||
4. **Copy the data** with the built-in command:
|
||||
```bash
|
||||
docker compose -f docker-compose.yml -f docker-compose.dev.yml --profile postgres \
|
||||
run --rm migrate meshcore-hub db migrate-to-postgres
|
||||
```
|
||||
It defaults the source to `sqlite:///{DATA_HOME}/collector/meshcore.db` and the target to your configured `DATABASE_*` connection. It copies every table in foreign-key order through the ORM (so SQLite's dynamically typed values are converted correctly — `0/1` → `boolean`, JSON text → `json`, naive datetimes → UTC `timestamptz`), then prints a per-table source-vs-target row-count reconciliation and fails on any mismatch. Use `--dry-run` to preview counts first, and `--truncate` to overwrite a non-empty target.
|
||||
5. **Start the stack on Postgres** with `DATABASE_BACKEND=postgres` set (see *Enabling Postgres* above).
|
||||
|
||||
> **Why not pgloader?** pgloader infers the target schema from SQLite's *dynamic* typing and produces wrong Postgres types (e.g. `is_observer` as `bigint` not `boolean`, JSON columns as `text`, no `timestamptz`), and no `alembic_version` consistent with the migration history. The built-in command reuses the ORM models, so types convert correctly and the schema is created by `db upgrade`.
|
||||
|
||||
> **Managed Postgres / non-superuser roles:** the migration disables foreign-key triggers during the copy via `session_replication_role = replica`, which requires a superuser. When the target role is not a superuser (typical for managed Postgres), the command automatically falls back to copying in parent-first order instead. Pass `--no-replication-role` to force the fallback explicitly.
|
||||
|
||||
## v0.13.0
|
||||
|
||||
### Raw Packets (capture, browse, and search wire packets)
|
||||
|
||||
@@ -85,6 +85,101 @@ def db_upgrade(revision: str, database_url: str | None) -> None:
|
||||
click.echo("Database upgrade complete.")
|
||||
|
||||
|
||||
@db.command("migrate-to-postgres")
|
||||
@click.option(
|
||||
"--source",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Source SQLite URL (default: sqlite:///{DATA_HOME}/collector/meshcore.db)",
|
||||
)
|
||||
@click.option(
|
||||
"--target",
|
||||
type=str,
|
||||
default=None,
|
||||
help="Target Postgres URL (default: the configured DATABASE_* connection)",
|
||||
)
|
||||
@click.option("--batch-size", type=int, default=2000, help="Rows per insert batch")
|
||||
@click.option(
|
||||
"--truncate",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Delete existing rows from target tables before loading",
|
||||
)
|
||||
@click.option(
|
||||
"--dry-run",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Report source/target row counts without writing",
|
||||
)
|
||||
@click.option(
|
||||
"--no-replication-role",
|
||||
is_flag=True,
|
||||
default=False,
|
||||
help="Don't disable FK triggers via session_replication_role (managed Postgres)",
|
||||
)
|
||||
def db_migrate_to_postgres(
|
||||
source: str | None,
|
||||
target: str | None,
|
||||
batch_size: int,
|
||||
truncate: bool,
|
||||
dry_run: bool,
|
||||
no_replication_role: bool,
|
||||
) -> None:
|
||||
"""Copy data from an existing SQLite database into PostgreSQL.
|
||||
|
||||
Run 'db upgrade' against the target first to create the schema. This command
|
||||
only moves data and never modifies the source.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
from meshcore_hub.common.config import CollectorSettings
|
||||
from meshcore_hub.common.db_migrate import migrate_sqlite_to_postgres
|
||||
|
||||
settings = CollectorSettings()
|
||||
source_url = (
|
||||
source or f"sqlite:///{Path(settings.data_home) / 'collector' / 'meshcore.db'}"
|
||||
)
|
||||
target_url = target or settings.effective_database_url
|
||||
target_schema = settings.effective_database_schema
|
||||
|
||||
click.echo(f"Source: {source_url}")
|
||||
click.echo(f"Target: {target_url} (schema: {target_schema})")
|
||||
if dry_run:
|
||||
click.echo("Mode: dry-run (no writes)")
|
||||
|
||||
try:
|
||||
result = migrate_sqlite_to_postgres(
|
||||
source_url,
|
||||
target_url,
|
||||
target_schema=target_schema,
|
||||
batch_size=batch_size,
|
||||
truncate=truncate,
|
||||
dry_run=dry_run,
|
||||
disable_replication_role=no_replication_role,
|
||||
)
|
||||
except (RuntimeError, ValueError) as exc:
|
||||
raise click.ClickException(str(exc)) from exc
|
||||
|
||||
click.echo("")
|
||||
if dry_run:
|
||||
# Preview only: the target is expected to be empty, so no OK/MISMATCH judgement.
|
||||
click.echo("table (source rows -> current target rows)")
|
||||
for t in result.tables:
|
||||
click.echo(f" {t.name:28} {t.source_rows:>8} -> {t.target_rows:>8}")
|
||||
click.echo("")
|
||||
click.echo("Dry run complete.")
|
||||
return
|
||||
|
||||
click.echo("table (source -> target)")
|
||||
for t in result.tables:
|
||||
status = "OK" if t.ok else "MISMATCH"
|
||||
click.echo(f" {t.name:28} {t.source_rows:>8} -> {t.target_rows:>8} {status}")
|
||||
if not result.ok:
|
||||
raise click.ClickException("Row-count mismatch between source and target")
|
||||
click.echo("")
|
||||
click.echo("Migration complete.")
|
||||
|
||||
|
||||
@db.command("downgrade")
|
||||
@click.option(
|
||||
"--revision",
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import Field, field_validator
|
||||
from pydantic import Field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
||||
@@ -24,6 +24,13 @@ class MQTTTransport(str, Enum):
|
||||
WEBSOCKETS = "websockets"
|
||||
|
||||
|
||||
class DatabaseBackend(str, Enum):
|
||||
"""Database backend selector."""
|
||||
|
||||
SQLITE = "sqlite"
|
||||
POSTGRES = "postgres"
|
||||
|
||||
|
||||
class CommonSettings(BaseSettings):
|
||||
"""Common settings shared by all components."""
|
||||
|
||||
@@ -39,6 +46,88 @@ class CommonSettings(BaseSettings):
|
||||
description="Base directory for service data (e.g., ./data or /data)",
|
||||
)
|
||||
|
||||
# Database backend selection and connection components.
|
||||
# SQLite is the zero-config default; set DATABASE_BACKEND=postgres (plus the
|
||||
# DATABASE_* component vars) to use Postgres. An explicit DATABASE_URL overrides
|
||||
# everything (managed/external Postgres, tests).
|
||||
database_backend: DatabaseBackend = Field(
|
||||
default=DatabaseBackend.SQLITE,
|
||||
description="Database backend: 'sqlite' (default) or 'postgres'",
|
||||
)
|
||||
database_url: Optional[str] = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Explicit SQLAlchemy database URL; overrides DATABASE_BACKEND/component vars. "
|
||||
"Default: sqlite:///{data_home}/collector/meshcore.db"
|
||||
),
|
||||
)
|
||||
database_host: Optional[str] = Field(
|
||||
default=None,
|
||||
description="Postgres host (required when DATABASE_BACKEND=postgres)",
|
||||
)
|
||||
database_port: int = Field(default=5432, description="Postgres port")
|
||||
database_name: str = Field(
|
||||
default="meshcorehub", description="Postgres database name"
|
||||
)
|
||||
database_schema: str = Field(
|
||||
default="meshcorehub",
|
||||
description="Postgres schema (namespace); override per instance on a shared cluster",
|
||||
)
|
||||
database_user: str = Field(default="meshcorehub", description="Postgres role/user")
|
||||
database_password: Optional[str] = Field(
|
||||
default=None,
|
||||
description="Postgres password (required when DATABASE_BACKEND=postgres)",
|
||||
)
|
||||
|
||||
@property
|
||||
def effective_database_url(self) -> str:
|
||||
"""Resolve the SQLAlchemy database URL.
|
||||
|
||||
Precedence: explicit DATABASE_URL > postgres (assembled from components) >
|
||||
SQLite default under DATA_HOME. Fails fast for a misconfigured postgres backend
|
||||
rather than silently falling back to SQLite.
|
||||
"""
|
||||
if self.database_url:
|
||||
return self.database_url
|
||||
if self.database_backend == DatabaseBackend.POSTGRES:
|
||||
missing = [
|
||||
name
|
||||
for name, value in (
|
||||
("DATABASE_HOST", self.database_host),
|
||||
("DATABASE_NAME", self.database_name),
|
||||
("DATABASE_USER", self.database_user),
|
||||
("DATABASE_PASSWORD", self.database_password),
|
||||
)
|
||||
if not value
|
||||
]
|
||||
if missing:
|
||||
raise ValueError(
|
||||
"DATABASE_BACKEND=postgres requires: " + ", ".join(missing)
|
||||
)
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
user = quote_plus(self.database_user)
|
||||
password = quote_plus(self.database_password or "")
|
||||
return (
|
||||
f"postgresql+psycopg2://{user}:{password}"
|
||||
f"@{self.database_host}:{self.database_port}/{self.database_name}"
|
||||
)
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path(self.data_home) / "collector" / "meshcore.db"
|
||||
return f"sqlite:///{db_path}"
|
||||
|
||||
@property
|
||||
def effective_database_schema(self) -> Optional[str]:
|
||||
"""Postgres schema to scope connections to, or None for SQLite.
|
||||
|
||||
Returns the schema only when the effective URL is Postgres; SQLite has no
|
||||
schema concept, so callers leave search_path untouched.
|
||||
"""
|
||||
if self.effective_database_url.startswith(("postgresql", "postgres")):
|
||||
return self.database_schema
|
||||
return None
|
||||
|
||||
# Logging
|
||||
log_level: LogLevel = Field(default=LogLevel.INFO, description="Logging level")
|
||||
|
||||
@@ -68,11 +157,7 @@ class CommonSettings(BaseSettings):
|
||||
class CollectorSettings(CommonSettings):
|
||||
"""Settings for the Collector component."""
|
||||
|
||||
# Database - default uses data_home/collector/meshcore.db
|
||||
database_url: Optional[str] = Field(
|
||||
default=None,
|
||||
description="SQLAlchemy database URL (default: sqlite:///{data_home}/collector/meshcore.db)",
|
||||
)
|
||||
# Database config (backend selector + connection) is inherited from CommonSettings.
|
||||
|
||||
# Seed home directory - contains initial data files (node_tags.yaml)
|
||||
seed_home: str = Field(
|
||||
@@ -171,16 +256,6 @@ class CollectorSettings(CommonSettings):
|
||||
|
||||
return str(Path(self.data_home) / "collector")
|
||||
|
||||
@property
|
||||
def effective_database_url(self) -> str:
|
||||
"""Get the effective database URL, using default if not set."""
|
||||
if self.database_url:
|
||||
return self.database_url
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path(self.data_home) / "collector" / "meshcore.db"
|
||||
return f"sqlite:///{db_path}"
|
||||
|
||||
@property
|
||||
def effective_seed_home(self) -> str:
|
||||
"""Get the effective seed home directory."""
|
||||
@@ -202,13 +277,6 @@ class CollectorSettings(CommonSettings):
|
||||
|
||||
return str(Path(self.effective_seed_home) / "channels.yaml")
|
||||
|
||||
@field_validator("database_url")
|
||||
@classmethod
|
||||
def validate_database_url(cls, v: Optional[str]) -> Optional[str]:
|
||||
"""Validate database URL format."""
|
||||
# None is allowed - will use default
|
||||
return v
|
||||
|
||||
|
||||
class APISettings(CommonSettings):
|
||||
"""Settings for the API component."""
|
||||
@@ -217,11 +285,7 @@ class APISettings(CommonSettings):
|
||||
api_host: str = Field(default="0.0.0.0", description="API server host")
|
||||
api_port: int = Field(default=8000, description="API server port")
|
||||
|
||||
# Database - default uses data_home/collector/meshcore.db (same as collector)
|
||||
database_url: Optional[str] = Field(
|
||||
default=None,
|
||||
description="SQLAlchemy database URL (default: sqlite:///{data_home}/collector/meshcore.db)",
|
||||
)
|
||||
# Database config (backend selector + connection) is inherited from CommonSettings.
|
||||
|
||||
# Authentication
|
||||
api_read_key: Optional[str] = Field(default=None, description="Read-only API key")
|
||||
@@ -252,23 +316,6 @@ class APISettings(CommonSettings):
|
||||
description="Cache TTL for dashboard endpoints (seconds)",
|
||||
)
|
||||
|
||||
@property
|
||||
def effective_database_url(self) -> str:
|
||||
"""Get the effective database URL, using default if not set."""
|
||||
if self.database_url:
|
||||
return self.database_url
|
||||
from pathlib import Path
|
||||
|
||||
db_path = Path(self.data_home) / "collector" / "meshcore.db"
|
||||
return f"sqlite:///{db_path}"
|
||||
|
||||
@field_validator("database_url")
|
||||
@classmethod
|
||||
def validate_database_url(cls, v: Optional[str]) -> Optional[str]:
|
||||
"""Validate database URL format."""
|
||||
# None is allowed - will use default
|
||||
return v
|
||||
|
||||
|
||||
class WebSettings(CommonSettings):
|
||||
"""Settings for the Web Dashboard component."""
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Database connection and session management."""
|
||||
|
||||
import os
|
||||
from contextlib import asynccontextmanager, contextmanager
|
||||
from typing import Any, AsyncGenerator, Generator
|
||||
|
||||
@@ -11,26 +12,70 @@ from sqlalchemy.orm import Session, sessionmaker
|
||||
from meshcore_hub.common.models.base import Base
|
||||
|
||||
|
||||
def _resolve_pg_schema(database_url: str, schema: str | None) -> str | None:
|
||||
"""Resolve the Postgres schema to scope a connection to (search_path).
|
||||
|
||||
Returns None for SQLite (no schema concept). For Postgres, an explicit ``schema``
|
||||
wins; otherwise it falls back to the ``DATABASE_SCHEMA`` env var. The CLI's
|
||||
``load_dotenv()`` and Docker both populate that var, so runtime entrypoints don't
|
||||
need to thread the schema through every constructor.
|
||||
"""
|
||||
if not database_url.startswith(("postgresql", "postgres")):
|
||||
return None
|
||||
if schema:
|
||||
return schema
|
||||
return os.environ.get("DATABASE_SCHEMA")
|
||||
|
||||
|
||||
def _to_async_url(database_url: str) -> str:
|
||||
"""Map a sync database URL to its async-driver equivalent.
|
||||
|
||||
Postgres always maps to asyncpg for the async engine, even when the sync URL
|
||||
names a sync driver (e.g. ``postgresql+psycopg2://``, which is what the config
|
||||
assembler produces) — otherwise async sessions would try to use the sync driver.
|
||||
SQLite maps to aiosqlite unless a driver is already specified.
|
||||
"""
|
||||
scheme = database_url.split("://", 1)[0]
|
||||
dialect = scheme.split("+", 1)[0]
|
||||
if dialect in ("postgresql", "postgres"):
|
||||
return database_url.replace(f"{scheme}://", "postgresql+asyncpg://", 1)
|
||||
if dialect == "sqlite":
|
||||
if "+" in scheme:
|
||||
return database_url
|
||||
return database_url.replace("sqlite://", "sqlite+aiosqlite://", 1)
|
||||
return database_url
|
||||
|
||||
|
||||
def create_database_engine(
|
||||
database_url: str,
|
||||
echo: bool = False,
|
||||
schema: str | None = None,
|
||||
) -> Engine:
|
||||
"""Create a SQLAlchemy database engine.
|
||||
|
||||
Args:
|
||||
database_url: SQLAlchemy database URL
|
||||
echo: Enable SQL query logging
|
||||
schema: Postgres schema to scope connections to via search_path. Defaults to
|
||||
the DATABASE_SCHEMA env var when not given. Ignored for SQLite.
|
||||
|
||||
Returns:
|
||||
SQLAlchemy Engine instance
|
||||
"""
|
||||
connect_args = {}
|
||||
connect_args: dict[str, Any] = {}
|
||||
engine_kwargs: dict[str, Any] = {}
|
||||
|
||||
# SQLite-specific configuration
|
||||
if database_url.startswith("sqlite"):
|
||||
connect_args["check_same_thread"] = False
|
||||
|
||||
# Scope Postgres connections to the configured schema via search_path. This keeps
|
||||
# the models schema-agnostic (no hardcoded schema=) so the same code serves SQLite,
|
||||
# single-instance Postgres, and multiple schema-isolated instances on one cluster.
|
||||
resolved_schema = _resolve_pg_schema(database_url, schema)
|
||||
if resolved_schema:
|
||||
connect_args["options"] = f"-csearch_path={resolved_schema}"
|
||||
|
||||
# Size the pool above the default Starlette threadpool (~40 threads) so
|
||||
# concurrent request handlers don't block waiting for a connection. Applies
|
||||
# to file-based SQLite and networked backends (e.g. a future Postgres).
|
||||
@@ -110,15 +155,20 @@ class DatabaseManager:
|
||||
to avoid leaking connections when only sync operations are needed.
|
||||
"""
|
||||
|
||||
def __init__(self, database_url: str, echo: bool = False):
|
||||
def __init__(
|
||||
self, database_url: str, echo: bool = False, schema: str | None = None
|
||||
):
|
||||
"""Initialize the database manager.
|
||||
|
||||
Args:
|
||||
database_url: SQLAlchemy database URL
|
||||
echo: Enable SQL query logging
|
||||
schema: Postgres schema to scope connections to (search_path). Defaults to
|
||||
the DATABASE_SCHEMA env var when not given; ignored for SQLite.
|
||||
"""
|
||||
self.database_url = database_url
|
||||
self._echo = echo
|
||||
self._schema = _resolve_pg_schema(database_url, schema)
|
||||
|
||||
# Ensure parent directory exists for SQLite databases
|
||||
if database_url.startswith("sqlite:///"):
|
||||
@@ -128,7 +178,7 @@ class DatabaseManager:
|
||||
db_path = Path(database_url.replace("sqlite:///", ""))
|
||||
db_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
self.engine = create_database_engine(database_url, echo=echo)
|
||||
self.engine = create_database_engine(database_url, echo=echo, schema=schema)
|
||||
self.session_factory = create_session_factory(self.engine)
|
||||
|
||||
# Lazy-initialized async engine (created on first async_session call)
|
||||
@@ -142,8 +192,16 @@ class DatabaseManager:
|
||||
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||
|
||||
async_url = self.database_url.replace("sqlite://", "sqlite+aiosqlite://")
|
||||
self._async_engine = create_async_engine(async_url, echo=self._echo)
|
||||
async_url = _to_async_url(self.database_url)
|
||||
async_connect_args: dict[str, Any] = {}
|
||||
# asyncpg sets search_path via server_settings (not the libpq -c options
|
||||
# string the sync psycopg2 engine uses). self._schema is already resolved
|
||||
# (explicit arg or DATABASE_SCHEMA env) and None for SQLite.
|
||||
if self._schema:
|
||||
async_connect_args["server_settings"] = {"search_path": self._schema}
|
||||
self._async_engine = create_async_engine(
|
||||
async_url, echo=self._echo, connect_args=async_connect_args
|
||||
)
|
||||
|
||||
# Apply the same SQLite pragmas as the sync engine (see
|
||||
# create_database_engine) for the async engine's connections.
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
"""Copy all data from a source (SQLite) database into a target (Postgres) database.
|
||||
|
||||
This powers ``meshcore-hub db migrate-to-postgres``. It operates at the SQLAlchemy
|
||||
Core level, iterating ``Base.metadata.sorted_tables`` and copying each table through
|
||||
the ORM's typed columns. The round-trip ``SQLite value -> Python object -> Postgres
|
||||
value`` is what makes the conversion correct (e.g. integer ``0/1`` -> ``bool``, JSON
|
||||
``TEXT`` -> ``dict``, datetime string -> ``timestamptz``) without any per-model code.
|
||||
|
||||
The schema must already exist in the target (created by ``db upgrade``); this module
|
||||
only moves data and never mutates the source.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import timezone
|
||||
from typing import Optional
|
||||
|
||||
import sqlalchemy as sa
|
||||
from sqlalchemy import func, insert, select, text
|
||||
from sqlalchemy.engine import Engine
|
||||
|
||||
from meshcore_hub.common.database import create_database_engine
|
||||
from meshcore_hub.common.models.base import Base
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class TableResult:
|
||||
"""Per-table outcome of a migration run."""
|
||||
|
||||
name: str
|
||||
source_rows: int
|
||||
target_rows: int
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return self.source_rows == self.target_rows
|
||||
|
||||
|
||||
@dataclass
|
||||
class MigrationResult:
|
||||
"""Aggregate outcome of a migration run."""
|
||||
|
||||
tables: list[TableResult] = field(default_factory=list)
|
||||
dry_run: bool = False
|
||||
|
||||
@property
|
||||
def ok(self) -> bool:
|
||||
return all(t.ok for t in self.tables)
|
||||
|
||||
|
||||
def _tz_aware_columns(table: sa.Table) -> list[str]:
|
||||
"""Names of timezone-aware DateTime columns on a table.
|
||||
|
||||
SQLite does not persist tzinfo, so these values read back naive and must be
|
||||
stamped UTC before insert into Postgres ``timestamptz`` (the app always writes
|
||||
UTC via utc_now()).
|
||||
"""
|
||||
return [
|
||||
col.name
|
||||
for col in table.columns
|
||||
if isinstance(col.type, sa.DateTime)
|
||||
and bool(getattr(col.type, "timezone", False))
|
||||
]
|
||||
|
||||
|
||||
def _count(engine: Engine, table: sa.Table) -> int:
|
||||
with engine.connect() as conn:
|
||||
return int(conn.execute(select(func.count()).select_from(table)).scalar() or 0)
|
||||
|
||||
|
||||
def _is_superuser(engine: Engine) -> bool:
|
||||
"""Whether the target role can SET session_replication_role (superuser-only)."""
|
||||
if engine.dialect.name != "postgresql":
|
||||
return False
|
||||
try:
|
||||
with engine.connect() as conn:
|
||||
value = conn.execute(
|
||||
text("SELECT current_setting('is_superuser')")
|
||||
).scalar()
|
||||
return str(value).lower() == "on"
|
||||
except Exception: # pragma: no cover - defensive
|
||||
return False
|
||||
|
||||
|
||||
def _copy_table(
|
||||
source_engine: Engine,
|
||||
tgt_conn: sa.engine.Connection,
|
||||
table: sa.Table,
|
||||
batch_size: int,
|
||||
) -> int:
|
||||
"""Stream rows from source and bulk-insert into the target connection."""
|
||||
tz_cols = _tz_aware_columns(table)
|
||||
copied = 0
|
||||
with source_engine.connect().execution_options(stream_results=True) as src:
|
||||
result = src.execute(select(table))
|
||||
for partition in result.partitions(batch_size):
|
||||
rows = []
|
||||
for row in partition:
|
||||
data = dict(row._mapping)
|
||||
for col in tz_cols:
|
||||
value = data.get(col)
|
||||
if value is not None and value.tzinfo is None:
|
||||
data[col] = value.replace(tzinfo=timezone.utc)
|
||||
rows.append(data)
|
||||
if rows:
|
||||
tgt_conn.execute(insert(table), rows)
|
||||
copied += len(rows)
|
||||
return copied
|
||||
|
||||
|
||||
def migrate_sqlite_to_postgres(
|
||||
source_url: str,
|
||||
target_url: str,
|
||||
*,
|
||||
target_schema: Optional[str] = None,
|
||||
batch_size: int = 2000,
|
||||
truncate: bool = False,
|
||||
dry_run: bool = False,
|
||||
disable_replication_role: bool = False,
|
||||
) -> MigrationResult:
|
||||
"""Copy every table from ``source_url`` into ``target_url``.
|
||||
|
||||
Args:
|
||||
source_url: SQLAlchemy URL of the source (SQLite) database.
|
||||
target_url: SQLAlchemy URL of the target (Postgres) database.
|
||||
target_schema: Postgres schema to load into (search_path). Defaults to the
|
||||
DATABASE_SCHEMA env var via the engine.
|
||||
batch_size: Rows per insert batch.
|
||||
truncate: Delete existing rows from target tables before loading.
|
||||
dry_run: Report source/target counts without writing.
|
||||
disable_replication_role: Skip the session_replication_role trick even if the
|
||||
target role is a superuser (e.g. to mirror managed-Postgres behaviour).
|
||||
|
||||
Returns:
|
||||
MigrationResult with per-table source/target row counts.
|
||||
"""
|
||||
if not target_url.startswith(("postgresql", "postgres")):
|
||||
raise ValueError("Target must be a PostgreSQL database URL")
|
||||
|
||||
source_engine = create_database_engine(source_url)
|
||||
target_engine = create_database_engine(target_url, schema=target_schema)
|
||||
tables = list(Base.metadata.sorted_tables) # parent-first; excludes alembic_version
|
||||
|
||||
try:
|
||||
# Pre-flight: schema present + (unless truncating) target empty.
|
||||
for table in tables:
|
||||
try:
|
||||
existing = _count(target_engine, table)
|
||||
except Exception as exc: # table missing -> schema not initialised
|
||||
raise RuntimeError(
|
||||
f"Target table {table.name!r} not found. Run 'meshcore-hub db "
|
||||
f"upgrade' against the target first."
|
||||
) from exc
|
||||
if existing and not (truncate or dry_run):
|
||||
raise RuntimeError(
|
||||
f"Target table {table.name!r} is not empty ({existing} rows). "
|
||||
f"Refusing to load; pass --truncate to overwrite."
|
||||
)
|
||||
|
||||
if dry_run:
|
||||
result = MigrationResult(dry_run=True)
|
||||
for table in tables:
|
||||
result.tables.append(
|
||||
TableResult(
|
||||
table.name,
|
||||
_count(source_engine, table),
|
||||
_count(target_engine, table),
|
||||
)
|
||||
)
|
||||
return result
|
||||
|
||||
use_replica = (
|
||||
target_engine.dialect.name == "postgresql"
|
||||
and not disable_replication_role
|
||||
and _is_superuser(target_engine)
|
||||
)
|
||||
if not use_replica:
|
||||
logger.info(
|
||||
"Not disabling FK triggers (session_replication_role); relying on "
|
||||
"parent-first table order."
|
||||
)
|
||||
|
||||
# Single transaction: all-or-nothing. A failure leaves the target empty.
|
||||
with target_engine.begin() as tgt:
|
||||
if use_replica:
|
||||
tgt.execute(text("SET session_replication_role = replica"))
|
||||
if truncate:
|
||||
for table in reversed(tables): # children first
|
||||
tgt.execute(table.delete())
|
||||
for table in tables:
|
||||
copied = _copy_table(source_engine, tgt, table, batch_size)
|
||||
logger.info("Copied %s rows into %s", copied, table.name)
|
||||
if use_replica:
|
||||
tgt.execute(text("SET session_replication_role = DEFAULT"))
|
||||
|
||||
# Reconcile.
|
||||
result = MigrationResult()
|
||||
for table in tables:
|
||||
result.tables.append(
|
||||
TableResult(
|
||||
table.name,
|
||||
_count(source_engine, table),
|
||||
_count(target_engine, table),
|
||||
)
|
||||
)
|
||||
return result
|
||||
finally:
|
||||
source_engine.dispose()
|
||||
target_engine.dispose()
|
||||
@@ -3,8 +3,7 @@
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Index, String
|
||||
from sqlalchemy.dialects.sqlite import JSON
|
||||
from sqlalchemy import DateTime, ForeignKey, Index, JSON, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now
|
||||
|
||||
@@ -8,13 +8,13 @@ from sqlalchemy import (
|
||||
DateTime,
|
||||
Float,
|
||||
ForeignKey,
|
||||
Insert,
|
||||
Integer,
|
||||
Index,
|
||||
String,
|
||||
UniqueConstraint,
|
||||
update,
|
||||
)
|
||||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||||
from sqlalchemy.orm import Mapped, Session, mapped_column, relationship
|
||||
|
||||
from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now
|
||||
@@ -122,21 +122,40 @@ def add_event_observer(
|
||||
|
||||
now = observed_at or datetime.now(timezone.utc)
|
||||
|
||||
stmt = (
|
||||
sqlite_insert(EventObserver)
|
||||
.values(
|
||||
id=str(uuid4()),
|
||||
event_type=event_type,
|
||||
event_hash=event_hash,
|
||||
observer_node_id=observer_node_id,
|
||||
snr=snr,
|
||||
path_len=path_len,
|
||||
observed_at=now,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
# Both SQLite and Postgres expose on_conflict_do_nothing() with the same signature,
|
||||
# but the INSERT construct must come from the matching dialect or it emits SQL for
|
||||
# the wrong backend. Build the statement in each branch (rather than aliasing the
|
||||
# insert() function) so the two dialect-specific Insert types stay distinct.
|
||||
values = {
|
||||
"id": str(uuid4()),
|
||||
"event_type": event_type,
|
||||
"event_hash": event_hash,
|
||||
"observer_node_id": observer_node_id,
|
||||
"snr": snr,
|
||||
"path_len": path_len,
|
||||
"observed_at": now,
|
||||
"created_at": now,
|
||||
"updated_at": now,
|
||||
}
|
||||
conflict_cols = ["event_hash", "observer_node_id"]
|
||||
stmt: Insert
|
||||
|
||||
if session.get_bind().dialect.name == "postgresql":
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
|
||||
stmt = (
|
||||
pg_insert(EventObserver)
|
||||
.values(**values)
|
||||
.on_conflict_do_nothing(index_elements=conflict_cols)
|
||||
)
|
||||
else:
|
||||
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
|
||||
|
||||
stmt = (
|
||||
sqlite_insert(EventObserver)
|
||||
.values(**values)
|
||||
.on_conflict_do_nothing(index_elements=conflict_cols)
|
||||
)
|
||||
.on_conflict_do_nothing(index_elements=["event_hash", "observer_node_id"])
|
||||
)
|
||||
result = session.execute(stmt)
|
||||
rowcount = getattr(result, "rowcount", 0)
|
||||
|
||||
|
||||
@@ -21,7 +21,7 @@ class Message(Base, UUIDMixin, TimestampMixin):
|
||||
text: Message content
|
||||
path_len: Number of hops
|
||||
txt_type: Message type indicator
|
||||
signature: Message signature (8 hex chars)
|
||||
signature: Message signature (hex), or the packet_hash fallback
|
||||
snr: Signal-to-noise ratio
|
||||
sender_timestamp: Sender's timestamp
|
||||
received_at: When received by interface
|
||||
@@ -60,7 +60,7 @@ class Message(Base, UUIDMixin, TimestampMixin):
|
||||
nullable=True,
|
||||
)
|
||||
signature: Mapped[Optional[str]] = mapped_column(
|
||||
String(8),
|
||||
String(32),
|
||||
nullable=True,
|
||||
)
|
||||
snr: Mapped[Optional[float]] = mapped_column(
|
||||
|
||||
@@ -3,8 +3,7 @@
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, String, Text
|
||||
from sqlalchemy.dialects.sqlite import JSON
|
||||
from sqlalchemy import DateTime, Float, ForeignKey, Index, Integer, JSON, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now
|
||||
|
||||
@@ -3,8 +3,7 @@
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from sqlalchemy import DateTime, ForeignKey, Index, LargeBinary, String
|
||||
from sqlalchemy.dialects.sqlite import JSON
|
||||
from sqlalchemy import DateTime, ForeignKey, Index, JSON, LargeBinary, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now
|
||||
|
||||
@@ -3,8 +3,7 @@
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, String
|
||||
from sqlalchemy.dialects.sqlite import JSON
|
||||
from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, JSON, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from meshcore_hub.common.models.base import Base, TimestampMixin, UUIDMixin, utc_now
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Shared pytest fixtures for all tests."""
|
||||
|
||||
import dotenv
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
@@ -7,6 +8,14 @@ from sqlalchemy.orm import sessionmaker
|
||||
from meshcore_hub.common import config as config_module
|
||||
from meshcore_hub.common.models import Base
|
||||
|
||||
# The CLI entrypoint (meshcore_hub.__main__) calls load_dotenv() at import time so
|
||||
# deployments can drop a .env in place. Importing it during collection (e.g. from
|
||||
# test_main.py) would otherwise leak a developer's repo-root .env straight into
|
||||
# os.environ for the whole session — bypassing _ignore_dotenv, which only stops
|
||||
# pydantic-settings from reading the file. conftest.py is imported before any test
|
||||
# module is collected, so neutralising load_dotenv here binds first.
|
||||
dotenv.load_dotenv = lambda *args, **kwargs: False
|
||||
|
||||
|
||||
def _settings_classes():
|
||||
"""CommonSettings and every subclass (recursively)."""
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
"""Tests for configuration settings."""
|
||||
|
||||
import pytest
|
||||
|
||||
from meshcore_hub.common.config import (
|
||||
CommonSettings,
|
||||
CollectorSettings,
|
||||
@@ -258,3 +260,71 @@ class TestWebSettings:
|
||||
|
||||
assert settings.feature_radio_config is False
|
||||
assert settings.features["radio_config"] is False
|
||||
|
||||
|
||||
class TestDatabaseBackendResolution:
|
||||
"""Tests for DATABASE_BACKEND selection and URL/schema resolution."""
|
||||
|
||||
def test_default_backend_is_sqlite_unchanged(self) -> None:
|
||||
"""No DB env vars -> the same SQLite path and no schema as before."""
|
||||
settings = CollectorSettings(_env_file=None, data_home="/data")
|
||||
|
||||
assert settings.database_backend.value == "sqlite"
|
||||
assert (
|
||||
settings.effective_database_url == "sqlite:////data/collector/meshcore.db"
|
||||
)
|
||||
assert settings.effective_database_schema is None
|
||||
|
||||
def test_postgres_backend_assembles_url_and_schema(self) -> None:
|
||||
"""Postgres backend assembles a URL from components and exposes the schema."""
|
||||
settings = APISettings(
|
||||
_env_file=None,
|
||||
database_backend="postgres",
|
||||
database_host="pg",
|
||||
database_password="pw",
|
||||
)
|
||||
|
||||
assert settings.effective_database_url == (
|
||||
"postgresql+psycopg2://meshcorehub:pw@pg:5432/meshcorehub"
|
||||
)
|
||||
assert settings.effective_database_schema == "meshcorehub"
|
||||
|
||||
def test_postgres_password_is_url_encoded(self) -> None:
|
||||
"""Special characters in the password are percent-encoded."""
|
||||
settings = APISettings(
|
||||
_env_file=None,
|
||||
database_backend="postgres",
|
||||
database_host="pg",
|
||||
database_password="s3cr3t/p@ss",
|
||||
)
|
||||
|
||||
assert "s3cr3t%2Fp%40ss" in settings.effective_database_url
|
||||
|
||||
def test_postgres_schema_override_per_instance(self) -> None:
|
||||
"""DATABASE_SCHEMA isolates instances sharing one database."""
|
||||
settings = APISettings(
|
||||
_env_file=None,
|
||||
database_backend="postgres",
|
||||
database_host="pg",
|
||||
database_password="pw",
|
||||
database_schema="stg",
|
||||
)
|
||||
|
||||
assert settings.effective_database_schema == "stg"
|
||||
|
||||
def test_postgres_missing_required_vars_fails_fast(self) -> None:
|
||||
"""Misconfigured postgres backend raises rather than silently using SQLite."""
|
||||
settings = APISettings(_env_file=None, database_backend="postgres")
|
||||
|
||||
with pytest.raises(ValueError, match="DATABASE_BACKEND=postgres requires"):
|
||||
_ = settings.effective_database_url
|
||||
|
||||
def test_explicit_url_overrides_backend(self) -> None:
|
||||
"""An explicit DATABASE_URL wins even when a backend is selected."""
|
||||
settings = CollectorSettings(
|
||||
_env_file=None,
|
||||
database_backend="postgres",
|
||||
database_url="postgresql+psycopg2://u:p@h/db",
|
||||
)
|
||||
|
||||
assert settings.effective_database_url == "postgresql+psycopg2://u:p@h/db"
|
||||
|
||||
@@ -2,9 +2,14 @@
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
|
||||
from meshcore_hub.common.database import create_database_engine
|
||||
from meshcore_hub.common.database import (
|
||||
_resolve_pg_schema,
|
||||
_to_async_url,
|
||||
create_database_engine,
|
||||
)
|
||||
|
||||
|
||||
class TestSqlitePragmas:
|
||||
@@ -34,3 +39,42 @@ class TestSqlitePragmas:
|
||||
assert conn.execute(text("SELECT 1")).scalar() == 1
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
class TestAsyncUrlMapping:
|
||||
"""Map sync URLs to their async-driver equivalents for the async engine."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"sync_url,expected",
|
||||
[
|
||||
("sqlite:///x.db", "sqlite+aiosqlite:///x.db"),
|
||||
("sqlite+aiosqlite:///x.db", "sqlite+aiosqlite:///x.db"),
|
||||
("postgresql://u:p@h/db", "postgresql+asyncpg://u:p@h/db"),
|
||||
("postgres://u:p@h/db", "postgresql+asyncpg://u:p@h/db"),
|
||||
# config assembles +psycopg2; the async engine must still use asyncpg
|
||||
("postgresql+psycopg2://u:p@h/db", "postgresql+asyncpg://u:p@h/db"),
|
||||
("postgresql+asyncpg://u:p@h/db", "postgresql+asyncpg://u:p@h/db"),
|
||||
],
|
||||
)
|
||||
def test_to_async_url(self, sync_url: str, expected: str) -> None:
|
||||
assert _to_async_url(sync_url) == expected
|
||||
|
||||
|
||||
class TestSchemaResolution:
|
||||
"""search_path schema resolution (explicit arg vs DATABASE_SCHEMA env)."""
|
||||
|
||||
def test_sqlite_never_has_schema(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("DATABASE_SCHEMA", "ignored")
|
||||
assert _resolve_pg_schema("sqlite:///x.db", None) is None
|
||||
assert _resolve_pg_schema("sqlite:///x.db", "explicit") is None
|
||||
|
||||
def test_explicit_schema_wins(self) -> None:
|
||||
assert _resolve_pg_schema("postgresql://u@h/db", "prod") == "prod"
|
||||
|
||||
def test_falls_back_to_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("DATABASE_SCHEMA", "stg")
|
||||
assert _resolve_pg_schema("postgresql://u@h/db", None) == "stg"
|
||||
|
||||
def test_none_when_no_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("DATABASE_SCHEMA", raising=False)
|
||||
assert _resolve_pg_schema("postgresql://u@h/db", None) is None
|
||||
|
||||
@@ -0,0 +1,207 @@
|
||||
"""Tests for the SQLite -> Postgres data migration helper.
|
||||
|
||||
The full round-trip is validated against a live Postgres; here we cover the
|
||||
dialect-agnostic pieces (the Postgres-target guard, tz-aware column detection, and
|
||||
the Core copy/stream/normalize logic exercised SQLite -> SQLite) so they run in CI.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, func, select
|
||||
|
||||
from meshcore_hub.common import db_migrate
|
||||
from meshcore_hub.common.db_migrate import (
|
||||
_copy_table,
|
||||
_is_superuser,
|
||||
_tz_aware_columns,
|
||||
migrate_sqlite_to_postgres,
|
||||
)
|
||||
from meshcore_hub.common.models import Base
|
||||
|
||||
# Import models so their tables register on Base.metadata.
|
||||
import meshcore_hub.common.models.node # noqa: F401
|
||||
import meshcore_hub.common.models.raw_packet # noqa: F401
|
||||
|
||||
|
||||
def test_target_must_be_postgres() -> None:
|
||||
"""The command refuses a non-Postgres target."""
|
||||
with pytest.raises(ValueError, match="PostgreSQL"):
|
||||
migrate_sqlite_to_postgres("sqlite:///a.db", "sqlite:///b.db")
|
||||
|
||||
|
||||
def test_tz_aware_columns_detects_timestamptz() -> None:
|
||||
"""Timezone-aware DateTime columns are identified for UTC normalization."""
|
||||
cols = _tz_aware_columns(Base.metadata.tables["raw_packets"])
|
||||
|
||||
assert "received_at" in cols
|
||||
assert "created_at" in cols
|
||||
assert "packet_hash" not in cols # not a datetime
|
||||
|
||||
|
||||
def test_copy_table_roundtrips_rows_and_boolean() -> None:
|
||||
"""_copy_table streams rows across engines, preserving the boolean value.
|
||||
|
||||
(UTC normalization of naive datetimes is validated against Postgres, where
|
||||
timestamptz round-trips reliably; SQLite does not retain tzinfo.)
|
||||
"""
|
||||
nodes = Base.metadata.tables["nodes"]
|
||||
src = create_engine("sqlite:///:memory:")
|
||||
dst = create_engine("sqlite:///:memory:")
|
||||
Base.metadata.create_all(src)
|
||||
Base.metadata.create_all(dst)
|
||||
|
||||
now = datetime(2026, 6, 13, 10, 0)
|
||||
with src.begin() as conn:
|
||||
conn.execute(
|
||||
nodes.insert(),
|
||||
[
|
||||
{
|
||||
"id": f"n{i}",
|
||||
"public_key": f"key{i}",
|
||||
"is_observer": i == 0,
|
||||
"first_seen": now,
|
||||
"created_at": now,
|
||||
"updated_at": now,
|
||||
}
|
||||
for i in range(3)
|
||||
],
|
||||
)
|
||||
|
||||
with dst.begin() as conn:
|
||||
copied = _copy_table(src, conn, nodes, batch_size=2) # forces >1 batch
|
||||
|
||||
assert copied == 3
|
||||
with dst.connect() as conn:
|
||||
assert conn.execute(select(func.count()).select_from(nodes)).scalar() == 3
|
||||
observers = conn.execute(
|
||||
select(func.count()).select_from(nodes).where(nodes.c.is_observer.is_(True))
|
||||
).scalar()
|
||||
assert observers == 1 # boolean preserved across the copy
|
||||
|
||||
src.dispose()
|
||||
dst.dispose()
|
||||
|
||||
|
||||
def test_is_superuser_false_for_non_postgres() -> None:
|
||||
"""session_replication_role is Postgres-only; SQLite is never a superuser target."""
|
||||
engine = create_engine("sqlite:///:memory:")
|
||||
try:
|
||||
assert _is_superuser(engine) is False
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _seed_nodes(engine, count: int) -> None:
|
||||
nodes = Base.metadata.tables["nodes"]
|
||||
now = datetime(2026, 6, 13, 10, 0)
|
||||
with engine.begin() as conn:
|
||||
conn.execute(
|
||||
nodes.insert(),
|
||||
[
|
||||
{
|
||||
"id": f"n{i}",
|
||||
"public_key": f"key{i}",
|
||||
"is_observer": i == 0,
|
||||
"first_seen": now,
|
||||
"created_at": now,
|
||||
"updated_at": now,
|
||||
}
|
||||
for i in range(count)
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def _patch_engines(monkeypatch, tmp_path):
|
||||
"""Route create_database_engine to SQLite files keyed by URL.
|
||||
|
||||
Lets the Postgres-targeting migration flow run end-to-end SQLite -> SQLite in CI:
|
||||
a ``postgresql://`` target URL satisfies the guard while the real engine is a
|
||||
local SQLite file, so the schema/empty/truncate/copy logic is exercised without
|
||||
a live Postgres.
|
||||
"""
|
||||
src_url = f"sqlite:///{tmp_path / 'src.db'}"
|
||||
target_url = "postgresql://fake/target" # satisfies the Postgres guard
|
||||
src_engine = create_engine(src_url)
|
||||
tgt_engine = create_engine(f"sqlite:///{tmp_path / 'tgt.db'}")
|
||||
Base.metadata.create_all(src_engine)
|
||||
|
||||
def fake_create_engine(url, echo=False, schema=None):
|
||||
if url == src_url:
|
||||
return src_engine
|
||||
if url == target_url:
|
||||
return tgt_engine
|
||||
raise AssertionError(f"unexpected url {url!r}")
|
||||
|
||||
monkeypatch.setattr(db_migrate, "create_database_engine", fake_create_engine)
|
||||
return src_url, target_url, src_engine, tgt_engine
|
||||
|
||||
|
||||
def _count_rows(engine) -> int:
|
||||
nodes = Base.metadata.tables["nodes"]
|
||||
with engine.connect() as conn:
|
||||
return int(conn.execute(select(func.count()).select_from(nodes)).scalar() or 0)
|
||||
|
||||
|
||||
def test_migrate_dry_run_reports_counts_without_writing(_patch_engines) -> None:
|
||||
"""Dry run reports source/target counts and leaves the target untouched."""
|
||||
src_url, target_url, src_engine, tgt_engine = _patch_engines
|
||||
Base.metadata.create_all(tgt_engine)
|
||||
_seed_nodes(src_engine, 3)
|
||||
|
||||
result = migrate_sqlite_to_postgres(src_url, target_url, dry_run=True)
|
||||
|
||||
assert result.dry_run is True
|
||||
nodes_result = next(t for t in result.tables if t.name == "nodes")
|
||||
assert nodes_result.source_rows == 3
|
||||
assert nodes_result.target_rows == 0
|
||||
assert _count_rows(tgt_engine) == 0 # nothing written
|
||||
|
||||
|
||||
def test_migrate_copies_all_rows(_patch_engines) -> None:
|
||||
"""A full run copies rows and reconciles source/target counts as OK."""
|
||||
src_url, target_url, src_engine, tgt_engine = _patch_engines
|
||||
Base.metadata.create_all(tgt_engine)
|
||||
_seed_nodes(src_engine, 5)
|
||||
|
||||
result = migrate_sqlite_to_postgres(src_url, target_url, batch_size=2)
|
||||
|
||||
assert result.ok is True
|
||||
assert _count_rows(tgt_engine) == 5
|
||||
nodes_result = next(t for t in result.tables if t.name == "nodes")
|
||||
assert nodes_result.source_rows == nodes_result.target_rows == 5
|
||||
|
||||
|
||||
def test_migrate_refuses_non_empty_target(_patch_engines) -> None:
|
||||
"""Without --truncate, a non-empty target is refused before any write."""
|
||||
src_url, target_url, src_engine, tgt_engine = _patch_engines
|
||||
Base.metadata.create_all(tgt_engine)
|
||||
_seed_nodes(src_engine, 2)
|
||||
_seed_nodes(tgt_engine, 1)
|
||||
|
||||
with pytest.raises(RuntimeError, match="not empty"):
|
||||
migrate_sqlite_to_postgres(src_url, target_url)
|
||||
|
||||
|
||||
def test_migrate_truncate_overwrites_target(_patch_engines) -> None:
|
||||
"""--truncate clears existing target rows before loading from source."""
|
||||
src_url, target_url, src_engine, tgt_engine = _patch_engines
|
||||
Base.metadata.create_all(tgt_engine)
|
||||
_seed_nodes(src_engine, 2)
|
||||
_seed_nodes(tgt_engine, 4)
|
||||
|
||||
result = migrate_sqlite_to_postgres(src_url, target_url, truncate=True)
|
||||
|
||||
assert result.ok is True
|
||||
assert _count_rows(tgt_engine) == 2
|
||||
|
||||
|
||||
def test_migrate_errors_when_target_schema_missing(_patch_engines) -> None:
|
||||
"""A target without the schema (no create_all) fails with a clear message."""
|
||||
src_url, target_url, src_engine, tgt_engine = _patch_engines
|
||||
# Note: target schema intentionally not created.
|
||||
_seed_nodes(src_engine, 1)
|
||||
|
||||
with pytest.raises(RuntimeError, match="Run 'meshcore-hub db upgrade'"):
|
||||
migrate_sqlite_to_postgres(src_url, target_url)
|
||||
@@ -0,0 +1,127 @@
|
||||
"""Tests for the top-level CLI, focused on ``db migrate-to-postgres``.
|
||||
|
||||
The migration engine itself is covered in test_common/test_db_migrate.py; here we
|
||||
verify the command's wiring: option plumbing, dry-run vs. real output, and how the
|
||||
MigrationResult (or an error) maps onto exit codes and messages.
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from click.testing import CliRunner
|
||||
|
||||
from meshcore_hub.__main__ import cli
|
||||
from meshcore_hub.common.db_migrate import MigrationResult, TableResult
|
||||
|
||||
|
||||
def _result(*tables: TableResult, dry_run: bool = False) -> MigrationResult:
|
||||
return MigrationResult(tables=list(tables), dry_run=dry_run)
|
||||
|
||||
|
||||
def test_migrate_to_postgres_success() -> None:
|
||||
"""A matching run reports OK per table and exits 0."""
|
||||
runner = CliRunner()
|
||||
fake = _result(TableResult("nodes", 5, 5))
|
||||
|
||||
with patch(
|
||||
"meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres",
|
||||
return_value=fake,
|
||||
) as mock_migrate:
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
"db",
|
||||
"migrate-to-postgres",
|
||||
"--source",
|
||||
"sqlite:///src.db",
|
||||
"--target",
|
||||
"postgresql://u@h/db",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "nodes" in result.output
|
||||
assert "OK" in result.output
|
||||
assert "Migration complete." in result.output
|
||||
# Flags thread through to the engine call.
|
||||
_, kwargs = mock_migrate.call_args
|
||||
assert kwargs["dry_run"] is False
|
||||
assert kwargs["truncate"] is False
|
||||
|
||||
|
||||
def test_migrate_to_postgres_dry_run() -> None:
|
||||
"""Dry run prints a preview and never renders OK/MISMATCH judgements."""
|
||||
runner = CliRunner()
|
||||
fake = _result(TableResult("nodes", 3, 0), dry_run=True)
|
||||
|
||||
with patch(
|
||||
"meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres",
|
||||
return_value=fake,
|
||||
) as mock_migrate:
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
"db",
|
||||
"migrate-to-postgres",
|
||||
"--source",
|
||||
"sqlite:///src.db",
|
||||
"--target",
|
||||
"postgresql://u@h/db",
|
||||
"--dry-run",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "dry-run" in result.output
|
||||
assert "Dry run complete." in result.output
|
||||
assert "OK" not in result.output
|
||||
assert mock_migrate.call_args.kwargs["dry_run"] is True
|
||||
|
||||
|
||||
def test_migrate_to_postgres_mismatch_exits_nonzero() -> None:
|
||||
"""A row-count mismatch surfaces as a ClickException (non-zero exit)."""
|
||||
runner = CliRunner()
|
||||
fake = _result(TableResult("nodes", 5, 4)) # ok == False
|
||||
|
||||
with patch(
|
||||
"meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres",
|
||||
return_value=fake,
|
||||
):
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
"db",
|
||||
"migrate-to-postgres",
|
||||
"--source",
|
||||
"sqlite:///src.db",
|
||||
"--target",
|
||||
"postgresql://u@h/db",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "MISMATCH" in result.output
|
||||
assert "mismatch" in result.output.lower()
|
||||
|
||||
|
||||
def test_migrate_to_postgres_value_error_becomes_click_exception() -> None:
|
||||
"""A ValueError from the engine (e.g. bad target) maps to a clean CLI error."""
|
||||
runner = CliRunner()
|
||||
|
||||
with patch(
|
||||
"meshcore_hub.common.db_migrate.migrate_sqlite_to_postgres",
|
||||
side_effect=ValueError("Target must be a PostgreSQL database URL"),
|
||||
):
|
||||
result = runner.invoke(
|
||||
cli,
|
||||
[
|
||||
"db",
|
||||
"migrate-to-postgres",
|
||||
"--source",
|
||||
"sqlite:///src.db",
|
||||
"--target",
|
||||
"sqlite:///bad.db",
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code != 0
|
||||
assert "PostgreSQL" in result.output
|
||||
Reference in New Issue
Block a user