diff --git a/CHANGELOG.md b/CHANGELOG.md index 8337d38..0e91351 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **Postgres is selectable from `Server`** (`jvspatial/api/components/database_configurator.py`). + `initialize_graph_context()` now builds a `PostgresDB` prime database for + `db_type="postgres"` (alias `"postgresql"`) instead of raising + `ValueError: Unsupported database type: postgres`. The backend and + `create_database("postgres", ...)` already worked; only the `Server` path + was missing, so every API-layer deployment — and anything built on it — was + locked out of a documented backend. + +- **`ServerConfig.database` carries Postgres settings** (`jvspatial/api/config_groups.py`, + `jvspatial/env_adapter.py`) — `postgres_dsn`, `postgres_min_pool_size`, + `postgres_max_pool_size`, `postgres_pooler_mode`, populated from the + already-allowlisted `JVSPATIAL_POSTGRES_*` env keys. Connection settings now + flow through the config object like every other backend's rather than being + readable only by the driver. Unset values still defer to `PostgresDB`'s own + defaults. Coverage: `tests/test_env_adapter_postgres.py`, + `tests/api/components/test_database_configurator.py`. + - **`resolve_sort_value(record, field)`** (`jvspatial/db/database.py`) — the dotted-path resolution `finalize_find_results` uses, exported so adapters and cursor logic resolve a sort field the same way. Added to the module's @@ -24,6 +41,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **`PostgresDB` pool is event-loop aware** (`jvspatial/db/postgres.py`). The + `asyncpg` pool and its lock bind to the loop that created them, but + `_ensure_pool()` memoized both for the lifetime of the instance. A host that + bootstrapped in one `asyncio.run()` and then served from a second loop hit + `cannot perform operation: another operation is in progress` / + `ConnectionDoesNotExistError` on its first query. The pool and lock are now + rebuilt when the running loop changes. Coverage: + `tests/db/test_postgres_unit.py::TestPoolLoopAffinity`. + - **Partial-index repair log is INFO, not WARNING** (`jvspatial/db/sqlite.py`). Dropping a non-partial index so it can be recreated with `WHERE` is expected one-shot migration noise; log at info. Also satisfy ruff SIM110 in diff --git a/docs/md/postgres-guide.md b/docs/md/postgres-guide.md index 6bbf593..06b7fb5 100644 --- a/docs/md/postgres-guide.md +++ b/docs/md/postgres-guide.md @@ -77,6 +77,37 @@ alice = await User.create(name="Alice", email="alice@example.com") loaded = await User.get(alice.id) ``` +### Using Postgres with `Server` + +`Server` selects the prime database from `db_type`, so Postgres is available +to the API layer the same way JSON or MongoDB is: + +```python +from jvspatial.api.server import Server + +server = Server( + db_type="postgres", + postgres_dsn="postgresql://user:pw@localhost:5432/mydb", +) +``` + +`postgresql` is accepted as an alias for `postgres`. Every connection setting +also resolves from the environment through `ServerConfig`: + +```bash +JVSPATIAL_DB_TYPE=postgres +JVSPATIAL_POSTGRES_DSN=postgresql://user:pw@localhost:5432/mydb +``` + +Settings you leave unset fall through to `PostgresDB`'s own defaults, so the +DSN is the only value most deployments need. `JVSPATIAL_DB_PATH` does not +apply — it is a file-backend setting. + +Note that the logging database is a separate store and has no Postgres +backend; if `JVSPATIAL_LOG_DB_TYPE` is unset it inherits `JVSPATIAL_DB_TYPE` +and falls back to a JSON file log. Set it explicitly when the graph runs on +Postgres. + ## Schema Each collection becomes one table. The shape: @@ -152,6 +183,15 @@ JVSPATIAL_POSTGRES_MIN_POOL_SIZE=5 JVSPATIAL_POSTGRES_MAX_POOL_SIZE=25 ``` +### Event loops + +The pool belongs to the event loop that created it. If a host bootstraps +inside one `asyncio.run()` and then serves from a second loop — the common +CLI-then-ASGI-server startup — `PostgresDB` notices the loop change and +rebuilds the pool on first use in the new loop. The abandoned pool's +connections are dropped rather than closed, since closing them would mean +awaiting on a loop that no longer runs. + ### Pooler compatibility (PgBouncer / RDS Proxy) Transaction-mode poolers (PgBouncer transaction-pooling, AWS RDS Proxy) reuse diff --git a/jvspatial/api/components/database_configurator.py b/jvspatial/api/components/database_configurator.py index a486c64..1bf88cb 100644 --- a/jvspatial/api/components/database_configurator.py +++ b/jvspatial/api/components/database_configurator.py @@ -79,6 +79,25 @@ def _resolve_mongodb_connection(self) -> Tuple[str, str]: db_name = (db.db_database_name or "").strip() or "jvdb" return uri, db_name + def _resolve_postgres_kwargs(self) -> Dict[str, Any]: + """Resolve PostgresDB connection kwargs from server configuration. + + Only settings the operator actually supplied are returned, so + anything left unset falls through to ``PostgresDB``'s own env + defaults rather than being pinned to a value here. + """ + db = self.config.database + kwargs: Dict[str, Any] = {} + if db.postgres_dsn: + kwargs["dsn"] = db.postgres_dsn + if db.postgres_min_pool_size is not None: + kwargs["min_size"] = db.postgres_min_pool_size + if db.postgres_max_pool_size is not None: + kwargs["max_size"] = db.postgres_max_pool_size + if db.postgres_pooler_mode: + kwargs["pooler_mode"] = db.postgres_pooler_mode + return kwargs + def _resolve_observability_kwargs(self) -> Dict[str, Any]: """Resolve optional DB observability wrapper kwargs from env.""" raw_enabled = str(os.environ.get("JVSPATIAL_OBSERVABILITY_ENABLED", "")).lower() @@ -173,6 +192,12 @@ def initialize_graph_context(self) -> Optional[GraphContext]: aws_secret_access_key=self.config.database.dynamodb_secret_access_key, **observability_kwargs, ) + elif db_type in ("postgres", "postgresql"): + prime_db = create_database( + db_type="postgres", + **self._resolve_postgres_kwargs(), + **observability_kwargs, + ) else: raise ValueError(f"Unsupported database type: {db_type}") diff --git a/jvspatial/api/config_groups.py b/jvspatial/api/config_groups.py index 7aef8ab..3105cf7 100644 --- a/jvspatial/api/config_groups.py +++ b/jvspatial/api/config_groups.py @@ -45,6 +45,23 @@ class DatabaseConfig(BaseModel): default=None, validation_alias="AWS_SECRET_ACCESS_KEY" ) + # PostgreSQL Configuration (only used if db_type is "postgres"/"postgresql"). + # Unset values fall through to PostgresDB's own env defaults. + postgres_dsn: Optional[str] = Field( + default=None, validation_alias="JVSPATIAL_POSTGRES_DSN" + ) + postgres_min_pool_size: Optional[int] = Field( + default=None, validation_alias="JVSPATIAL_POSTGRES_MIN_POOL_SIZE" + ) + postgres_max_pool_size: Optional[int] = Field( + default=None, validation_alias="JVSPATIAL_POSTGRES_MAX_POOL_SIZE" + ) + postgres_pooler_mode: Optional[str] = Field( + default=None, + validation_alias="JVSPATIAL_POSTGRES_POOLER_MODE", + description="'session' (default) or 'transaction' for PgBouncer / RDS Proxy.", + ) + class SecurityConfig(BaseModel): """Security configuration group.""" diff --git a/jvspatial/db/postgres.py b/jvspatial/db/postgres.py index 690615b..8f01d70 100644 --- a/jvspatial/db/postgres.py +++ b/jvspatial/db/postgres.py @@ -337,6 +337,10 @@ def __init__( self._pool: Optional["Pool"] = None self._pool_lock = asyncio.Lock() + # The loop ``_pool`` (and ``_pool_lock``) belong to. asyncpg pools + # and asyncio primitives bind to the loop they were created on, so + # a pool carried into a second loop is unusable. + self._pool_loop: Optional[asyncio.AbstractEventLoop] = None # Collections we've already created the table + base indexes for. # Avoids running CREATE TABLE IF NOT EXISTS on the hot path. @@ -350,8 +354,39 @@ def __init__( # ---- pool lifecycle ---------------------------------------------------- + def _discard_pool_from_dead_loop(self) -> None: + """Drop a pool that belongs to a different event loop. + + Hosts that bootstrap inside one ``asyncio.run()`` and then serve + from a second loop (the CLI-then-ASGI-server pattern) would + otherwise reuse a pool bound to the first, now-closed loop, and + every query fails with ``cannot perform operation: another + operation is in progress``. The stale pool's connections are + abandoned rather than closed — ``close()`` would have to await on + the dead loop — and Postgres reaps them when the sockets drop. + """ + if self._pool is None: + return + try: + running = asyncio.get_running_loop() + except RuntimeError: # pragma: no cover - callers are async + return + if self._pool_loop is running: + return + + logger.debug( + "PostgresDB: event loop changed; rebuilding pool (was %r, now %r)", + self._pool_loop, + running, + ) + self._pool = None + # ``asyncio.Lock`` binds to its loop too, so it has to go as well. + self._pool_lock = asyncio.Lock() + self._collections_bootstrapped.clear() + async def _ensure_pool(self) -> "Pool": """Lazily create the asyncpg pool. Idempotent + concurrency-safe.""" + self._discard_pool_from_dead_loop() if self._pool is not None: return self._pool async with self._pool_lock: @@ -377,6 +412,7 @@ async def _ensure_pool(self) -> "Pool": self.pooler_mode, ) self._pool = await asyncpg.create_pool(**create_kwargs) + self._pool_loop = asyncio.get_running_loop() return self._pool async def close(self) -> None: @@ -384,6 +420,7 @@ async def close(self) -> None: if self._pool is not None: await self._pool.close() self._pool = None + self._pool_loop = None self._collections_bootstrapped.clear() # ---- tenant scope (C6) ------------------------------------------------- diff --git a/jvspatial/env_adapter.py b/jvspatial/env_adapter.py index 1593b5d..c65523d 100644 --- a/jvspatial/env_adapter.py +++ b/jvspatial/env_adapter.py @@ -124,9 +124,17 @@ def server_config_overrides_from_env() -> Dict[str, Any]: ("JVSPATIAL_DYNAMODB_TABLE_NAME", "dynamodb_table_name"), ("JVSPATIAL_DYNAMODB_REGION", "dynamodb_region"), ("JVSPATIAL_DYNAMODB_ENDPOINT_URL", "dynamodb_endpoint_url"), + ("JVSPATIAL_POSTGRES_DSN", "postgres_dsn"), + ("JVSPATIAL_POSTGRES_POOLER_MODE", "postgres_pooler_mode"), ): if (t := _opt_str(ek)) is not None: db[dk] = t + for ek, dk in ( + ("JVSPATIAL_POSTGRES_MIN_POOL_SIZE", "postgres_min_pool_size"), + ("JVSPATIAL_POSTGRES_MAX_POOL_SIZE", "postgres_max_pool_size"), + ): + if (n := _opt_int(ek)) is not None: + db[dk] = n if db: o["database"] = db diff --git a/tests/api/components/test_database_configurator.py b/tests/api/components/test_database_configurator.py index 05b72c7..9b4c5a0 100644 --- a/tests/api/components/test_database_configurator.py +++ b/tests/api/components/test_database_configurator.py @@ -249,6 +249,90 @@ async def test_initialize_graph_context_dynamodb(self): slow_query_ms=100.0, ) + @pytest.mark.asyncio + async def test_initialize_graph_context_postgres(self): + """Test GraphContext initialization with PostgreSQL.""" + config = ServerConfig() + config.database.db_type = "postgres" + config.database.postgres_dsn = "postgresql://u:p@localhost:5432/testdb" + config.database.postgres_min_pool_size = 3 + config.database.postgres_max_pool_size = 7 + config.database.postgres_pooler_mode = "transaction" + configurator = DatabaseConfigurator(config) + + with patch( + "jvspatial.api.components.database_configurator.create_database" + ) as mock_create: + mock_db = MagicMock() + mock_create.return_value = mock_db + + with patch( + "jvspatial.api.components.database_configurator.get_database_manager" + ) as mock_get_manager: + mock_manager = MagicMock() + mock_manager.get_current_database.return_value = mock_db + mock_get_manager.side_effect = RuntimeError() + + with patch( + "jvspatial.api.components.database_configurator.set_database_manager" + ): + with patch( + "jvspatial.api.components.database_configurator.GraphContext" + ) as mock_context: + mock_ctx_instance = MagicMock() + mock_context.return_value = mock_ctx_instance + + with patch( + "jvspatial.api.components.database_configurator.set_default_context" + ): + result = configurator.initialize_graph_context() + + assert result == mock_ctx_instance + mock_create.assert_called_once_with( + db_type="postgres", + dsn="postgresql://u:p@localhost:5432/testdb", + min_size=3, + max_size=7, + pooler_mode="transaction", + observe=False, + slow_query_ms=100.0, + ) + + @pytest.mark.asyncio + async def test_initialize_graph_context_postgresql_alias(self): + """``postgresql`` is accepted as an alias for ``postgres``.""" + config = ServerConfig() + config.database.db_type = "postgresql" + configurator = DatabaseConfigurator(config) + + with patch( + "jvspatial.api.components.database_configurator.create_database" + ) as mock_create: + mock_create.return_value = MagicMock() + with patch( + "jvspatial.api.components.database_configurator.get_database_manager", + side_effect=RuntimeError(), + ): + with patch( + "jvspatial.api.components.database_configurator.set_database_manager" + ): + with patch( + "jvspatial.api.components.database_configurator.GraphContext", + return_value=MagicMock(), + ): + with patch( + "jvspatial.api.components.database_configurator.set_default_context" + ): + configurator.initialize_graph_context() + + # Unset connection settings are omitted so PostgresDB's own env + # defaults still apply. + mock_create.assert_called_once_with( + db_type="postgres", + observe=False, + slow_query_ms=100.0, + ) + @pytest.mark.asyncio async def test_initialize_graph_context_no_db_type(self, configurator): """Test that None is returned when db_type is not set.""" diff --git a/tests/db/test_postgres_unit.py b/tests/db/test_postgres_unit.py index aeb3099..d9df858 100644 --- a/tests/db/test_postgres_unit.py +++ b/tests/db/test_postgres_unit.py @@ -248,3 +248,87 @@ def test_malformed_vector_left_to_translator_to_reject(self) -> None: filtered, field, _, _, _ = db._pop_vector_clause("doc", q) assert filtered == q # untouched assert field is None + + +class TestPoolLoopAffinity: + """A pool belongs to the loop that created it. + + Hosts that bootstrap in one ``asyncio.run()`` and then serve from a + second loop (the CLI-then-uvicorn pattern) would otherwise reuse a + pool bound to a dead loop, and every query fails with + ``cannot perform operation: another operation is in progress``. + """ + + @staticmethod + def _fake_pool() -> object: + class _Pool: + pass + + return _Pool() + + def test_pool_reused_within_one_loop(self, monkeypatch: pytest.MonkeyPatch) -> None: + db = PostgresDB(dsn="postgresql://nope/none") + created = [] + + async def fake_create_pool(**_kwargs: object) -> object: + pool = self._fake_pool() + created.append(pool) + return pool + + monkeypatch.setattr("asyncpg.create_pool", fake_create_pool) + + async def scenario() -> None: + first = await db._ensure_pool() + second = await db._ensure_pool() + assert first is second + + asyncio.run(scenario()) + assert len(created) == 1 + + def test_pool_rebuilt_after_loop_change( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + db = PostgresDB(dsn="postgresql://nope/none") + created = [] + + async def fake_create_pool(**_kwargs: object) -> object: + pool = self._fake_pool() + created.append(pool) + return pool + + monkeypatch.setattr("asyncpg.create_pool", fake_create_pool) + + pools: list = [] + # Two separate asyncio.run() calls == two distinct event loops. + asyncio.run(_collect_pool(db, pools)) + asyncio.run(_collect_pool(db, pools)) + + assert len(created) == 2 + assert pools[0] is not pools[1] + + def test_lock_is_not_bound_to_the_dead_loop( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The rebuild must replace ``_pool_lock`` too. + + ``asyncio.Lock`` binds to the loop it is first awaited on, so + carrying the old lock across would trade one cross-loop failure + for another. + """ + db = PostgresDB(dsn="postgresql://nope/none") + + async def fake_create_pool(**_kwargs: object) -> object: + return self._fake_pool() + + monkeypatch.setattr("asyncpg.create_pool", fake_create_pool) + + pools: list = [] + asyncio.run(_collect_pool(db, pools)) + first_lock = db._pool_lock + + asyncio.run(_collect_pool(db, pools)) + assert db._pool_lock is not first_lock + + +async def _collect_pool(db: PostgresDB, sink: list) -> None: + sink.append(await db._ensure_pool()) diff --git a/tests/test_env_adapter_postgres.py b/tests/test_env_adapter_postgres.py new file mode 100644 index 0000000..faae296 --- /dev/null +++ b/tests/test_env_adapter_postgres.py @@ -0,0 +1,53 @@ +"""``JVSPATIAL_POSTGRES_*`` env keys map onto ServerConfig.database.""" + +from __future__ import annotations + +import pytest + +from jvspatial.env_adapter import server_config_overrides_from_env + +_POSTGRES_ENV_KEYS = ( + "JVSPATIAL_POSTGRES_DSN", + "JVSPATIAL_POSTGRES_MIN_POOL_SIZE", + "JVSPATIAL_POSTGRES_MAX_POOL_SIZE", + "JVSPATIAL_POSTGRES_POOLER_MODE", +) + + +@pytest.fixture(autouse=True) +def _clear_postgres_env(monkeypatch: pytest.MonkeyPatch) -> None: + """Host ``.env`` values must not leak into these assertions.""" + for key in _POSTGRES_ENV_KEYS: + monkeypatch.delenv(key, raising=False) + + +def test_postgres_env_maps_into_database_group( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("JVSPATIAL_DB_TYPE", "postgres") + monkeypatch.setenv("JVSPATIAL_POSTGRES_DSN", "postgresql://u:p@host:5432/db") + monkeypatch.setenv("JVSPATIAL_POSTGRES_MIN_POOL_SIZE", "1") + monkeypatch.setenv("JVSPATIAL_POSTGRES_MAX_POOL_SIZE", "4") + monkeypatch.setenv("JVSPATIAL_POSTGRES_POOLER_MODE", "transaction") + + db = server_config_overrides_from_env()["database"] + + assert db["db_type"] == "postgres" + assert db["postgres_dsn"] == "postgresql://u:p@host:5432/db" + assert db["postgres_min_pool_size"] == 1 + assert db["postgres_max_pool_size"] == 4 + assert db["postgres_pooler_mode"] == "transaction" + + +def test_postgres_keys_absent_when_unset(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("JVSPATIAL_DB_TYPE", "json") + + db = server_config_overrides_from_env().get("database", {}) + + for key in ( + "postgres_dsn", + "postgres_min_pool_size", + "postgres_max_pool_size", + "postgres_pooler_mode", + ): + assert key not in db