Phase 3: Postgres container + make migrations Postgres-clean

- compose: add optional postgres service (postgres:17-alpine, profile
  'postgres', healthcheck, postgres_data volume); POSTGRES_* derive from
  DATABASE_* (single source of truth). DATABASE_* env added to migrate/
  collector/api; migrate depends_on postgres with required:false so SQLite
  deployments are unaffected.
- alembic/env: resolve the URL via CommonSettings.effective_database_url so
  DATABASE_BACKEND=postgres is honoured (previously DATABASE_URL/DATA_HOME
  only -> would silently migrate SQLite).
- migrations: normalize_public_key uses STRING_AGG + HAVING COUNT(*) on
  Postgres (was SQLite GROUP_CONCAT + alias); raw_packets uses sa.JSON()
  not the sqlite dialect type.
- database: fix _to_async_url to map postgresql+psycopg2:// (what the config
  assembles) to asyncpg, so API async sessions work on Postgres; resolve the
  search_path schema from DATABASE_SCHEMA env when not passed explicitly.

Validated against a live postgres:17: db upgrade builds all 13 tables in the
meshcorehub schema with correct native types (is_observer boolean, decoded
json) and alembic_version stamped; the upsert, JSON/timestamptz round-trip,
and asyncpg async sessions all work. SQLite suite still green (1061 passed).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Louis KingandClaude Opus 4.8 committed 2026-06-13 22:26:45 +01:00
1 parent f342f5bc70
commit caef666c02
6 files changed
+168 -38

No files matched your search

+13 -16
View File
@@ -20,25 +20,22 @@ 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:
@@ -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(