From 07c1099ef9da90d135c797372b2536622a1d6927 Mon Sep 17 00:00:00 2001 From: Eldon Marks Date: Tue, 13 Oct 2026 03:49:38 -0400 Subject: [PATCH] fix(db): complete native transaction reads and release 0.2.2 --- AGENTS.md | 4 +- CHANGELOG.md | 29 + PRD.md | 2 +- README.md | 2 +- ROADMAP.md | 4 +- SPEC.md | 44 +- docs/md/README.md | 21 +- docs/md/api-architecture.md | 562 +++----------- docs/md/authentication.md | 4 +- docs/md/cursor-pagination.md | 6 +- docs/md/design-decisions.md | 4 +- docs/md/error-handling.md | 6 +- docs/md/examples.md | 777 +------------------- docs/md/file-storage-usage.md | 10 +- docs/md/graph-visualization.md | 2 +- docs/md/migration.md | 4 +- docs/md/observability.md | 49 +- docs/md/optimization.md | 6 +- docs/md/postgres-guide.md | 8 + docs/md/quick-start-guide.md | 2 +- docs/md/transactions.md | 109 +++ jvspatial/core/context.py | 26 +- jvspatial/db/README.md | 9 +- jvspatial/db/_observable.py | 58 +- jvspatial/db/factory.py | 13 +- jvspatial/db/mongodb.py | 8 +- jvspatial/db/postgres.py | 456 +++++++++--- jvspatial/db/transaction.py | 148 +++- jvspatial/observability/README.md | 4 +- jvspatial/observability/__init__.py | 4 +- jvspatial/version.py | 2 +- tests/core/test_graph_transaction.py | 45 ++ tests/db/test_bounded_query_execution.py | 43 +- tests/db/test_mongo_async_driver.py | 48 +- tests/db/test_observable_database.py | 91 +-- tests/db/test_postgres_integration.py | 90 +++ tests/db/test_postgres_transaction_reads.py | 97 +++ tests/db/test_transaction_api.py | 123 ++++ 38 files changed, 1383 insertions(+), 1537 deletions(-) create mode 100644 docs/md/transactions.md create mode 100644 tests/db/test_postgres_transaction_reads.py create mode 100644 tests/db/test_transaction_api.py diff --git a/AGENTS.md b/AGENTS.md index 439a4de..4101587 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -90,7 +90,7 @@ OAuth hosts in production must set `JVSPATIAL_OAUTH_KEY_ENCRYPTION_KEY` from a s ### 11. Cross-context persistence -The in-memory matcher and PostgreSQL pushdown treat both missing and JSON `null` as absent for `$exists`; MongoDB retains its native field-presence semantics. `graph_transaction` isolates request identity state and invalidates touched entities in the parent cache after commit or rollback. Index setup must be scoped to the database instance so one store cannot suppress another store's indexes. Preserve these contracts when changing query translation, caches, or transactions. +The in-memory matcher and PostgreSQL pushdown treat both missing and JSON `null` as absent for `$exists`; MongoDB retains its native field-presence semantics. `graph_transaction` isolates request identity state, never returns outer-context cached entities inside a transaction, and invalidates touched entities in the parent cache after commit or rollback. Native transaction handles retain the shared read contract on their own connection/session; unsupported backends must continue to fail capability checks. Index setup must be scoped to the database instance so one store cannot suppress another store's indexes. Preserve these contracts when changing query translation, caches, or transactions. Composite keyset continuation includes every sort value and direction plus ID. Never apply LIMIT before residual filtering or unsupported ordering. A configured fallback scan budget raises on exhaustion rather than returning a partial result. Bulk cache entries require confirmed outcomes; ambiguous writes are invalidated. Preserve JSON path locks/atomic durability and streaming handle cleanup on cancellation. @@ -126,7 +126,7 @@ See [graph-projection.md](docs/md/graph-projection.md) and SPEC §6.9. | Subclass lookup returns wrong class | Two unrelated classes share `__name__` | Set `__entity_name__: ClassVar[Optional[str]] = "Distinct"` on one | | Tests pass locally, fail under serverless mode | Forgot to test both modes | Use `reset_serverless_mode_cache()` between tests | | 401 with valid token | Auth state on wrong database | Auth is **always** on prime DB; do not relocate | -| Slow query never logged | `slow_query_ms` not configured | Set in `create_database(observe=True, slow_query_ms=N)` | +| Slow query absent from logs | Successful operations are deliberately silent | Set `observe=True` with a metrics recorder; inspect the slow-operation counter | | Walker visits the same node forever | Protection disabled | Re-enable `protection_enabled=True` | | `JVSPATIAL_FOO` env var ignored | Not in allowlist | Add to `jvspatial/env_adapter.py` allowlist | | SQLite query falls back to in-memory filter | Operator not yet pushed down by `SQLiteTranslator` | Check `jvspatial/db/_sqlite_translate.py` for supported operators | diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f66339..a288f4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.2.2] - 2026-10-13 + +### Fixed + +- Transaction reads now expose `find_one`, `count`, `find_ids`, `find_many`, + `find_iter` and incident-edge counts through the shared transaction contract. + PostgreSQL uses its held connection for native reads; MongoDB fallbacks retain + their session. Unsupported backends still advertise no native transactions. +- PostgreSQL transaction operations serialize access to their connection, + reject use after completion, retain tenant scope, and bootstrap new collections + within the same transaction without acquiring a second connection. Collection + bootstrap caches are promoted only after commit. +- Graph transactions no longer reuse entities bound to an outer context or + populate shared process caches with uncommitted data. Parent-cache invalidation + remains in place after commit and rollback. + +- Graph transaction activation now fails closed when a deployment returns no + transaction handle. Shared transaction contexts roll back on cancellation, + and MongoDB commit/rollback await asynchronous session cleanup. + +### Changed + +- Successful database observations produce no log records, including slow calls + and DEBUG-enabled runs. Failures emit structured ERROR records; operation and + slow-call metrics remain available. Successful calls also skip log-payload + construction and message formatting. +- Updated transaction, paging, observability and backend support documentation; + corrected stale specification and roadmap claims and refreshed the docs index. + ## [0.2.1] - 2026-10-12 ### Fixed diff --git a/PRD.md b/PRD.md index 29be4f7..f48fe1f 100644 --- a/PRD.md +++ b/PRD.md @@ -14,7 +14,7 @@ jvspatial is an **async-first, serverless-compatible, object-spatial Python libr It is *not*: -- A graph database engine. jvspatial layers a graph model on top of existing backends (MongoDB, SQLite, JSON files, DynamoDB) and unifies query semantics across them. +- A graph database engine. jvspatial layers a graph model on top of existing backends (PostgreSQL, MongoDB, SQLite, JSON files, DynamoDB) and unifies query semantics across them. - A web framework. It builds on FastAPI; it does not replace it. - A workflow engine. Walkers traverse data; they are not a generic task orchestrator. - A UI library. Visualization helpers exist (DOT, Mermaid), but rendering is out of scope. diff --git a/README.md b/README.md index 24c849e..89bcb4b 100644 --- a/README.md +++ b/README.md @@ -273,7 +273,7 @@ server = Server( - [Custom Database Guide](https://github.com/TrueSelph/jvspatial/blob/main/docs/md/custom-database-guide.md) - Implementing custom database backends - [Graph Visualization](https://github.com/TrueSelph/jvspatial/blob/main/docs/md/graph-visualization.md) - Export graphs in DOT/Mermaid formats - [Pagination](https://github.com/TrueSelph/jvspatial/blob/main/docs/md/pagination.md) - ObjectPager usage -- [Observability](https://github.com/TrueSelph/jvspatial/blob/main/docs/md/observability.md) - Structured DB logging, slow-query threshold, MetricsRecorder, OpenTelemetry adapter +- [Observability](https://github.com/TrueSelph/jvspatial/blob/main/docs/md/observability.md) - Failure-only DB logging, operation/slow-call metrics, OpenTelemetry adapter - [Performance Benchmarks](https://github.com/TrueSelph/jvspatial/blob/main/docs/md/benchmarks.md) - Regression-detection bench suite (pytest-benchmark + CI workflow) - [API Stability](https://github.com/TrueSelph/jvspatial/blob/main/docs/md/stability.md) - Public/internal/experimental tiers and deprecation policy diff --git a/ROADMAP.md b/ROADMAP.md index 39e9107..6893b0e 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -24,7 +24,7 @@ Recent waves have systematically eliminated constant-time-comparison gaps, fail- - `find_many`, `bulk_save`, native `count()` pushdown landed in 0.0.7. Continue: - Native bulk-update equivalents on backends that support them (Mongo `bulk_write`, SQLite `executemany UPDATE`). -- Continue surfacing slow-query telemetry via the observability layer; tune `slow_query_ms` defaults per backend. +- Surface slow-query counts and durations through observability metrics; successful operations remain silent in application logs. Tune `slow_query_ms` for workload-specific metrics. ### 1.3 Observability (active) @@ -53,7 +53,7 @@ No migration framework. Field removals or renames break existing records silentl ### 2.2 Multi-database transactions are best-effort only -Only MongoDB supports ACID transactions. Cross-backend operations have no built-in coordination. +PostgreSQL and topology-qualified MongoDB support native ACID transactions. Cross-backend operations have no built-in coordination; JSON, SQLite and DynamoDB do not offer native graph transactions. - **Decision needed**: document the limit explicitly in handler patterns or build an opt-in saga primitive. - **Location**: `jvspatial/db/transaction.py`, `jvspatial/db/database.py:48+`. diff --git a/SPEC.md b/SPEC.md index 88d9de0..12e0901 100644 --- a/SPEC.md +++ b/SPEC.md @@ -274,7 +274,7 @@ cases are documented rather than normalized: Adapters declare capabilities as class attributes: -- `supports_transactions: bool` — `True` for MongoDB (replica set); `False` for SQLite (best-effort), JSON, DynamoDB. +- `supports_transactions: bool` — the adapter-level flag is `True` for PostgreSQL and MongoDB and `False` for SQLite, JSON and DynamoDB. MongoDB also requires a successful `await db.is_transactional()` topology probe; its static flag alone is not deployment qualification. Unsupported backends do not offer multi-write atomicity. Node adjacency always lives in the edge collection (indexed on `source` / `target` by `Edge.get_indexes`). There is no `edge_ids_mode` toggle and no @@ -292,9 +292,33 @@ Callers branching on capabilities should test the flag, not the adapter class. | SQLite | `jvspatial/db/sqlite.py` | No (single-conn fsync) | `aiosqlite`; Mongo→SQL via `SQLiteTranslator` | | MongoDB | `jvspatial/db/mongodb.py` | Yes | Motor default; opt-in PyMongo Async; native bulk/cursor/compound ops | | DynamoDB | `jvspatial/db/dynamodb.py` | No | `aioboto3`; throttle-retry; `BatchGetItem` chunks of 100 | +| PostgreSQL | `jvspatial/db/postgres.py` | Yes | Held-connection ACID transactions, RLS tenant scope and native query paging | For a file-backed SQLite database reused on a different event loop, the adapter closes the old `aiosqlite` connection and its worker thread before opening a connection on the new loop. An in-memory SQLite database retains its existing connection so its contents are not lost (`jvspatial/db/sqlite.py`). +### 4.3.1 Transaction read and cache contract + +`Transaction` supplies `find_one`, `count`, `find_ids`, `find_many`, `find_iter` +and `count_incident_edges` alongside CRUD. All fallback reads route through the +transaction handle, not its outer database. PostgreSQL overrides native reads +on its held connection, serializes concurrent operations, rejects completed +handles and keeps tenant scope and collection bootstrap on that connection. +MongoDB inherited reads retain the native session; fallback count/projection/ +bulk work may be less efficient than native aggregation. JSON, SQLite and +DynamoDB remain unsupported for native graph transactions. If an adapter cannot +open a handle, `graph_transaction` raises `TransactionUnavailable` without +binding an ordinary context. `transaction_context` rolls back on cancellation; +MongoDB completion awaits native session cleanup for both driver shapes. + +`graph_transaction` uses a separate request identity map. It does not read outer +process-cache entities or write transaction-local entities into the shared +cache. Reads from an explicit different context do not reuse a request identity +entry bound to the transaction. Touched parent cache keys are invalidated after +both commit and rollback. Collection DDL cache entries are promoted only after +commit; rollback permits subsequent bootstrap. +Sources: `jvspatial/db/transaction.py`, `jvspatial/db/postgres.py`, +`jvspatial/core/context.py`. See [Transactions](docs/md/transactions.md). + ### 4.4 Atomic IO (JSON) `jvspatial/db/_atomic.py` provides crash-safe writes: temp file → `fsync` → `rename` → `fsync(directory)`. Per-file mutex via `PathLockManager` (`_path_locks.py`) serializes concurrent writes to the same record. Bounded LRU prevents lock-table growth. @@ -841,14 +865,26 @@ TTL is per-cache-instance; default `None` (cache until eviction). Invalidation i ### 14.1 MetricsRecorder protocol -`MetricsRecorder` defines `record_db_op`, `record_hook`, `record_cache_event`. Implementations may push to Prometheus, StatsD, or any sink. The library provides: +`MetricsRecorder` defines `record_duration`, `increment_counter` and `record_value`. Implementations may push to Prometheus, StatsD, or any sink. The library provides: -- `NullRecorder` — default; no-op. +- `NullMetricsRecorder` — default; no-op. - OpenTelemetry adapter (`observability/otel.py`) — optional, requires `[otel]` extra. ### 14.2 Structured DB logging -`jvspatial/db/_observable.py` — wraps a `Database` to emit `{backend, op, collection, duration_ms, success, result_count}`. Successful routine timings use DEBUG; failed operations and durations at or above the configurable `slow_query_ms` threshold use WARNING. Log filtering does not disable metrics. Residual scans add execution mode, partial-pushdown status and examined/decoded work; observed PostgreSQL calls add pool wait, and retry helpers add driver attempts/retry count. Iterator active duration is separate from total consumer-inclusive duration. These are driver/candidate metrics, not physical protocol round trips or server rows scanned. +`create_database(observe=True)` wraps any backend and records metrics for each +wrapped operation. Successful operations produce no log record at any logger +level, including calls reaching `slow_query_ms`; those calls increment a slow +metric. Failures emit ERROR with the existing `{backend, op, collection, +duration_ms, success, result_count}` field schema where applicable. Log filtering +does not disable metrics. Successful calls skip logging payload construction +and message formatting. Residual scans add execution mode, partial-pushdown +status and examined/decoded work; observed PostgreSQL calls add pool wait, and +retry helpers add driver attempts/retry count. Iterator active duration is +separate from total consumer-inclusive duration. These are driver/candidate +metrics, not physical protocol round trips or server rows scanned. +Source: `jvspatial/db/_observable.py`. + ### 14.3 GraphContext PerformanceMonitor diff --git a/docs/md/README.md b/docs/md/README.md index c6b7f21..f8a6844 100644 --- a/docs/md/README.md +++ b/docs/md/README.md @@ -9,7 +9,7 @@ This directory holds detailed how-to and reference documentation. For higher-lev - [AGENTS.md](../../AGENTS.md) — agent maintenance guide - [CHANGELOG.md](../../CHANGELOG.md) — release history -The documents below answer *how* to use each subsystem. Every link resolves; entries marked **NEW** were added since the previous index revision. +The documents below answer *how* to use each subsystem. Every link resolves; entries marked were added since the previous index revision. --- @@ -57,11 +57,11 @@ The documents below answer *how* to use each subsystem. Every link resolves; ent | Document | What's in it | |---|---| | [Authentication](authentication.md) | JWT, API key, refresh-token flows. | -| [API Keys](api-keys.md) | **NEW** Key management endpoints and hashing model. | +| [API Keys](api-keys.md) | Key management endpoints and hashing model. | | [Password Migration Guide](password-migration-guide.md) | bcrypt upgrade path. | -| [Security Review](security-review.md) | **NEW** Audit findings and resolved fixes. | -| [Security Operational Notes](security-operational-notes.md) | **NEW** Runtime security guidance. | -| [Production Deployment](production-deployment.md) | **NEW** Hardening checklist for production. | +| [Security Review](security-review.md) | Audit findings and resolved fixes. | +| [Security Operational Notes](security-operational-notes.md) | Runtime security guidance. | +| [Production Deployment](production-deployment.md) | Hardening checklist for production. | ## Database and Storage @@ -70,6 +70,9 @@ The documents below answer *how* to use each subsystem. Every link resolves; ent | [MongoDB Query Interface](mongodb-query-interface.md) | Mongo-style operators across all backends. | | [Custom Database Guide](custom-database-guide.md) | Implementing and registering a new `Database` adapter. | | [DynamoDB Guide](dynamodb-guide.md) | DynamoDB-specific setup and limits. | +| [PostgreSQL Guide](postgres-guide.md) | Connection setup, native transactions, tenant scope, pooling and vector support. | +| [Transactions](transactions.md) | Shared transaction reads, backend support, commit/rollback and cache isolation. | +| [Cursor Pagination](cursor-pagination.md) | Deterministic ID/keyset pages, continuation and iterator semantics. | | [File Storage Architecture](file-storage-architecture.md) | Storage interface, security layer, version model. | | [File Storage Usage](file-storage-usage.md) | Upload, list, download, versioning. | | [Caching](caching.md) | Memory / Redis / layered caches; read-through DB wrapper. | @@ -79,11 +82,11 @@ The documents below answer *how* to use each subsystem. Every link resolves; ent | Document | What's in it | |---|---| -| [Observability](observability.md) | **NEW** Structured DB op logs, `MetricsRecorder`, slow-query threshold, OTEL adapter. | +| [Observability](observability.md) | Failure-only DB logs, per-operation/slow-call metrics and OpenTelemetry adapter. | | [Logging Service](logging-service.md) | Persistent application logging. | | [Custom Log Levels](custom-log-levels.md) | Adding domain levels (`AUDIT`, `SECURITY`, …). | -| [Benchmarks](benchmarks.md) | **NEW** Regression-detection bench suite. | -| [Serverless Mode](serverless-mode.md) | **NEW** Detection precedence, mode-dependent defaults, LWA env. | +| [Benchmarks](benchmarks.md) | Regression-detection bench suite. | +| [Serverless Mode](serverless-mode.md) | Detection precedence, mode-dependent defaults, LWA env. | | [Scheduler](scheduler.md) | In-process scheduler and decorators. | | [Webhook Architecture](webhook-architecture.md) | Webhook routing, signature verification. | | [Webhooks Quickstart](webhooks-quickstart.md) | Common webhook patterns. | @@ -97,7 +100,7 @@ The documents below answer *how* to use each subsystem. Every link resolves; ent | [Design Decisions](design-decisions.md) | Philosophy and tradeoffs. | | [Module Responsibility Matrix](module-responsibility-matrix.md) | Which package owns which concern. | | [Import Patterns](import-patterns.md) | Stable vs internal imports. | -| [Stability](stability.md) | **NEW** Public / internal / experimental tiers + deprecation policy. | +| [Stability](stability.md) | Public / internal / experimental tiers + deprecation policy. | | [Optimization](optimization.md) | Performance tuning: `neighborhood`, walker prefetch, Postgres paths, fast deserialize. | ## Configuration diff --git a/docs/md/api-architecture.md b/docs/md/api-architecture.md index 40b5a4a..730734f 100644 --- a/docs/md/api-architecture.md +++ b/docs/md/api-architecture.md @@ -1,508 +1,110 @@ -# jvspatial API Architecture +# API architecture -The jvspatial API module provides a comprehensive FastAPI integration layer that enables RESTful access to graph-based spatial data operations. The architecture is built around a central [`Server`](../../jvspatial/api/server.py:38) class that orchestrates all components. +jvspatial's [`Server`](../../jvspatial/api/server.py) composes FastAPI app +construction, endpoint registration, middleware, lifecycle and the database +context. Applications use the public `jvspatial.api` exports; implementation +mixins and component paths are maintainer references, not import contracts. -## Core Architecture +## Configuration and server lifecycle -### Server Class +`Server` accepts configuration through +[`ServerConfig`](../../jvspatial/api/config.py) and its grouped settings in +[`config_groups.py`](../../jvspatial/api/config_groups.py). Database, auth, CORS, +security, file storage and scheduler options follow those schemas. Environment +adaptation uses the declared `JVSPATIAL_*` allowlist rather than arbitrary +configuration keys. -The [`Server`](../../jvspatial/api/server.py:38) class is the main orchestrator that manages: +[`server_app_factory.py`](../../jvspatial/api/server_app_factory.py), +[`server_registration.py`](../../jvspatial/api/server_registration.py), +[`server_lifecycle.py`](../../jvspatial/api/server_lifecycle.py) and +[`server_run.py`](../../jvspatial/api/server_run.py) implement the composition. +The [`LifecycleManager`](../../jvspatial/api/services/lifecycle.py) integrates +startup and shutdown with FastAPI lifespan. -- FastAPI application lifecycle via [`LifecycleManager`](../../jvspatial/api/services/lifecycle.py:1) -- Endpoint registration through [`EndpointRegistryService`](../../jvspatial/api/services/endpoint_registry.py:88) -- Middleware configuration via [`MiddlewareManager`](../../jvspatial/api/services/middleware.py:1) -- GraphContext integration for database operations -- Optional file storage and scheduler services +See [Server API](server-api.md) for supported configuration and hooks, and +[Production Deployment](production-deployment.md) for production requirements. -```python -from jvspatial.api import Server - -server = Server( - title="My Spatial API", - description="Graph-based API", - version="1.0.0", - db_type="json", - db_path="./jvdb" -) -``` - -### Configuration - -API configuration is managed through [`ServerConfig`](../../jvspatial/api/config.py:12) which supports: - -- API metadata (title, description, version) -- Server settings (host, port, debug mode) -- CORS configuration -- Database configuration (type, path, connection strings) -- File storage settings (local/S3) -- Proxy settings for temporary file URLs -- Lifecycle hooks (startup/shutdown) - -Environment variables follow the pattern `JVSPATIAL_*` for configuration overrides. - -### Middleware stack order - -FastAPI applies HTTP middleware in **reverse** registration order (last registered runs first on the request path). jvspatial splits wiring across two places on purpose: - -1. **[`MiddlewareManager`](../../jvspatial/api/middleware/manager.py)** ([`configure_all`](../../jvspatial/api/middleware/manager.py)) — security headers, CORS, optional webhook middleware, file-storage routes, and any custom middleware registered via [`Server.middleware`](../../jvspatial/api/server_lifecycle.py). **JWT/API-key auth is intentionally not registered here** so its order relative to rate limiting stays explicit. - -2. **[`ServerConfigurator`](../../jvspatial/api/server_configurator.py)** ([`configure_all`](../../jvspatial/api/server_configurator.py)) — per-request error-logging context (contextvars reset), **rate limiting**, **[`AuthenticationMiddleware`](../../jvspatial/api/components/auth_middleware.py)**, and unified exception handlers. - -**Security intent:** CORS and static security headers run early; rate limits and auth apply together before your route handlers. `/auth/*` routes that use FastAPI `Depends` for bearer tokens bypass the auth middleware when the framework has already resolved credentials (see auth middleware docstring). Adjust exempt paths and backends via `ServerConfig` and the injectable rate-limit backend. - -## Core Components - -### 1. Context Management - -[`context.py`](../../jvspatial/api/context.py:1) provides thread-safe server instance management: +## Endpoint discovery and routing -- [`get_current_server()`](../../jvspatial/api/context.py:20) - Retrieve current server from context -- [`set_current_server()`](../../jvspatial/api/context.py:43) - Set server in current context -- [`ServerContext`](../../jvspatial/api/context.py:66) - Context manager for temporary server switching +The public `@endpoint` decorator supports functions and Walker classes. +It records declarations in the +[deferred registry](../../jvspatial/api/decorators/deferred_registry.py); +Server registration resolves those declarations into routes. Applications +should follow the [Endpoint Registration Guide](endpoint-registration-guide.md) +for construction and module-import ordering. -This enables package-level decorators like `@walker_endpoint` to work without explicit server references. - -### 2. Endpoint System - -The endpoint system consists of two main files: - -**[`endpoint/router.py`](../../jvspatial/api/endpoint/router.py:1)** - Endpoint registration and routing: -- [`EndpointRouter`](../../jvspatial/api/endpoint/router.py:514) - Main router class for walker and function endpoints -- [`EndpointField`](../../jvspatial/api/endpoint/router.py:50) - Enhanced Pydantic field with endpoint configuration -- [`ParameterModelFactory`](../../jvspatial/api/endpoint/router.py:153) - Creates parameter models from Walker classes - -**[`endpoint/response.py`](../../jvspatial/api/endpoint/response.py:1)** - Response utilities: -- [`EndpointResponseHelper`](../../jvspatial/api/endpoint/response.py:65) - Semantic response methods (success, error, created, etc.) -- [`EndpointResponse`](../../jvspatial/api/endpoint/response.py:8) - Response wrapper with status codes and headers - -#### Endpoint Types - -**Walker Endpoints** - Graph traversal via HTTP: ```python -from jvspatial.api import endpoint -from jvspatial.api.decorators import EndpointField +from jvspatial.api import Server, endpoint -@endpoint("/process", methods=["POST"]) -class DataProcessor(Walker): - query: str = EndpointField(description="Search query") - limit: int = EndpointField(default=10, ge=1, le=100) -``` +@endpoint("/ready", methods=["GET"], auth=False) +async def ready() -> dict: + return {"state": "ready"} -**Function Endpoints** - Direct API functions: -```python -from jvspatial.api import endpoint +server = Server(title="Example API", db_type="json", db_path="./data") -@endpoint("/users/count", methods=["GET"]) -async def get_user_count(endpoint) -> Dict[str, Any]: - users = await User.find({}) - count = len(users) - return endpoint.success(data={"count": count}) +if __name__ == "__main__": + server.run() ``` -### 3. Services Layer - -[`services/`](../../jvspatial/api/services/__init__.py:1) provides core service implementations: - -**[`EndpointRegistryService`](../../jvspatial/api/services/endpoint_registry.py:88)** - Central endpoint tracking: -- [`register_walker()`](../../jvspatial/api/services/endpoint_registry.py:132) - Register walker endpoints -- [`register_function()`](../../jvspatial/api/services/endpoint_registry.py:185) - Register function endpoints -- [`list_all()`](../../jvspatial/api/services/endpoint_registry.py:390) - List all registered endpoints -- [`unregister_walker()`](../../jvspatial/api/services/endpoint_registry.py:273) - Remove walker endpoints -- Dynamic endpoint tracking and management - -**[`LifecycleManager`](../../jvspatial/api/services/lifecycle.py:1)** - Application lifecycle: -- Startup and shutdown hook registration -- Lifespan context manager for FastAPI -- Asynchronous hook execution - -**[`MiddlewareManager`](../../jvspatial/api/services/middleware.py:1)** - Middleware configuration: -- CORS middleware setup -- Custom middleware registration -- Middleware application to FastAPI app - -**[`FileStorageService`](../../jvspatial/api/services/file_storage.py:1)** - File operations: -- Upload/download file handling -- Proxy URL generation for temporary access -- Local and S3 storage backend support - -**[`PackageDiscoveryService`](../../jvspatial/api/services/discovery.py:1)** - Automatic endpoint discovery: -- Discovers walker and function endpoints from installed packages -- Pattern-based package scanning -- Dynamic endpoint registration - -### 4. Authentication System - -[`auth/`](../../jvspatial/api/auth/__init__.py:1) provides comprehensive authentication and authorization: - -**Core Entities** ([`auth/entities.py`](../../jvspatial/api/auth/entities.py:1)): -- [`User`](../../jvspatial/api/auth/entities.py:1) - User accounts with roles and permissions -- [`Session`](../../jvspatial/api/auth/entities.py:1) - JWT session management -- [`APIKey`](../../jvspatial/api/auth/entities.py:1) - API key authentication - -**Middleware** ([`auth/middleware.py`](../../jvspatial/api/auth/middleware.py:1)): -- [`AuthenticationMiddleware`](../../jvspatial/api/auth/middleware.py:1) - Request authentication -- [`JWTManager`](../../jvspatial/api/auth/middleware.py:1) - JWT token handling -- [`RateLimiter`](../../jvspatial/api/auth/middleware.py:1) - Rate limiting - -**Decorators** - Use the unified `@endpoint` decorator with parameters: -- `@endpoint(..., auth=True)` - Authenticated endpoint -- `@endpoint(..., auth=True, roles=["admin"])` - Role-based access control -- `@endpoint(..., auth=True, permissions=["read_data"])` - Permission-based access control -- `@endpoint(..., webhook=True)` - Webhook endpoint - -```python -from jvspatial.api import endpoint - -@endpoint("/protected/data", methods=["GET"], auth=True, permissions=["read_data"]) -async def get_protected_data(endpoint): - return endpoint.success(data={"secret": "info"}) - -@endpoint("/admin/process", methods=["POST"], auth=True, roles=["admin"]) -class AdminProcessor(Walker): - pass -``` - -### 5. Webhook System - -[`integrations/webhooks/`](../../jvspatial/api/integrations/webhooks/) provides webhook endpoint support: - -**[`integrations/webhooks/helpers.py`](../../jvspatial/api/integrations/webhooks/helpers.py:1)** - Webhook endpoint helpers: -- [`create_webhook_endpoint_helper()`](../../jvspatial/api/integrations/webhooks/helpers.py:24) - Create webhook endpoint helper -- Webhook payload injection -- Webhook response handling - -**[`integrations/webhooks/middleware.py`](../../jvspatial/api/integrations/webhooks/middleware.py:1)** - Request processing: -- HMAC signature verification -- Payload parsing (JSON/XML/binary) -- Idempotency key handling -- Path-based authentication token extraction - -**[`integrations/webhooks/models.py`](../../jvspatial/api/integrations/webhooks/models.py:1)** - Webhook data models: -- Webhook request tracking -- Idempotency management -- Webhook configuration storage - -**[`integrations/webhooks/utils.py`](../../jvspatial/api/integrations/webhooks/utils.py:1)** - Utility functions: -- HMAC signature generation and verification -- Content type detection -- Payload parsing helpers - -### 6. Scheduler (Optional) - -[`scheduler/`](../../jvspatial/api/scheduler/__init__.py:1) provides optional task scheduling: - -**Installation**: `pip install jvspatial[scheduler]` - -**Core Components**: -- [`SchedulerService`](../../jvspatial/api/scheduler/scheduler.py:1) - Task scheduling engine -- `@on_schedule` - Decorator for scheduled functions -- [`SchedulerMiddleware`](../../jvspatial/api/scheduler/middleware.py:1) - FastAPI lifecycle integration - -```python -from jvspatial.api.scheduler import on_schedule - -@on_schedule("every 30 minutes") -async def cleanup_old_data(): - # Cleanup logic - pass -``` - -### 7. Exception Handling - -**[`exceptions.py`](../../jvspatial/api/exceptions.py:1)** - Standardized exception hierarchy: -- [`JVSpatialAPIException`](../../jvspatial/api/exceptions.py:11) - Base exception class -- [`AuthenticationError`](../../jvspatial/api/exceptions.py:61) - Authentication failures -- [`AuthorizationError`](../../jvspatial/api/exceptions.py:102) - Permission denials -- [`ResourceError`](../../jvspatial/api/exceptions.py:143) - Resource operations -- [`ValidationError`](../../jvspatial/api/exceptions.py:180) - Input validation -- [`StorageError`](../../jvspatial/api/exceptions.py:214) - File storage errors -- [`WebhookError`](../../jvspatial/api/exceptions.py:259) - Webhook processing - -**[`error_handler.py`](../../jvspatial/api/components/error_handler.py:1)** - Centralized error handling: -- [`APIErrorHandler`](../../jvspatial/api/components/error_handler.py:125) - Unified error handling class -- [`handle_exception()`](../../jvspatial/api/components/error_handler.py:137) - Centralized exception handler -- Handles all exception types: `JVSpatialAPIException`, `HTTPException`, `ValidationError`, `httpx` exceptions, and unexpected errors - -All exceptions provide consistent JSON responses with error codes, messages, timestamps, paths, and optional details. Error logging is centralized to prevent duplicates, with stack traces for 5xx errors and clean logs for 4xx errors. +The explicitly public readiness route does not demonstrate authenticated +production deployment. Set authentication and route permissions for your +application's actual requirements. -### 8. Constants and Protocols +The current routing implementation lives in [`endpoints/`](../../jvspatial/api/endpoints/): -**[`constants.py`](../../jvspatial/api/constants.py:1)** - Centralized constants: -- [`APIRoutes`](../../jvspatial/api/constants.py:10) - Route path constants -- [`HTTPMethods`](../../jvspatial/api/constants.py:28) - HTTP method constants -- [`Collections`](../../jvspatial/api/constants.py:40) - Database collection names -- [`LogIcons`](../../jvspatial/api/constants.py:51) - Emoji icons for logging -- [`ErrorMessages`](../../jvspatial/api/constants.py:77) - Standard error messages -- [`Defaults`](../../jvspatial/api/constants.py:112) - Default configuration values +- [`registry.py`](../../jvspatial/api/endpoints/registry.py) tracks endpoint declarations. +- [`factory.py`](../../jvspatial/api/endpoints/factory.py) builds routes and validation. +- [`router.py`](../../jvspatial/api/endpoints/router.py) contains the router surface. +- [`response.py`](../../jvspatial/api/endpoints/response.py) provides `ResponseHelper` + and `format_response`, exported through `jvspatial.api`. -**[`protocols.py`](../../jvspatial/api/protocols.py:1)** - Type protocols for interfaces: -- [`FileStorageProvider`](../../jvspatial/api/protocols.py:12) - File storage interface -- [`ProxyManager`](../../jvspatial/api/protocols.py:84) - URL proxy interface -- [`EndpointRegistry`](../../jvspatial/api/protocols.py:145) - Endpoint registry interface -- [`LifecycleManager`](../../jvspatial/api/protocols.py:202) - Lifecycle management interface -- [`MiddlewareManager`](../../jvspatial/api/protocols.py:256) - Middleware configuration interface -- [`PackageDiscovery`](../../jvspatial/api/protocols.py:309) - Package discovery interface -- [`HealthChecker`](../../jvspatial/api/protocols.py:324) - Health check interface +Use [Decorator Reference](decorator-reference.md) for field and endpoint +parameters and [REST API](rest-api.md) for request/response examples. -## Request/Response Flow +## Authentication and middleware -### Walker Endpoint Request Flow +Authentication implementation lives in +[`auth/service.py`](../../jvspatial/api/auth/service.py), +[`auth/models.py`](../../jvspatial/api/auth/models.py) and +[`auth/rbac.py`](../../jvspatial/api/auth/rbac.py). User, API-key, session and +refresh-token state belongs to the prime database. Access checks do not move +to a secondary logging database. -1. **Request arrives** → FastAPI routing -2. **Middleware processing** → Authentication, CORS, etc. -3. **Parameter validation** → Pydantic model created from `EndpointField` definitions -4. **Walker instantiation** → Parameters mapped to Walker fields -5. **Endpoint helper injection** → `walker.endpoint` created for semantic responses -6. **Graph traversal** → `walker.spawn(start_node)` executes -7. **Response collection** → Results gathered via `walker.get_report()` -8. **Response formatting** → JSON response with appropriate status code +[`ServerConfigurator`](../../jvspatial/api/server_configurator.py) installs +request context, rate limiting, authentication and exception handlers. +[`MiddlewareManager`](../../jvspatial/api/middleware/manager.py) configures +security headers, CORS and optional/custom middleware. FastAPI executes HTTP +middleware in reverse registration order. -### Function Endpoint Request Flow +The [authentication middleware](../../jvspatial/api/components/auth_middleware.py) +enforces declared auth, roles and permissions. Raw FastAPI routes may delegate +authentication only to an enforcing `SecurityBase` dependency; a function name +or an arbitrary `Depends` declaration does not prove authentication. Auth +entrypoint limits remain independent of the global rate limiter. -1. **Request arrives** → FastAPI routing -2. **Middleware processing** → Authentication, CORS, etc. -3. **Parameter injection** → Function parameters bound -4. **Endpoint helper injection** → `endpoint` parameter injected -5. **Function execution** → Business logic runs -6. **Response formatting** → `endpoint.success()`, `endpoint.error()`, etc. +See [Authentication](authentication.md), [API Keys](api-keys.md), +[Rate Limiting](rate-limiting.md) and [Security Operational Notes](security-operational-notes.md). -### Webhook Request Flow +## Context and transactions -1. **Request arrives** → Webhook-specific route -2. **Webhook middleware** → HMAC verification, payload parsing, idempotency check -3. **Payload injection** → Webhook data attached to `request.state` -4. **Handler execution** → Function or Walker processes payload -5. **200 OK response** → Always return success (webhook best practice) - -## Key Design Patterns - -### 1. Context-Based Server Access - -Server instances are managed via context variables, enabling: -- Thread-safe server access -- Package-level decorators without explicit server references -- Multiple server instances in testing scenarios - -### 2. Service-Oriented Architecture - -Each major feature is isolated in a service: -- **Separation of concerns** - Services focus on single responsibilities -- **Testability** - Services can be tested independently -- **Extensibility** - New services can be added without modifying core - -### 3. Protocol-Based Interfaces - -Type protocols define clear contracts: -- **Flexibility** - Multiple implementations possible -- **Type safety** - Static type checking with mypy -- **Documentation** - Protocols serve as interface documentation - -### 4. Decorator-Based Registration - -Endpoints are registered via decorators: -- **Clean syntax** - Natural Python idioms -- **Metadata preservation** - Function signatures maintained -- **Dynamic discovery** - Endpoints can be discovered from packages - -### 5. Semantic Response Helpers - -Response helpers provide semantic methods: -- **Readability** - `endpoint.success()` vs manual status codes -- **Consistency** - Standardized response formats -- **Type safety** - Proper return type hints - -## Integration Points - -### Database Integration - -The Server class integrates with jvspatial's GraphContext: -- Automatic database initialization based on `ServerConfig` -- Support for JSON and MongoDB backends -- Optional explicit GraphContext configuration - -```python -server = Server( - title="My API", - db_type="mongodb", - db_connection_string="mongodb://localhost:27017", - db_database_name="mydb" -) - -# Or set explicitly -ctx = GraphContext(database=my_database) -server.set_graph_context(ctx) -``` - -### File Storage Integration - -File storage is optional and configured via `ServerConfig`: -- Local filesystem storage with configurable root directory -- S3-compatible storage with AWS credentials -- Proxy URL generation for temporary file access -- Automatic cleanup and expiration handling - -### Scheduler Integration - -Scheduler is optional and requires separate installation: -- Integrates with FastAPI lifecycle (startup/shutdown) -- Supports cron-like scheduling syntax -- Thread-based execution separate from request handling -- Automatic task discovery from decorated functions - -## API Endpoint Structure - -Standard endpoint patterns follow RESTful conventions: - -``` -/ - Root endpoint (API info) -/health - Health check endpoint -/api/ - API prefix for user endpoints - /api/walkers/* - Walker-based graph operations - /api/functions/* - Function-based operations -/api/files/ - File HTTP routes (if enabled; OpenAPI tag Files) - /api/files/upload - File upload - /api/files/{path} - File serve (GET) and delete (DELETE) - /api/files/proxy - Proxy URL admin (when proxy enabled) -/p/{code} - Proxy URL short links (if enabled) -``` - -## Best Practices - -### 1. Use Semantic Response Methods - -```python -# ✅ Good - Semantic and clear -@endpoint("/users/{user_id}") -async def get_user(user_id: str, endpoint): - user = await User.get(user_id) - if not user: - return endpoint.not_found(message="User not found") - return endpoint.success(data={"user": await user.export()}) - -# ❌ Avoid - Manual status codes -async def get_user(user_id: str): - user = await User.get(user_id) - if not user: - return JSONResponse(status_code=404, content={"error": "Not found"}) - return {"user": await user.export()} -``` - -### 2. Use EndpointField for Parameter Configuration - -```python -# ✅ Good - Explicit field configuration -@walker_endpoint("/search") -class SearchWalker(Walker): - query: str = EndpointField( - description="Search query", - examples=["python", "fastapi"], - min_length=1, - max_length=100 - ) - limit: int = EndpointField( - default=10, - ge=1, - le=100, - description="Maximum results" - ) -``` - -### 3. Leverage Service Isolation - -```python -# ✅ Good - Use service methods -server = Server(title="My API") -registry = server._endpoint_registry - -# Check if endpoint exists before registering -if not registry.has_walker(MyWalker): - registry.register_walker(MyWalker, "/process", ["POST"]) - -# ❌ Avoid - Direct manipulation of internal state -server._walker_registry[MyWalker] = {...} -``` - -### 4. Handle Errors Consistently - -```python -# ✅ Good - Use standardized exceptions -from jvspatial.api.exceptions import ResourceNotFoundError - -@endpoint("/items/{item_id}") -async def get_item(item_id: str, endpoint): - item = await Item.get(item_id) - if not item: - raise ResourceNotFoundError( - message="Item not found", - details={"item_id": item_id} - ) - return endpoint.success(data=await item.export()) -``` - -### 5. Follow Webhook Best Practices - -```python -# ✅ Good - Always return 200 for webhooks -@endpoint("/webhook/github", webhook=True) -async def github_webhook(payload: dict, endpoint): - try: - # Process webhook - await process_github_event(payload) - return endpoint.response(content={"status": "processed"}) - except Exception as e: - # Log error but still return 200 - logger.error(f"Webhook error: {e}") - return endpoint.response(content={"status": "received", "error": "logged"}) -``` - -## Architecture Diagram - -``` -┌─────────────────────────────────────────────────────────────┐ -│ Server │ -│ ┌───────────┐ ┌──────────────┐ ┌───────────────────┐ │ -│ │ Config │ │ GraphContext│ │ LifecycleManager │ │ -│ └───────────┘ └──────────────┘ └───────────────────┘ │ -└─────────────────────────────────────────────────────────────┘ - │ - ┌──────────────────┼──────────────────┐ - │ │ │ -┌───────▼───────┐ ┌──────▼──────┐ ┌───────▼────────┐ -│ Endpoint │ │ Services │ │ Middleware │ -│ System │ │ Layer │ │ Manager │ -├───────────────┤ ├─────────────┤ ├────────────────┤ -│ - Router │ │ - Registry │ │ - Auth │ -│ - Response │ │ - Discovery │ │ - CORS │ -│ - Fields │ │ - File │ │ - Custom │ -└───────────────┘ │ - Lifecycle │ └────────────────┘ - └─────────────┘ - │ │ -┌───────▼───────┐ ┌──────▼──────────────────┐ -│ Auth System │ │ Optional Components │ -├───────────────┤ ├─────────────────────────┤ -│ - Users │ │ - Scheduler │ -│ - Sessions │ │ - Webhooks │ -│ - API Keys │ │ - File Storage │ -│ - Decorators │ └─────────────────────────┘ -└───────────────┘ -``` +[`ServerContext`](../../jvspatial/api/context.py), `get_current_server` and +`set_current_server` bind the active server. `GraphContext` binds entity +persistence to its database, cache and monitor. Applications must keep +workspace/access policy separate from context selection: a context is a +persistence boundary, not an authorization grant. -## Summary +Multi-write atomic entity/edge work uses [Transactions](transactions.md), +with native backend qualification and transaction-local cache isolation. +See [Context Management](context-management-guide.md) for scoped context usage. -The jvspatial API architecture provides a complete FastAPI integration with: +## Optional integrations -- **Flexible endpoint system** supporting both walkers and functions -- **Comprehensive authentication** with JWT, API keys, and RBAC -- **Service-oriented design** for maintainability and extensibility -- **Optional components** (scheduler, webhooks, file storage) -- **Type-safe interfaces** via protocols and Pydantic models -- **Consistent error handling** through standardized exceptions -- **Context-based management** for thread-safe server access -- **Semantic response helpers** for clean, readable code +- [Webhooks](webhook-architecture.md): + [`integrations/webhooks/`](../../jvspatial/api/integrations/webhooks/). +- Scheduling: [`integrations/scheduler/`](../../jvspatial/api/integrations/scheduler/) + and the [runnable scheduler examples](../../examples/scheduler/README.md). +- File storage: [File Storage Usage](file-storage-usage.md) and + [File Storage Architecture](file-storage-architecture.md). -The architecture emphasizes: -- Clean separation of concerns -- Type safety and protocol-based interfaces -- Extensibility through services and middleware -- Developer experience via semantic APIs -- Production readiness with authentication and error handling \ No newline at end of file +Internal integration classes and directory names can evolve. Use public Server, +decorator and configuration contracts when integrating an application. diff --git a/docs/md/authentication.md b/docs/md/authentication.md index a549f8d..07946f6 100644 --- a/docs/md/authentication.md +++ b/docs/md/authentication.md @@ -950,11 +950,11 @@ server = Server( ## See Also -- [Authentication Examples](../examples/api/authenticated_endpoints_example.py) - Complete working examples +- [Authentication Examples](../../examples/api/authenticated_endpoints_example.py) - Complete working examples - [REST API Integration](rest-api.md) - API endpoint patterns - [Server API Documentation](server-api.md) - Server configuration - [Security Quickstart](auth-quickstart.md) - Get started quickly --- -**[← Back to README](../../README.md)** | **[Authentication Examples →](../examples/api/authenticated_endpoints_example.py)** +**[← Back to README](../../README.md)** | **[Authentication Examples →](../../examples/api/authenticated_endpoints_example.py)** diff --git a/docs/md/cursor-pagination.md b/docs/md/cursor-pagination.md index d6d7a90..5c9fd0c 100644 --- a/docs/md/cursor-pagination.md +++ b/docs/md/cursor-pagination.md @@ -115,9 +115,9 @@ async with transaction_context(db) as txn: ... ``` -(Note: `find_iter` is not yet wired through the transaction handle for -all backends — use it from outside the transaction when consistency -isn't critical, or fall back to `txn.find` for snapshotted reads.) +`Transaction.find_iter` pages through the transaction handle, so each read +uses the same transaction-visible state. Adapters may override it with a +native cursor implementation. ## Sort + filter composition diff --git a/docs/md/design-decisions.md b/docs/md/design-decisions.md index ae51375..f8ba1a3 100644 --- a/docs/md/design-decisions.md +++ b/docs/md/design-decisions.md @@ -271,7 +271,7 @@ async def analyze(self, here: City): ## See Also -- [Core Concepts](core-concepts.md) - [Entity Reference](entity-reference.md) -- [Walker Patterns](walker-patterns.md) +- [Entity Reference](entity-reference.md) +- [Graph Traversal](graph-traversal.md) - [API Integration](rest-api.md) \ No newline at end of file diff --git a/docs/md/error-handling.md b/docs/md/error-handling.md index 069ec17..1ce6fff 100644 --- a/docs/md/error-handling.md +++ b/docs/md/error-handling.md @@ -258,6 +258,6 @@ The system handles exception chaining properly: ## See Also -- [Database Configuration Guide](configuration.md) -- [Walker Patterns](walker-patterns.md) -- [Advanced Error Handling](advanced-error-handling.md) \ No newline at end of file +- [Database Configuration Guide](server-api.md) +- [Graph Traversal](graph-traversal.md) +- [Error-handling examples](../../examples/error_handling/README.md) \ No newline at end of file diff --git a/docs/md/examples.md b/docs/md/examples.md index 1a792b9..9f89251 100644 --- a/docs/md/examples.md +++ b/docs/md/examples.md @@ -1,757 +1,38 @@ -# jvspatial Examples +# Examples -This document showcases key examples that demonstrate the range of capabilities and features of the jvspatial library - a powerful object-spatial ORM for building connected graph applications with spatial awareness. +The canonical runnable index is [examples/README.md](../../examples/README.md). +Examples are grouped by feature rather than located directly in the repository +root. Each group has setup and usage notes. -## 🎯 **Standard Implementation Examples** (Recommended Starting Point) +| Area | Index | +| --- | --- | +| Entities, graph traversal and visualization | [Core examples](../../examples/core/README.md) | +| Authenticated and public HTTP endpoints | [API examples](../../examples/api/README.md) | +| Backends, queries and database contexts | [Database examples](../../examples/database/README.md) | +| File storage | [Storage examples](../../examples/storage/README.md) | +| Scheduled work | [Scheduler examples](../../examples/scheduler/README.md) | +| Failure handling | [Error-handling examples](../../examples/error_handling/README.md) | +| Logging | [Logging examples](../../examples/logging/README.md) | -These examples serve as the **standard reference implementation** for building custom jvspatial APIs: - -### 1. **Authenticated API Example** (Complete CRUD with Authentication) -**File**: [`examples/api/authenticated_endpoints_example.py`](../../examples/api/authenticated_endpoints_example.py) - -**What it demonstrates**: -- ✅ Complete CRUD operations (Create, Read, Update, Delete) -- ✅ JWT-based authentication and authorization -- ✅ Permission and role-based access control -- ✅ Entity-centric database operations (`UserNode.get()`, `ProductNode.create()`, etc.) -- ✅ Pagination with `ObjectPager` -- ✅ Response schemas with examples -- ✅ Walker-based analytics endpoints -- ✅ Automatic authentication endpoint registration (login, register, logout) - -**Key Features**: -```python -# Server with authentication enabled -server = Server( - title="Authenticated CRUD API Example", - auth_enabled=True, - auth_enabled=True, - db_type="json", - db_path="./jvdb" -) - -# Protected endpoint with permissions -@endpoint("/users", methods=["GET"], auth=True, permissions=["read_users"]) -async def list_users(page: int = 1, per_page: int = 10): - pager = ObjectPager(UserNode, page_size=per_page) - users = await pager.get_page(page=page) - import asyncio - users_list = await asyncio.gather(*[user.export() for user in users]) - return {"users": users_list, ...} - -# Entity-centric CRUD operations -user = await UserNode.create(name="John", email="john@example.com") -await user.save() -await user.delete() -``` - -**Use this example when**: You need a full-featured API with authentication, CRUD operations, and role-based access control. - -### 2. **Unauthenticated API Example** (Public Read-Only API) -**File**: [`examples/api/unauthenticated_endpoints_example.py`](../../examples/api/unauthenticated_endpoints_example.py) - -**What it demonstrates**: -- ✅ Public endpoints (no authentication required) -- ✅ Read-only operations (GET endpoints) -- ✅ Listing with pagination and filtering -- ✅ Entity-centric retrieval operations -- ✅ Response schemas with examples -- ✅ **No authentication endpoints** (login/register/logout are NOT registered) - -**Key Features**: -```python -# Server without authentication -server = Server( - title="Public API Example", - auth_enabled=False, # No auth endpoints will be registered - db_type="json", - db_path="./jvdb_public" -) - -# Public endpoint with pagination -@endpoint("/articles", methods=["GET"]) -async def list_articles(page: int = 1, per_page: int = 10, category: Optional[str] = None): - pager = ObjectPager(ArticleNode, page_size=per_page) - articles = await pager.get_page(page=page, additional_filters={"category": category} if category else {}) - import asyncio - articles_list = await asyncio.gather(*[article.export() for article in articles]) - return {"articles": articles_list, ...} -``` - -**Use this example when**: You need a public API for reading data, content delivery, or read-only services. - ---- - -## Table of Contents - -**Standard Examples** (Start Here): -1. [Authenticated API Example](#1-authenticated-api-example-complete-crud-with-authentication) ⭐ -2. [Unauthenticated API Example](#2-unauthenticated-api-example-public-read-only-api) ⭐ - -**Additional Examples**: -3. [Graph Visualization](#graph-visualization) - Export graphs in DOT/Mermaid formats -4. [Core ORM Demo](#core-orm-demo) - Basic object-spatial ORM concepts -5. [Travel Graph](#travel-graph) - Spatial operations and walker patterns -6. [Agent Graph](#agent-graph) - Hierarchical systems with API endpoints -7. [Authentication Demo](#authentication-demo) - JWT tokens, API keys, and RBAC -8. [Dynamic Server](#dynamic-server) - Runtime endpoint registration -9. [GraphContext Demo](#graphcontext-demo) - Database dependency injection -10. [Semantic Filtering](#semantic-filtering) - Advanced query capabilities - ---- - -## Graph Visualization - -**File**: [`examples/core/graph_visualization_example.py`](../../examples/core/graph_visualization_example.py) - -**What it demonstrates**: -- ✅ Creating interconnected graph structures -- ✅ Exporting graphs in DOT (Graphviz) format -- ✅ Exporting graphs in Mermaid format -- ✅ Custom filtering and styling -- ✅ Multiple layout options -- ✅ File output and return value patterns - -**Key Features**: -```python -from jvspatial.core.graph import generate_graph_dot, generate_graph_mermaid - -# Generate DOT format (returns value, optionally saves to file) -dot_graph = await generate_graph_dot( - context, - graph_name="company_graph", - rankdir="LR", - node_shape="box", - include_attributes=True, - output_file="graph.dot" # Optional: saves to file -) - -# Generate Mermaid format -mermaid_graph = await generate_graph_mermaid( - context, - graph_type="flowchart", - direction="LR", - include_attributes=True, - output_file="graph.mermaid" # Optional: saves to file -) - -# Using GraphContext convenience method -graph = await context.export_graph( - format="dot", - output_file="graph.dot", - rankdir="LR" -) -``` - -**Use this example when**: You need to visualize your graph structure for documentation, debugging, or analysis. - -**See also**: [Graph Visualization Guide](../graph-visualization.md) - Complete documentation - - -# Usage -walker = TravelWalker() -await walker.spawn(start=city) -report = walker.get_report() # List of reported items -visited = walker.cities_visited # Internal state -``` - ---- - -## Travel Graph - -**File**: [`examples/travel_graph.py`](examples/travel_graph.py) - -**What it demonstrates**: -- Spatial calculations and geographic data -- Complex walker patterns with state management -- MongoDB-style spatial queries -- Edge-typed graph traversal - -### Key Features Shown - -**Spatial Queries**: -```python path=examples/travel_graph.py start=69 -bounded_cities = await City.find({ - "$and": [ - {"context.latitude": {"$gte": min_lat, "$lte": max_lat}}, - {"context.longitude": {"$gte": min_lon, "$lte": max_lon}}, - ] -}) -``` - -**Haversine Distance Calculation**: -```python path=examples/travel_graph.py start=31 -def calculate_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float: - """Calculate distance between two coordinates using Haversine formula.""" - earth_radius = 6371 # Earth's radius in kilometers - # ... mathematical calculation -``` - -**Edge-Type Filtering**: -```python path=examples/travel_graph.py start=140 -highway_neighbors = await here.nodes(direction="out", edge=[Highway]) -``` - -**Stateful Walker with Cargo Management**: -```python path=examples/travel_graph.py start=162 -class FreightTrain(Walker): - max_cargo_capacity: int = 5000 # tons - current_cargo_weight: int = 0 - - def __init__(self, **kwargs): - super().__init__(**kwargs) - self.route_cities = [] # Track visited cities - self.cargo_list = [] # Track cargo loaded - - @on_visit(City) - async def load_cargo(self, here: City): - self.route_cities.append(here.name) - # Complex cargo loading logic - self.report({"cargo_loaded": {...}}) # Report each cargo operation -``` - ---- - -## Agent Graph - -**File**: [`examples/agent_graph.py`](examples/agent_graph.py) - -**What it demonstrates**: -- Hierarchical agent systems -- API endpoint integration with walker patterns -- Entity-centric CRUD operations -- Type annotations and validation - -### Key Features Shown - -**Hierarchical Structure**: Root → App → Agents → MyAgent → Actions - -**API Endpoint Integration with Semantic Responses**: -```python path=examples/agent_graph.py start=116 -from jvspatial.api.decorators import EndpointField - -@endpoint("/api/agents/interact", methods=["POST"]) -class InteractWalker(Walker): - target_agent_name: str = EndpointField( - default="", - description="Name of specific agent to target (optional)", - examples=["ProductionAgent", "TestAgent"], - ) - - @on_visit(Node) - async def process(self, here: Node): - # Use endpoint semantic responses for API endpoints - return self.endpoint.success( - data={"result": "processed"}, - message="Processing complete" - ) -``` - -**Entity-Centric CRUD**: -```python path=examples/agent_graph.py start=147 -app_nodes = await App.find({"context.status": "active"}) -``` - -**Complex Query Building**: -```python path=examples/agent_graph.py start=193 -query_filters: Dict[str, Any] = {"context.published": True} -if not self.include_inactive: - query_filters["context.status"] = "active" -``` - -**Spatial Properties in Agents**: -```python path=examples/agent_graph.py start=53 -class MyAgent(Node): - name: str = "" - published: bool = True - latitude: float = 0.0 - longitude: float = 0.0 - capabilities: List[str] = Field(default_factory=list) -``` - ---- - -## Authentication Demo - -**File**: [`examples/auth_demo.py`](examples/auth_demo.py) - -**What it demonstrates**: -- Complete JWT token authentication system -- API key authentication for services -- Role-based access control (RBAC) -- Spatial region and node type permissions -- Multi-level endpoint protection -- User management and session handling -- Rate limiting and security middleware - -### Key Features Shown - -**Authentication Configuration**: -```python path=examples/auth_demo.py start=81 -configure_auth( - # JWT Configuration - jwt_secret_key=jwt_secret, - jwt_algorithm="HS256", - jwt_expiration_hours=24, - jwt_refresh_expiration_days=7, # Shorter for demo - - # API Key Configuration - api_key_header="X-API-Key", - api_key_query_param="api_key", - - # Rate Limiting - rate_limit_enabled=True, - default_rate_limit_per_hour=100, # Lower for demo - - # Security (relaxed for demo) - require_https=False, - session_cookie_secure=False, - session_cookie_httponly=True -) -``` - -**User Creation with Spatial Permissions**: -```python path=examples/auth_demo.py start=175 -# 3. Regional Analyst with Spatial Restrictions -regional_user = await User.create( - email="regional_user@example.com", - password_hash=User.hash_password("regional123"), - is_admin=False, - is_active=True, - roles=["analyst", "regional_viewer"], - permissions=["read_spatial_data", "analyze_data", "export_reports"], - allowed_regions=["north_america"], # Only North America - allowed_node_types=["City", "Highway"], # No POIs - max_traversal_depth=15, - rate_limit_per_hour=2000 -) -``` - -**Multi-Level Endpoint Protection**: -```python path=examples/auth_demo.py start=366 -@endpoint("/public/info", methods=["GET"]) -async def public_info(): - """Public endpoint - no authentication required.""" - return { - "message": "This is public information", - "service": "jvspatial Authentication Demo", - "version": "1.0.0", - "authentication": "Not required for this endpoint", - "timestamp": datetime.now().isoformat() - } - -@endpoint("/protected/data", auth=True, methods=["GET"]) -async def protected_data(request: Request): - """Protected endpoint requiring authentication.""" - current_user = get_current_user(request) - - return { - "message": "This is protected data", - "authentication": "Required - JWT token", - "user": { - "email": current_user.email, - "roles": current_user.roles, - "permissions": current_user.permissions - }, - "timestamp": datetime.now().isoformat() - } - -@endpoint("/admin/users", auth=True, roles=["admin"], methods=["GET"]) -async def admin_list_users(request: Request): - """Admin endpoint to list all users.""" - current_user = get_current_user(request) - - all_users = await User.all() - # Efficient counting - total_users = await User.count() # Much faster than len(all_users) - return { - "message": "User management - Admin access only", - "authentication": "JWT token + admin role required", - "admin_user": current_user.email, - "total_users": total_users, - "users": await asyncio.gather(*[user.export() for user in all_users]) - } -``` - -**Spatial Access Control in Walker Endpoints**: -```python path=examples/auth_demo.py start=483 -@endpoint( - "/protected/spatial-query", - auth=True, - methods=["POST"], - permissions=["read_spatial_data"] -) -class ProtectedSpatialQuery(Walker): - """Protected spatial query walker with region-based access control.""" - - region: str = EndpointField( - description="Target region to query", - examples=["north_america", "europe", "asia"] - ) - - @on_visit(Node) - async def spatial_query(self, here: Node): - current_user = get_current_user(self.request) - - # Check if user can access this region - if not current_user.can_access_region(self.region): - return self.endpoint.forbidden( - message="Access denied to region", - details={ - "region": self.region, - "user_allowed_regions": current_user.allowed_regions - } - ) -``` - -**API Key Authentication**: -```python path=examples/auth_demo.py start=705 -@endpoint("/api/export/cities", methods=["GET"]) -async def api_key_export_cities(request: Request): - """Endpoint accessible via API key (demonstrates service authentication).""" - - # This endpoint will be accessible via API key due to middleware - # Check if authenticated via API key - api_key_user = getattr(request.state, 'api_key_user', None) - jwt_user = getattr(request.state, 'current_user', None) - - auth_method = "Unknown" - user_info = {} - - if api_key_user: - auth_method = "API Key" - user_info = { - "api_key_name": api_key_user.get("name", "Unknown"), - "key_id": api_key_user.get("key_id", "Unknown") - } - elif jwt_user: - auth_method = "JWT Token" - user_info = { - "username": jwt_user.username, - "roles": jwt_user.roles - } -``` - -**Demo Accounts and Usage**: -```python path=examples/auth_demo.py start=334 -"demo_accounts": { - "admin": { - "username": "demo_admin", - "password": "admin123", - "permissions": "Full admin access" - }, - "user": { - "username": "demo_user", - "password": "user123", - "permissions": "Read access to North America and Europe" - }, - "regional": { - "username": "regional_user", - "password": "regional123", - "permissions": "Analyst access limited to North America" - } -} -``` - -### Running the Authentication Demo - -```bash -cd examples/ -python auth_demo.py -``` - -Then visit: -- **Dashboard**: http://localhost:8000/auth-demo - Demo overview and instructions -- **API Docs**: http://localhost:8000/docs - Interactive API documentation -- **Login**: POST /auth/login - Authenticate to get JWT tokens -- **Public**: GET /public/* - No authentication required -- **Protected**: GET /protected/* - JWT token required -- **Admin**: GET /admin/* - Admin role required - -### Authentication Flow Examples - -**1. User Registration:** -```bash -curl -X POST "http://localhost:8000/auth/register" \ - -H "Content-Type: application/json" \ - -d '{ - "username": "newuser", - "email": "user@example.com", - "password": "password123", - "confirm_password": "password123" - }' -``` - -**2. User Login:** -```bash -curl -X POST "http://localhost:8000/auth/login" \ - -H "Content-Type: application/json" \ - -d '{ - "username": "demo_user", - "password": "user123" - }' -``` - -**3. Access Protected Endpoint:** -```bash -# Use token from login response -TOKEN="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." - -curl -H "Authorization: Bearer $TOKEN" \ - "http://localhost:8000/protected/data" -``` - -**4. API Key Usage:** -```bash -curl -H "X-API-Key: demo-export-key-12345" \ - "http://localhost:8000/api/export/cities" -``` - -**5. Spatial Query with Permissions:** -```bash -curl -X POST "http://localhost:8000/protected/spatial-query" \ - -H "Authorization: Bearer $TOKEN" \ - -H "Content-Type: application/json" \ - -d '{ - "region": "north_america", - "node_type": "City" - }' -``` - ---- - -## Dynamic Server - -**File**: [`examples/dynamic_server_demo.py`](examples/dynamic_server_demo.py) - -**What it demonstrates**: -- Runtime endpoint registration -- Package discovery patterns -- Shared server instances -- Startup hooks and initialization - -### Key Features Shown - -**Server Creation with Enhanced Config**: -```python path=examples/dynamic_server_demo.py start=59 -server = create_server( - title="Dynamic Task Management API", - description="Advanced task management with dynamic endpoint registration", - version="2.0.0", - debug=True, - db_type="json", - db_path="jvdb/dynamic_demo", -) -``` - -**Package Discovery**: -```python path=examples/dynamic_server_demo.py start=75 -server.enable_package_discovery( - enabled=True, patterns=["*_tasks", "*_workflows", "task_*", "demo_*"] -) -``` - -**Startup Hooks**: -```python path=examples/dynamic_server_demo.py start=83 -@server.on_startup -async def initialize_sample_tasks(): - """Create sample data on startup.""" - tasks = [ - await Task.create( - title="System Analysis", - description="Analyze current system architecture", - priority="high", - ), - # ... more tasks - ] -``` - -**Dynamic Walker Registration**: -```python path=examples/dynamic_server_demo.py start=140 -@endpoint("/tasks/create") -class CreateTask(Walker): - title: str = EndpointField( - description="Task title", - examples=["Fix login bug", "Update documentation"], - min_length=3, - max_length=200, - ) -``` - ---- - -## GraphContext Demo - -**File**: [`examples/graphcontext_demo.py`](examples/graphcontext_demo.py) - -**What it demonstrates**: -- Database dependency injection -- Multiple database contexts -- Testing patterns with isolation -- Clean API design - -### Key Features Shown - -**Standard API**: -```python path=examples/graphcontext_demo.py start=65 -# Standard entity creation syntax -chicago = await City.create( - name="Chicago", population=2700000, latitude=41.88, longitude=-87.63 -) -``` - -**Explicit GraphContext**: -```python path=examples/graphcontext_demo.py start=102 -ctx = GraphContext(database=custom_db) - -seattle = await ctx.create_node( - City, name="Seattle", population=750000, latitude=47.61, longitude=-122.33 -) -``` - -**Multiple Database Contexts**: -```python path=examples/graphcontext_demo.py start=141 -# Main database for application data -main_ctx = GraphContext(database=main_db) - -# Analytics database for logging/metrics -analytics_ctx = GraphContext(database=analytics_db) -``` - -**Multi-Database Support with Prime Database**: -For managing multiple databases with a prime database for core persistence, see [`examples/database/multi_database_example.py`](../../examples/database/multi_database_example.py): -```python -from jvspatial.db import create_database, get_database_manager, switch_database, unregister_database - -# Prime database is automatically created for auth/sessions -manager = get_database_manager() -prime_db = manager.get_prime_database() - -# Create and register additional databases -app_db = create_database("json", base_path="./app_data", register=True, name="app") - -# Switch between databases -switch_database("app") # For application data -switch_database("prime") # Back to prime - -# Remove a database -unregister_database("app") -``` - -**Testing Pattern**: -```python path=examples/graphcontext_demo.py start=194 -# Create isolated test database -test_db = create_database(db_type="json", base_path=test_db_path) -test_ctx = GraphContext(database=test_db) -``` - ---- - -## Semantic Filtering - -**File**: [`examples/semantic_filtering.py`](examples/semantic_filtering.py) - -**What it demonstrates**: -- Advanced query capabilities with MongoDB-style operators -- Database-level optimization -- Complex filtering combining nodes and edges -- Performance-oriented query patterns - -### Key Features Shown - -**Simple Type Filtering**: -```python path=examples/semantic_filtering.py start=155 -cities = await new_york.nodes(node="City") -``` - -**Property Filtering via kwargs**: -```python path=examples/semantic_filtering.py start=160 -ma_connections = await new_york.nodes(state="MA") -``` - -**Complex Node Filtering with MongoDB Operators**: -```python path=examples/semantic_filtering.py start=169 -large_cities = await new_york.nodes( - node=[{"City": {"context.population": {"$gte": 1_000_000}}}] -) -``` - -**Complex Edge Filtering**: -```python path=examples/semantic_filtering.py start=189 -fast_highways = await new_york.nodes( - edge=[{"Highway": {"context.speed_limit": {"$gte": 65}}}] -) -``` - -**Combined Filters**: -```python path=examples/semantic_filtering.py start=195 -good_free_roads = await new_york.nodes( - edge=[{ - "Highway": { - "context.condition": {"$ne": "poor"}, - "context.toll_road": False, - } - }] -) -``` - ---- - -## Additional Examples - -The `examples/` directory contains many more specialized examples: - -### Core Examples (All Passing) -- **`walker_traversal_demo.py`** - Walker traversal patterns -- **`walker_events_demo.py`** - Walker event communication system -- **`walker_reporting_demo.py`** - Walker reporting system with `report()` pattern -- **`database/multi_database_example.py`** - **Multi-database support with prime database** - Managing multiple databases, switching, and removing non-prime databases -- **`database_switching_example.py`** - Multiple database backends -- **`multi_target_hooks_demo.py`** - Advanced walker hooks with `@on_visit(TypeA, TypeB)` -- **`object_pagination_demo.py`** - Pagination and performance -- **`testing_with_graphcontext.py`** - Testing strategies with database isolation -- **`semantic_filtering.py`** - Advanced MongoDB-style queries -- **`enhanced_nodes_filtering.py`** - Enhanced filtering capabilities -- **`unified_query_interface_example.py`** - Advanced query interfaces -- **`exception_handling_demo.py`** - Error handling patterns - -### Server Examples -- **`simple_dynamic_example.py`** - Dynamic endpoint registration patterns -- **`server_demo.py`** - Comprehensive server features -- **`auth_demo.py`** - JWT and API key authentication (runs indefinitely) -- **`scheduler_example.py`** - Scheduled tasks (runs indefinitely) -- **`dynamic_server_demo.py`** - Advanced dynamic endpoints -- **`endpoint_decorator_demo.py`** - Endpoint decorator patterns -- **`fastapi_server.py`** - FastAPI integration - -## Running the Examples - -Each example can be run independently: +Run from the repository root after installing the dependencies documented in +the relevant group's README: ```bash -cd examples/ -python walker_traversal_demo.py -python travel_graph.py -python agent_graph.py -# ... etc +python examples/database/sqlite_example.py +python examples/storage/file_storage_demo.py +python examples/core/graph_visualization_example.py ``` -Some examples (like `dynamic_server_demo.py`) start web servers and should be accessed via HTTP endpoints. - -## Key Architectural Concepts - -1. **Objects**: Simple data storage units -2. **Nodes**: Connected objects that can be visited by walkers -3. **Edges**: Typed connections between nodes with their own properties -4. **Walkers**: Traverse nodes along edges, implementing business logic -5. **GraphContext**: Dependency injection for database connections -6. **Spatial Awareness**: Built-in support for geographic data and calculations - -The examples demonstrate how these concepts work together to create powerful, maintainable graph applications with spatial capabilities. +Server examples start local HTTP services; database and cloud-storage examples +may require a reachable backend or configured credentials. Inspect their setup +before running them against an existing installation. Example completion does +not qualify production authorization, storage durability or deployed topology. -## See Also +## Related guides -- **[Authentication System Documentation](authentication.md)** - Complete authentication guide with JWT, API keys, and RBAC -- **[Authentication Quickstart](auth-quickstart.md)** - Get authentication working in 5 minutes -- **[REST API with Authentication](rest-api.md#authentication-integration)** - Securing your API endpoints -- **[Server API Guide](server-api.md)** - Complete server configuration and setup +- [Quick Start](quick-start-guide.md) +- [Entity Reference](entity-reference.md) +- [Authentication](authentication.md) +- [Graph Visualization](graph-visualization.md) +- [Graph Context for Inference](graph-projection.md) +- [Transactions](transactions.md) diff --git a/docs/md/file-storage-usage.md b/docs/md/file-storage-usage.md index 23f484c..2ddc92d 100644 --- a/docs/md/file-storage-usage.md +++ b/docs/md/file-storage-usage.md @@ -1642,14 +1642,14 @@ async def upload_user_file(file: UploadFile, endpoint): ### Related Documentation - [File Storage Architecture](file-storage-architecture.md) - Technical architecture details -- [Server Configuration](server-configuration.md) - Complete server setup guide -- [Walker Guide](walker-guide.md) - Walker development documentation -- [API Reference](api-reference.md) - Complete API documentation +- [Server API](server-api.md) - Complete server setup guide +- [Graph Traversal](graph-traversal.md) - Walker development documentation +- [API Architecture](api-architecture.md) - Complete API documentation ### Example Projects -- [`examples/file_storage_demo.py`](../../examples/file_storage_demo.py) - Complete working example -- [`examples/s3_storage_example.py`](../../examples/s3_storage_example.py) - S3 integration example +- [`examples/storage/file_storage_demo.py`](../../examples/storage/file_storage_demo.py) - Complete working example +- [S3 storage interface](../../jvspatial/storage/interfaces/s3.py) - S3 implementation reference ### External Resources diff --git a/docs/md/graph-visualization.md b/docs/md/graph-visualization.md index 5ef1821..c73706f 100644 --- a/docs/md/graph-visualization.md +++ b/docs/md/graph-visualization.md @@ -606,7 +606,7 @@ dot = await generate_graph_dot( - [Graph Traversal](graph-traversal.md) - Traversing graphs with Walkers - [Node Operations](node-operations.md) - Working with nodes - [Entity Reference](entity-reference.md) - Complete entity API -- [Examples](../examples/core/graph_visualization_example.py) - Complete example +- [Examples](../../examples/core/graph_visualization_example.py) - Complete example --- diff --git a/docs/md/migration.md b/docs/md/migration.md index cfaf8ca..80ce49b 100644 --- a/docs/md/migration.md +++ b/docs/md/migration.md @@ -708,8 +708,8 @@ server = Server( ## See Also - [Design Decisions](design-decisions.md) -- [Core Concepts](core-concepts.md) +- [Entity Reference](entity-reference.md) - [API Reference](rest-api.md) -- [Best Practices](best-practices.md) +- [Contributing Guide](contributing.md) - [Authentication Guide](authentication.md) - [API Keys Guide](api-keys.md) \ No newline at end of file diff --git a/docs/md/observability.md b/docs/md/observability.md index 51a908d..71e3646 100644 --- a/docs/md/observability.md +++ b/docs/md/observability.md @@ -6,8 +6,10 @@ unless you turn it on. ## Structured database logging -Wrapping a database with `observe=True` emits a single structured log -line per operation with this fixed schema: +Wrapping a database with `observe=True` emits structured logs only when an +operation fails. Successful operations, including slow calls, stay out of +application logs; per-operation detail and slow-call counts remain available +through the metrics recorder. Failures log at ERROR using this schema: | Field | Type | Notes | | --------------- | -------- | ------------------------------------------------ | @@ -18,12 +20,7 @@ line per operation with this fixed schema: | `success` | bool | False if the call raised | | `result_count` | int? | Where applicable: 0/1 for single-doc ops, list length for `find`, the count for `count` | -Successful operation timings are emitted at **DEBUG** level. Failed -operations and operations with `duration_ms >= slow_query_ms` use **WARNING** -(the default slow-query threshold is 100 ms). Routine timings therefore stay -out of INFO-level application logs; metrics are recorded regardless of the -logger level. -The standard fields land in `record.__dict__` via the logging +Failure records land in `record.__dict__` via the logging `extra=` channel, so structured-log handlers (json formatters, `structlog`, OpenTelemetry log exporters) pick them up directly. @@ -37,25 +34,13 @@ db = create_database( slow_query_ms=50.0, # tighten or loosen as needed ) -# Enable DEBUG on jvspatial.db.observable to see routine operation timings. +# Successful operations do not log; failures remain visible. await db.get("node", "abc") -# 2026-05-08T... DEBUG jvspatial.db.observable -# db.get on 'node' took 1.42ms -# {backend: JsonDB, op: get, collection: node, success: true, -# duration_ms: 1.42, result_count: 1} +# Failures emit a structured ERROR record. ``` -The logger name is `jvspatial.db.observable`. Add a handler / level -filter against that name when you want to direct DB telemetry to a -specific sink. For example, to opt into routine timing logs: - -```python -import logging - -logging.getLogger("jvspatial.db.observable").setLevel(logging.DEBUG) -``` - -The application must also configure a handler that accepts DEBUG records. +The logger name is `jvspatial.db.observable`. Route that logger when you want +database failure records sent to a specific sink. ## Metrics @@ -158,19 +143,17 @@ backend (JsonDB / SQLiteDB / MongoDB / DynamoDB) +— [if observe] ObservableDatabase ``` -That ordering means the structured log line measures the -**user-visible** latency including cache hits and misses — which is -what SLO calculations need. The cache hit/miss distinction is -visible in two places: the `duration_ms` field (a hit is much -faster) and via `db.cache_stats()` if you reach through to the -inner cache wrapper. +That ordering means metrics and slow-query counters measure the +**user-visible** latency including cache hits and misses. The cache +hit/miss distinction is visible in `duration_ms` (a hit is much faster) +and via `db.cache_stats()` if you reach through to the inner cache wrapper. ## Serverless The metrics layer itself works fine in serverless — every emission -is a synchronous in-process call. The structured log lines are -written through the standard `logging` module, which the runtime's -log forwarder will pick up. +is a synchronous in-process call. Slow-query warnings are written +through the standard `logging` module, which the runtime's log forwarder +will pick up. The cache layer (`CachingDatabase`) is automatically disabled in serverless mode because cold starts make a per-process cache useless; diff --git a/docs/md/optimization.md b/docs/md/optimization.md index 3637f08..885f616 100644 --- a/docs/md/optimization.md +++ b/docs/md/optimization.md @@ -464,7 +464,7 @@ configure_pool( ## See Also -- [Database Configuration](configuration.md) +- [Database Configuration](server-api.md) - [Caching System](caching.md) -- [Walker Patterns](walker-patterns.md) -- [Monitoring Guide](monitoring.md) \ No newline at end of file +- [Graph Traversal](graph-traversal.md) +- [Observability](observability.md) \ No newline at end of file diff --git a/docs/md/postgres-guide.md b/docs/md/postgres-guide.md index 3fff171..cc4c491 100644 --- a/docs/md/postgres-guide.md +++ b/docs/md/postgres-guide.md @@ -319,11 +319,19 @@ async with transaction_context(db) as txn: else: await txn.save("user", user_record) await txn.save("audit_log", log_record) + # Query methods (find, find_one, count, find_ids, find_many, + # find_iter) read from this transaction's held connection. # Commits on context exit; rolls back on exception. ``` The transaction holds one dedicated connection out of the pool for its lifetime, so be careful not to hold transactions across long-running awaits. +Concurrent handle operations serialize access to that connection, and completed +handles reject further reads or writes. Tenant scope and new-collection +bootstrap use the held connection; a rollback does not mark collection DDL as +committed. Graph transactions also isolate identity/process caches. See the +[shared transaction contract](transactions.md) for other backends and fallback +performance limits. ## Bulk writes diff --git a/docs/md/quick-start-guide.md b/docs/md/quick-start-guide.md index cf52c18..844b5b9 100644 --- a/docs/md/quick-start-guide.md +++ b/docs/md/quick-start-guide.md @@ -337,7 +337,7 @@ python walkers/walker_traversal_demo.py ### **Join the Community** -- 📖 **Documentation**: [docs/](../README.md) +- 📖 **Documentation**: [docs/](../../README.md) - 💬 **Discussions**: GitHub Discussions - 🐛 **Issues**: GitHub Issues - 📧 **Email**: support@jvspatial.com diff --git a/docs/md/transactions.md b/docs/md/transactions.md new file mode 100644 index 0000000..077de63 --- /dev/null +++ b/docs/md/transactions.md @@ -0,0 +1,109 @@ +# Transactions + +Native transactions are available on PostgreSQL and on transaction-capable +MongoDB deployments. JSON, SQLite and DynamoDB do not advertise native +transactions. Single-record atomic writes on those stores are not multi-write +rollback or cross-store atomicity. + +## Capability and lifecycle + +Check `db.supports_transactions`, rather than detecting an adapter class. +MongoDB requires a replica set or sharded cluster. Its static support flag +identifies the adapter feature; `await db.is_transactional()` probes the +deployment. The flag alone does not prove topology support. Unsupported `transaction_context` yields `None` by default. +Callers requiring atomic work must reject that case rather than silently +performing unprotected writes. + +```python +from jvspatial.db import create_database +from jvspatial.db.transaction import transaction_context + +db = create_database("postgres", dsn="postgresql://localhost/mydb") +async with transaction_context(db) as tx: + if tx is None: + raise RuntimeError("This operation requires native transactions") + await tx.save("object", { + "id": "o.Record.example", + "entity": "Record", + "context": {"state": "ready"}, + }) + ids = await tx.find_ids("object", {"entity": "Record"}, limit=20) + records = await tx.find_many("object", ids) + total = await tx.count("object", {"entity": "Record"}) +``` + +Normal exit commits; an exception or cancellation rolls back. Keep external network calls and +long computation outside the transaction. A PostgreSQL handle owns one pool +connection until exit, and rejects operations after it has completed. Concurrent +operations on that handle are serialized to prevent overlapping asyncpg +commands; callers still own application-level coordination of mutable entities. +New collection creation and its indexes run on the held connection, so a +one-connection pool does not require a second connection for bootstrap. +Rollback discards pending bootstrap cache entries. MongoDB commit/rollback +await session cleanup for both asynchronous driver shapes. + +## Read contract + +| Method | Result | Implementation | +| --- | --- | --- | +| `get(collection, id)` | Record or `None` | Native transaction read | +| `find(collection, query, sort=, limit=)` | Record list | Native transaction read | +| `find_one(collection, query)` | Record or `None` | One matching record | +| `count(collection, query=None)` | Integer | PostgreSQL native count; shared fallback reads transaction-visible matches | +| `find_ids(collection, query, limit=, sort=, offset=)` | Ordered ID list | PostgreSQL native projection; shared fallback retains deterministic ID tie-breaking | +| `find_many(collection, ids)` | ID-to-record dictionary | PostgreSQL native set read; shared fallback uses handle-local `get` | +| `find_iter(collection, query, sort=, batch_size=, cursor=)` | Async iterator | Shared composite-keyset paging through transaction reads | +| `count_incident_edges(collection, node_ids)` | ID-to-degree dictionary | Shared transaction-visible edge count | + +Fallbacks preserve the same connection/session and visibility, but may perform +more work than native aggregates or projections. In particular, the MongoDB +shared count and ID projection can materialize records, and shared multi-get +is sequential. They are correctness fallbacks, not a claim of identical +round-trip counts or performance across adapters. Iteration requires a positive +batch size and preserves every sort key plus ID in continuation. Backend-native +ordering differences remain documented in [Cursor Pagination](cursor-pagination.md). + +## Entity and graph operations + +Raw transaction methods operate on database records. Entity `save`/`get` and +relationship operations must use `graph_transaction` to bind their context to +the transaction handle; using the outer database explicitly escapes that scope. +If a deployment returns no transaction, `graph_transaction` raises +`TransactionUnavailable` before binding a context. It never silently falls back +to the outer store. Every application-created Node still needs its normal rooted structural edge +in the same transaction. + +```python +from jvspatial.core import graph_transaction + +# Item is an application-defined Node already attached to its application root. +async with graph_transaction(db): + item = await Item.get(item_id) + if item is not None: + item.title = new_title + await item.save() +``` + +A graph transaction gets an isolated request identity map. It neither returns +outer-context objects from the shared cache nor publishes uncommitted objects +into that cache. An explicitly selected different context does not reuse an +identity-map entry bound to the transaction. Touched parent cache keys are +invalidated after commit and rollback. Archived application state and any +long-lived object references remain the application's responsibility. + +PostgreSQL's active `db.tenant(...)` scope is carried onto the held transaction +connection with transaction-local settings. This supports an application's +configured RLS policy; it does not create an access policy or replace +application authorization. + +## Monitoring and limits + +The optional observable wrapper records metrics and failure-only ERROR logs +for operations invoked through that wrapper. It forwards native transaction +handles; raw handle operations are not automatically instrumented as wrapped +database operations. Successful calls, including slow calls, never produce +routine observation logs. See [Observability](observability.md). + +Use backend-specific integration tests to qualify rollback, tenant scope, +concurrency and deployed MongoDB topology. Neither the shared API nor an +in-memory test double proves cross-store atomicity or live replica-set behavior. diff --git a/jvspatial/core/context.py b/jvspatial/core/context.py index 368c49e..78c6ded 100644 --- a/jvspatial/core/context.py +++ b/jvspatial/core/context.py @@ -587,9 +587,22 @@ async def _get_from_cache(self, entity_id: str) -> Optional[Any]: imap = _request_identity_map.get() if imap is not None and entity_id in imap: result = imap[entity_id] + # Identity-map entries are context-bound. A transaction replaces + # the request map, but callers may explicitly use another context + # while that map is active; do not return an entity attached to a + # different database handle in that case. + if getattr(result, "_graph_context", self) is self: + if self._perf_monitoring_enabled and self._perf_monitor: + self._perf_monitor.record_cache_hit() + return result + # A graph transaction owns an isolated identity map. Consulting the + # process cache here could return an entity bound to the outer + # database context, causing reads to escape the transaction and + # writes to bypass its connection (or expose uncommitted state). + if self._transaction_cache_keys is not None: if self._perf_monitoring_enabled and self._perf_monitor: - self._perf_monitor.record_cache_hit() - return result + self._perf_monitor.record_cache_miss() + return None result = await self._cache.get(entity_id) # Track cache hit/miss if self._perf_monitoring_enabled and self._perf_monitor: @@ -606,6 +619,8 @@ async def _add_to_cache(self, entity_id: str, entity: Any) -> None: imap = _request_identity_map.get() if imap is not None: imap[entity_id] = entity + if self._transaction_cache_keys is not None: + return await self._cache.set(entity_id, entity) async def _evict_from_cache(self, entity_id: str) -> None: @@ -2275,7 +2290,8 @@ async def graph_transaction(database: Optional[Any] = None): back every write in the block. Requires a backend that exposes ``begin_transaction``, - ``commit_transaction``, and ``rollback_transaction`` (Postgres today). + ``commit_transaction``, and ``rollback_transaction`` (PostgreSQL and + transaction-capable MongoDB deployments). Usage: async with graph_transaction(db) as ctx: @@ -2293,6 +2309,10 @@ async def graph_transaction(database: Optional[Any] = None): "graph_transaction requires a database with begin/commit/rollback" ) txn = await db.begin_transaction() + if txn is None: + raise TransactionUnavailable( + "The configured deployment cannot start a native transaction" + ) ctx = GraphContext(txn) parent_ctx = _default_context_var.get() or _module_default_context ctx._transaction_cache_keys = set() diff --git a/jvspatial/db/README.md b/jvspatial/db/README.md index b557e2a..56429ad 100644 --- a/jvspatial/db/README.md +++ b/jvspatial/db/README.md @@ -8,7 +8,7 @@ Database abstraction and backends — JSON, SQLite, MongoDB, DynamoDB, Postgres. ## Purpose -`db/` defines the `Database` ABC and ships four production backends, all sharing a Mongo-style query DSL. Internal wrappers (`_cache`, `_observable`) add cross-cutting concerns to any backend through factory flags. +`db/` defines the `Database` ABC and ships five built-in backends, all sharing a Mongo-style query DSL. Internal wrappers (`_cache`, `_observable`) add cross-cutting concerns to any backend through factory flags. ## Layout @@ -18,7 +18,7 @@ db/ ├── factory.py # create_database, register_database_type, switch_database ├── manager.py # DatabaseManager + prime DB convention ├── query.py # QueryEngine + Mongo-style operators -├── transaction.py # Transaction context + MongoDBTransaction +├── transaction.py # Shared transaction read API + MongoDBTransaction ├── jsondb.py # JSON file-per-record backend ├── sqlite.py # aiosqlite backend + Mongo→SQL translator ├── mongodb.py # motor backend @@ -27,7 +27,7 @@ db/ ├── _atomic.py # internal: crash-safe write helper ├── _path_locks.py # internal: bounded-LRU per-path locks ├── _cache.py # internal: read-through cache wrapper -├── _observable.py # internal: structured log + metrics wrapper +├── _observable.py # internal: metrics + failure-log wrapper └── _sqlite_translate.py # internal: Mongo → SQL translator ``` @@ -86,9 +86,10 @@ db/ - [docs/md/graph-context.md](../../docs/md/graph-context.md) - [docs/md/caching.md](../../docs/md/caching.md) - [docs/md/observability.md](../../docs/md/observability.md) +- [docs/md/transactions.md](../../docs/md/transactions.md) - [docs/md/optimization.md](../../docs/md/optimization.md) - [docs/md/benchmarks.md](../../docs/md/benchmarks.md) ## Stability -Public surface (above) is stable. Underscore-prefixed modules are internal. The structured log field set from `ObservableDatabase` is part of the public contract — schema changes require a deprecation cycle. +Public surface (above) is stable. Underscore-prefixed modules are internal. Failure-only ERROR logs from `ObservableDatabase` retain the public structured field schema documented in [observability](../../docs/md/observability.md). Successful and slow operations are measured through metrics, without log records. diff --git a/jvspatial/db/_observable.py b/jvspatial/db/_observable.py index 1b1d0a3..145e6f2 100644 --- a/jvspatial/db/_observable.py +++ b/jvspatial/db/_observable.py @@ -1,12 +1,10 @@ """Observability wrapper for :class:`Database` instances. Wraps any backend (raw or already wrapped in :class:`CachingDatabase`) -and emits, for each operation: +and records, for each operation: -1. A structured log line with the standard fields - ``backend`` / ``op`` / ``collection`` / ``duration_ms`` / ``result_count`` - / ``success``. DEBUG for successful calls under the threshold; - WARNING for failures or calls reaching ``slow_query_ms``. +1. A structured ERROR for failed operations. Successful operations, including + calls above ``slow_query_ms``, produce no log record. 2. One metrics emission to the configured :class:`MetricsRecorder`: a duration histogram and a counter, both labeled with the standard dimensions. @@ -63,9 +61,8 @@ logger = logging.getLogger("jvspatial.db.observable") -# Default slow-query threshold. Anything above 100 ms gets surfaced as -# a WARNING so it shows up in normal log feeds without operator -# configuration. Tunable per-instance via ``slow_query_ms``. +# Default slow-query threshold. Calls above it increment a metric; they do not +# generate log records. Tunable per-instance via ``slow_query_ms``. DEFAULT_SLOW_QUERY_MS = 100.0 @@ -83,15 +80,16 @@ def _backend_label(inner: Database) -> str: class ObservableDatabase(Database): - """Wraps a :class:`Database` with structured logging + metrics. + """Wrap a :class:`Database` with per-operation metrics and sparse logs. Args: inner: The wrapped database. May itself be wrapped (e.g. by :class:`~jvspatial.db._cache.CachingDatabase`). metrics: A :class:`MetricsRecorder`. Defaults to :class:`NullMetricsRecorder` (zero overhead). - slow_query_ms: Threshold above which the per-op log line is - elevated from DEBUG to WARNING. Defaults to 100ms. + slow_query_ms: Threshold above which successful calls increment the + slow-operation metric. Successful calls are not logged. Failures + are logged at ERROR. Defaults to 100ms. """ def __init__( @@ -123,7 +121,7 @@ async def _instrument( *, result_count_extractor: Optional[Callable[[Any], int]] = None, ) -> Any: - """Run ``coro_factory()`` while emitting a log line and metric. + """Run ``coro_factory()`` while recording metrics and failure logs. ``coro_factory`` is a thunk so we measure the underlying call only, not the time spent constructing the coroutine. @@ -186,26 +184,24 @@ def _emit( "collection": collection, "success": success, } - log_extra = {**labels, "duration_ms": round(duration_ms, 3)} - if result_count is not None: - log_extra["result_count"] = result_count - - if query_plan: - log_extra.update(query_plan) - if active_duration_s is not None: - log_extra["active_duration_ms"] = round(active_duration_s * 1000, 3) - - # Structured log. WARNING when slow or failed, DEBUG otherwise. extra= - # supplies the structured fields to any handler that - # understands them (json formatters, structlog, etc.). + # Build failure-only log data lazily. Successful calls do not allocate + # a logging payload or format a message, even when DEBUG is enabled. slow = duration_ms >= self.slow_query_ms - msg = "db.%s on '%s' took %.2fms" % (op, collection, duration_ms) - if slow: - logger.warning("SLOW %s", msg, extra=log_extra) - elif not success: - logger.warning("FAILED %s", msg, extra=log_extra) - else: - logger.debug(msg, extra=log_extra) + if not success and logger.isEnabledFor(logging.ERROR): + log_extra = {**labels, "duration_ms": round(duration_ms, 3)} + if result_count is not None: + log_extra["result_count"] = result_count + if query_plan: + log_extra.update(query_plan) + if active_duration_s is not None: + log_extra["active_duration_ms"] = round(active_duration_s * 1000, 3) + logger.error( + "FAILED db.%s on '%s' took %.2fms", + op, + collection, + duration_ms, + extra=log_extra, + ) # Metrics. Best-effort -- do not surface backend errors here. try: diff --git a/jvspatial/db/factory.py b/jvspatial/db/factory.py index b283182..861e5ce 100644 --- a/jvspatial/db/factory.py +++ b/jvspatial/db/factory.py @@ -278,12 +278,12 @@ def create_database( TTL (LRU eviction only). observe: When ``True``, wrap the database in an :class:`~jvspatial.db._observable.ObservableDatabase` that - emits a structured log line and one metric per operation. + records metrics per operation and logs failures only. Defaults to ``False`` (no behavior change for existing callers). - slow_query_ms: Threshold above which the per-op log line is - elevated to WARNING. Only meaningful when - ``observe=True``. Defaults to 100ms. + slow_query_ms: Threshold above which successful operations are + counted as slow in metrics. Successful operations are not + logged. Only meaningful when ``observe=True``. Defaults to 100ms. metrics: Optional :class:`~jvspatial.observability.metrics.MetricsRecorder` implementation. Defaults to :class:`~jvspatial.observability.metrics.NullMetricsRecorder` @@ -345,9 +345,8 @@ def create_database( db = CachingDatabase(db, max_entries=cache_get_size, ttl_seconds=cache_get_ttl) # Optional observability layer (off by default). Applied AFTER the - # cache so the structured log line measures user-visible latency - # including cache hits/misses -- which is what SLO calculations - # need. + # cache so metrics and slow-query counters measure user-visible + # latency including cache hits/misses. if observe: db = ObservableDatabase(db, metrics=metrics, slow_query_ms=slow_query_ms) diff --git a/jvspatial/db/mongodb.py b/jvspatial/db/mongodb.py index c20cabc..0b972d5 100644 --- a/jvspatial/db/mongodb.py +++ b/jvspatial/db/mongodb.py @@ -1179,8 +1179,8 @@ async def begin_transaction(self): Returns a ``MongoDBTransaction`` whose session is bound to a ``start_transaction`` call. If the deployment does not support transactions (standalone / DocumentDB without transactions), the - returned object will be ``None`` and callers should fall back to - non-transactional writes. + returned object will be ``None``. Callers requiring atomicity must + reject that case rather than perform unprotected writes. """ from jvspatial.db.transaction import MongoDBTransaction @@ -1208,13 +1208,13 @@ async def commit_transaction(self, txn) -> None: """Commit the given transaction and end its session.""" if txn is not None: await txn.commit() - txn.session.end_session() + await _resolve_driver_result(txn.session.end_session()) async def rollback_transaction(self, txn) -> None: """Roll back the given transaction and end its session.""" if txn is not None: await txn.rollback() - txn.session.end_session() + await _resolve_driver_result(txn.session.end_session()) async def drop_deprecated_indexes(self, deprecated: Dict[str, List[str]]) -> None: """Drop indexes that have been removed or renamed in code. diff --git a/jvspatial/db/postgres.py b/jvspatial/db/postgres.py index 0a212dc..63ddcda 100644 --- a/jvspatial/db/postgres.py +++ b/jvspatial/db/postgres.py @@ -73,6 +73,7 @@ import logging import re import time +from functools import wraps from typing import ( TYPE_CHECKING, Any, @@ -96,6 +97,7 @@ finalize_find_results, ) from .query import QueryEngine +from .transaction import Transaction # Active tenant id for the current async task. ``None`` means no tenant # scope is in effect — operations run without setting the ``app.tenant_id`` @@ -117,6 +119,51 @@ def current_tenant() -> Optional[str]: logger = logging.getLogger(__name__) + +class _TaskReentrantLock: + """Serialize operations on one asyncpg connection, allowing same-task calls.""" + + def __init__(self) -> None: + self._lock = asyncio.Lock() + self._owner: Optional[asyncio.Task[Any]] = None + self._depth = 0 + + async def __aenter__(self) -> "_TaskReentrantLock": + task = asyncio.current_task() + if task is None: + raise RuntimeError( + "PostgreSQL transaction operations require an asyncio task" + ) + if self._owner is task: + self._depth += 1 + return self + await self._lock.acquire() + self._owner = task + self._depth = 1 + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + task = asyncio.current_task() + if task is not self._owner: + raise RuntimeError("PostgreSQL transaction lock released by non-owner") + self._depth -= 1 + if self._depth == 0: + self._owner = None + self._lock.release() + + +def _serialized_transaction_operation(method): + """Guard one transaction operation against concurrent asyncpg commands.""" + + @wraps(method) + async def wrapped(self: Any, *args: Any, **kwargs: Any) -> Any: + async with self._operation_lock: + self._ensure_active() + return await method(self, *args, **kwargs) + + return wrapped + + try: import asyncpg except ImportError: # pragma: no cover - handled at __init__ time @@ -308,12 +355,8 @@ class PostgresDB(Database): full set of configuration knobs. """ - # Postgres supports ACID transactions natively. The base - # ``begin_transaction`` machinery is wired in a later patch - # (C4: PG atomicity) — for now ``supports_transactions`` advertises - # the capability so callers can branch on it. Until ``begin_transaction`` - # lands, transactional context is offered only via ``UPDATE … RETURNING`` - # at the operation level (atomic single-row update). + # PostgreSQL supports full ACID transactions backed by a held pool + # connection. See ``begin_transaction`` and ``PostgresTransaction`` below. supports_transactions: bool = True supports_native_graph_paging: bool = True @@ -637,22 +680,53 @@ async def enable_rls( # ---- schema bootstrap -------------------------------------------------- - async def _bootstrap_collection(self, collection: str) -> None: + async def _bootstrap_collection( + self, + collection: str, + *, + connection: Optional[Any] = None, + pending: Optional[Set[str]] = None, + ) -> None: """Create the table + base indexes for ``collection`` if missing. Idempotent + cached. Safe to call from any code path before a CRUD op; the cost on the warm path after first call is one set membership check. """ - if collection in self._collections_bootstrapped: + if collection in self._collections_bootstrapped or ( + pending is not None and collection in pending + ): return lock = self._collection_bootstrap_locks.setdefault(collection, asyncio.Lock()) async with lock: - if collection in self._collections_bootstrapped: + if collection in self._collections_bootstrapped or ( + pending is not None and collection in pending + ): return - await self._create_collection_schema(collection) + if connection is None: + await self._create_collection_schema(collection) + self._collections_bootstrapped.add(collection) + else: + await self._create_collection_schema_on_connection( + collection, connection + ) + if pending is not None: + pending.add(collection) async def _create_collection_schema(self, collection: str) -> None: + pool = await self._ensure_pool() + async with pool.acquire() as conn: + await self._create_collection_schema_on_connection(collection, conn) + + async def _create_collection_schema_on_connection( + self, collection: str, conn: Any + ) -> None: + """Create a collection using a caller-owned connection. + + Transactions use this path to avoid checking out a second pool + connection while already holding one. The DDL remains part of the + caller's transaction and is only cached globally after commit. + """ col = _safe_collection(collection) schema = _safe_collection(self.schema_name) gin_name = _postgres_index_name(col, "data_gin") @@ -664,21 +738,18 @@ async def _create_collection_schema(self, collection: str) -> None: if self.gin_index == "full" else "" ) - pool = await self._ensure_pool() - async with pool.acquire() as conn: - # An asyncio lock only coordinates callers sharing this instance. - # PostgreSQL must also serialize first use across workers and - # independently constructed PostgresDB instances. Keep the - # advisory lock and DDL in one transaction so the lock survives - # until the table and all base indexes have been created. - async with conn.transaction(): - await conn.execute( - "SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", - schema, - col, - ) - await conn.execute( - f""" + # An asyncio lock only coordinates callers sharing this instance. + # PostgreSQL must also serialize first use across workers and + # independently constructed PostgresDB instances. Keep the advisory + # lock and DDL in a transaction so the lock lasts through creation. + async with conn.transaction(): + await conn.execute( + "SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", + schema, + col, + ) + await conn.execute( + f""" CREATE TABLE IF NOT EXISTS {schema}.{col} ( id TEXT PRIMARY KEY, entity TEXT NOT NULL, @@ -695,8 +766,7 @@ async def _create_collection_schema(self, collection: str) -> None: ON {schema}.{col} (tenant_id) WHERE tenant_id IS NOT NULL; """ - ) - self._collections_bootstrapped.add(collection) + ) # ---- payload helpers --------------------------------------------------- @@ -1106,30 +1176,35 @@ async def count_incident_edges( counts.update({row["nid"]: int(row["n"]) for row in rows}) return counts - async def find_ids( + async def _find_ids_with_connection( self, + connection: Any, collection: str, query: Dict[str, Any], *, - limit: Optional[int] = None, - sort: Optional[List[Tuple[str, int]]] = None, - offset: int = 0, - ) -> List[str]: - """Project physical IDs, with filtering/sorting/offset in PostgreSQL.""" + limit: Optional[int], + sort: Optional[List[Tuple[str, int]]], + offset: int, + ) -> Optional[List[str]]: + """Shared native ID projection for pooled and transactional reads. + + ``None`` signals that the caller must use its own transaction-safe + fallback for an unsupported filter or sort. + """ from .database import _keyset_sort_fields if offset < 0 or (limit is not None and limit < 0): raise ValueError("offset and limit must be nonnegative") + if limit == 0: + return [] translated = translate_query(query) ordering = translate_sort(_keyset_sort_fields(sort)) if translated is None or ordering is None: - return await super().find_ids( - collection, query, limit=limit, sort=sort, offset=offset - ) - await self._bootstrap_collection(collection) - col, schema = _safe_collection(collection), _safe_collection(self.schema_name) + return None where, values = translated params = list(values) + col = _safe_collection(collection) + schema = _safe_collection(self.schema_name) sql = f"SELECT id FROM {schema}.{col}" if where: sql += f" WHERE {where}" @@ -1139,10 +1214,68 @@ async def find_ids( sql += f" LIMIT ${len(params)}" params.append(offset) sql += f" OFFSET ${len(params)}" - async with self._acquire_conn() as conn: - rows = await conn.fetch(sql, *params) + rows = await connection.fetch(sql, *params) return [row["id"] for row in rows] + async def _count_with_connection( + self, connection: Any, collection: str, query: Dict[str, Any] + ) -> Optional[int]: + """Shared native COUNT; ``None`` signals an in-memory fallback.""" + translated = translate_query(query) if query else ("", []) + if translated is None: + return None + where_sql, params = translated + clause = f" WHERE {where_sql}" if where_sql else "" + col = _safe_collection(collection) + schema = _safe_collection(self.schema_name) + row = await connection.fetchrow( + f"SELECT COUNT(*) AS n FROM {schema}.{col}{clause}", *params + ) + return int(row["n"]) if row else 0 + + async def _find_many_with_connection( + self, connection: Any, collection: str, ids: List[str] + ) -> Dict[str, Dict[str, Any]]: + """Shared held-connection implementation for set ID lookups.""" + if not ids: + return {} + unique_ids = list(dict.fromkeys(str(value) for value in ids)) + col = _safe_collection(collection) + schema = _safe_collection(self.schema_name) + rows = await connection.fetch( + f"SELECT id, data FROM {schema}.{col} WHERE id = ANY($1::text[])", + unique_ids, + ) + return {row["id"]: self._record_from_row(row) for row in rows} + + async def find_ids( + self, + collection: str, + query: Dict[str, Any], + *, + limit: Optional[int] = None, + sort: Optional[List[Tuple[str, int]]] = None, + offset: int = 0, + ) -> List[str]: + """Project physical IDs, with filtering/sorting/offset in PostgreSQL.""" + if limit == 0: + return [] + await self._bootstrap_collection(collection) + async with self._acquire_conn() as conn: + ids = await self._find_ids_with_connection( + conn, + collection, + query, + limit=limit, + sort=sort, + offset=offset, + ) + if ids is not None: + return ids + return await super().find_ids( + collection, query, limit=limit, sort=sort, offset=offset + ) + async def count( self, collection: str, query: Optional[Dict[str, Any]] = None ) -> int: @@ -1154,16 +1287,13 @@ async def count( """ q = query or {} await self._bootstrap_collection(collection) - col = _safe_collection(collection) - schema = _safe_collection(self.schema_name) if not q: async with self._acquire_conn() as conn: - row = await conn.fetchrow(f"SELECT COUNT(*) AS n FROM {schema}.{col}") - return int(row["n"]) if row else 0 + result = await self._count_with_connection(conn, collection, q) + return result if result is not None else 0 - translated = translate_query(q) - if translated is None: + if translate_query(q) is None: total = 0 candidates = self._fallback_records(collection, q) try: @@ -1172,13 +1302,9 @@ async def count( finally: await candidates.aclose() return total - where_sql, params = translated - clause = f" WHERE {where_sql}" if where_sql else "" async with self._acquire_conn() as conn: - row = await conn.fetchrow( - f"SELECT COUNT(*) AS n FROM {schema}.{col}{clause}", *params - ) - return int(row["n"]) if row else 0 + result = await self._count_with_connection(conn, collection, q) + return result or 0 async def find_many( self, collection: str, ids: List[str] @@ -1186,19 +1312,9 @@ async def find_many( """Bulk-fetch by id with a single ``WHERE id = ANY($1)`` round trip.""" if not ids: return {} - unique_ids = list(dict.fromkeys(str(i) for i in ids)) await self._bootstrap_collection(collection) - col = _safe_collection(collection) - schema = _safe_collection(self.schema_name) async with self._acquire_conn() as conn: - rows = await conn.fetch( - f"SELECT id, data FROM {schema}.{col} WHERE id = ANY($1::text[])", - unique_ids, - ) - out: Dict[str, Dict[str, Any]] = {} - for row in rows: - out[row["id"]] = self._record_from_row(row) - return out + return await self._find_many_with_connection(conn, collection, ids) def _connected_nodes_sql( self, @@ -2270,8 +2386,21 @@ async def begin_transaction(self) -> "PostgresTransaction": pool = await self._ensure_pool() conn = await pool.acquire() txn = conn.transaction() - await txn.start() - return PostgresTransaction(self, conn, txn) + try: + await txn.start() + tenant_id = _active_tenant.get() + if tenant_id is not None: + await conn.execute( + "SELECT set_config('app.tenant_id', $1, true)", tenant_id + ) + return PostgresTransaction(self, conn, txn) + except BaseException: + try: + if getattr(txn, "is_active", False): + await txn.rollback() + finally: + await pool.release(conn) + raise async def commit_transaction(self, transaction: "PostgresTransaction") -> None: """Commit ``transaction`` and release its connection back to the pool.""" @@ -2319,7 +2448,7 @@ async def drop_deprecated_indexes(self, deprecated: Dict[str, List[str]]) -> Non # ---- transaction handle ----------------------------------------------------- -class PostgresTransaction: +class PostgresTransaction(Transaction): """Active asyncpg transaction handle. Holds a pooled connection + the asyncpg ``Transaction`` object for the @@ -2340,22 +2469,33 @@ class PostgresTransaction: """ def __init__(self, db: "PostgresDB", connection: Any, transaction: Any) -> None: + import uuid + + super().__init__(str(uuid.uuid4())) self._db = db self._connection = connection self._transaction = transaction self.is_active = True self.is_committed = False self.is_rolled_back = False - # Surface a unique id so observability layers can correlate. - import uuid + self._operation_lock = _TaskReentrantLock() + self._bootstrapped_collections: Set[str] = set() + self._closing = False - self.transaction_id = str(uuid.uuid4()) + def _ensure_active(self) -> None: + if self._closing or not self.is_active or self._connection is None: + raise DatabaseError("PostgreSQL transaction is no longer active") + @_serialized_transaction_operation async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: """Upsert ``data`` into ``collection`` within this transaction.""" col = _safe_collection(collection) schema = _safe_collection(self._db.schema_name) - await self._db._bootstrap_collection(collection) + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) rec_id, entity, tenant, data_json = self._db._split_payload(data) await self._connection.execute( f""" @@ -2374,6 +2514,7 @@ async def save(self, collection: str, data: Dict[str, Any]) -> Dict[str, Any]: ) return data + @_serialized_transaction_operation async def insert_if_absent( self, collection: str, @@ -2385,7 +2526,11 @@ async def insert_if_absent( self._db._validate_insert_if_absent(data, conflict_target) col = _safe_collection(collection) schema = _safe_collection(self._db.schema_name) - await self._db._bootstrap_collection(collection) + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) rec_id, entity, tenant, data_json = self._db._split_payload(data) row = await self._connection.fetchrow( f""" @@ -2415,21 +2560,31 @@ async def insert_if_absent( record=self._db._record_from_row(existing), created=False ) + @_serialized_transaction_operation async def get(self, collection: str, id: str) -> Optional[Dict[str, Any]]: """Fetch a single record by ``id`` from ``collection`` in this transaction.""" col = _safe_collection(collection) schema = _safe_collection(self._db.schema_name) - await self._db._bootstrap_collection(collection) + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) row = await self._connection.fetchrow( f"SELECT data FROM {schema}.{col} WHERE id = $1", str(id) ) return self._db._record_from_row(row) if row is not None else None + @_serialized_transaction_operation async def delete(self, collection: str, id: str) -> bool: """Delete record ``id`` from ``collection`` in this transaction.""" col = _safe_collection(collection) schema = _safe_collection(self._db.schema_name) - await self._db._bootstrap_collection(collection) + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) result = await self._connection.execute( f"DELETE FROM {schema}.{col} WHERE id = $1", str(id) ) @@ -2440,6 +2595,7 @@ async def delete(self, collection: str, id: str) -> bool: n = 0 return n > 0 + @_serialized_transaction_operation async def find( self, collection: str, @@ -2449,14 +2605,34 @@ async def find( sort: Optional[List[Tuple[str, int]]] = None, ) -> List[Dict[str, Any]]: """Run ``query`` against ``collection`` within this transaction.""" + if limit == 0: + return [] col = _safe_collection(collection) schema = _safe_collection(self._db.schema_name) - await self._db._bootstrap_collection(collection) + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) translated = translate_query(query) if query else ("", []) if translated is None: - rows = await self._connection.fetch(f"SELECT data FROM {schema}.{col}") - records = [self._db._record_from_row(r) for r in rows] - records = [r for r in records if QueryEngine.match(r, query)] + records = [] + sort_sql = translate_sort(sort) + order_clause = f" ORDER BY {sort_sql}" if sort_sql else "" + cursor = self._connection.cursor( + f"SELECT data FROM {schema}.{col}{order_clause}" + ) + async for row in cursor: + record = self._db._record_from_row(row) + if not QueryEngine.match(record, query): + continue + records.append(record) + if ( + limit is not None + and len(records) >= limit + and (not sort or sort_sql) + ): + break return finalize_find_results(records, sort=sort, limit=limit) where_sql, params = translated sort_sql = translate_sort(sort) @@ -2480,6 +2656,91 @@ async def find( records = finalize_find_results(records, sort=sort, limit=limit) return records + @_serialized_transaction_operation + async def find_one( + self, collection: str, query: Dict[str, Any] + ) -> Optional[Dict[str, Any]]: + """Return one match using this transaction's connection and snapshot.""" + rows = await self.find(collection, query, limit=1) + return rows[0] if rows else None + + @_serialized_transaction_operation + async def find_ids( + self, + collection: str, + query: Dict[str, Any], + *, + limit: Optional[int] = None, + sort: Optional[List[Tuple[str, int]]] = None, + offset: int = 0, + ) -> List[str]: + """Project IDs for paging without leaving the transaction.""" + if offset < 0 or (limit is not None and limit < 0): + raise ValueError("offset and limit must be nonnegative") + if limit == 0: + return [] + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) + ids = await self._db._find_ids_with_connection( + self._connection, + collection, + query, + limit=limit, + sort=sort, + offset=offset, + ) + if ids is None: + return await super().find_ids( + collection, query, limit=limit, sort=sort, offset=offset + ) + return ids + + @_serialized_transaction_operation + async def count( + self, collection: str, query: Optional[Dict[str, Any]] = None + ) -> int: + """Count rows in the same PostgreSQL transaction snapshot.""" + q = query or {} + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) + result = await self._db._count_with_connection(self._connection, collection, q) + if result is not None: + return result + + # Unsupported operators retain QueryEngine semantics without + # materializing the collection in memory. + total = 0 + col = _safe_collection(collection) + schema = _safe_collection(self._db.schema_name) + cursor = self._connection.cursor(f"SELECT data FROM {schema}.{col}") + async for row in cursor: + record = self._db._record_from_row(row) + total += bool(QueryEngine.match(record, q)) + return total + + @_serialized_transaction_operation + async def find_many( + self, collection: str, ids: List[str] + ) -> Dict[str, Dict[str, Any]]: + """Fetch a set of IDs in one query on the held connection.""" + if not ids: + return {} + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) + return await self._db._find_many_with_connection( + self._connection, collection, ids + ) + + @_serialized_transaction_operation async def find_one_and_update( self, collection: str, @@ -2496,7 +2757,11 @@ async def find_one_and_update( """ from .database import _normalize_id_query - await self._db._bootstrap_collection(collection) + await self._db._bootstrap_collection( + collection, + connection=self._connection, + pending=self._bootstrapped_collections, + ) col = _safe_collection(collection) schema = _safe_collection(self._db.schema_name) @@ -2572,23 +2837,34 @@ async def commit(self) -> None: """Commit the wrapped asyncpg transaction. Idempotent.""" if not self.is_active or self.is_committed or self.is_rolled_back: return - try: - await self._transaction.commit() - self.is_committed = True - finally: - self.is_active = False - await self._release() + self._closing = True + async with self._operation_lock: + if not self.is_active or self.is_committed or self.is_rolled_back: + return + try: + await self._transaction.commit() + self.is_committed = True + self._db._collections_bootstrapped.update( + self._bootstrapped_collections + ) + finally: + self.is_active = False + await self._release() async def rollback(self) -> None: """Roll back the wrapped asyncpg transaction. Idempotent.""" if not self.is_active or self.is_committed or self.is_rolled_back: return - try: - await self._transaction.rollback() - self.is_rolled_back = True - finally: - self.is_active = False - await self._release() + self._closing = True + async with self._operation_lock: + if not self.is_active or self.is_committed or self.is_rolled_back: + return + try: + await self._transaction.rollback() + self.is_rolled_back = True + finally: + self.is_active = False + await self._release() async def _release(self) -> None: if self._connection is None: diff --git a/jvspatial/db/transaction.py b/jvspatial/db/transaction.py index f74ae2b..cd7afab 100644 --- a/jvspatial/db/transaction.py +++ b/jvspatial/db/transaction.py @@ -1,10 +1,10 @@ """Transaction support for database operations. -Currently only :class:`MongoDBTransaction` offers native ACID semantics -(requires a replica set, even single-node). Other backends advertise -``Database.supports_transactions = False``; callers should detect that -flag via :func:`transaction_context` and fall back to non-transactional -writes. +PostgreSQL and :class:`MongoDBTransaction` offer native ACID semantics +(MongoDB requires a replica set, even single-node). Backends without native +transactions advertise ``Database.supports_transactions = False``; callers +should detect that flag via :func:`transaction_context` rather than relying +on best-effort writes. JsonDB is a development-only backend with no transaction support — the former ``JsonDBTransaction(best_effort=True)`` buffered-commit mode was @@ -16,7 +16,7 @@ import logging from abc import ABC, abstractmethod from contextlib import asynccontextmanager -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, AsyncIterator, Dict, List, Optional, Tuple logger = logging.getLogger(__name__) @@ -98,6 +98,136 @@ async def find( """ pass + async def find_one( + self, collection: str, query: Dict[str, Any] + ) -> Optional[Dict[str, Any]]: + """Find one matching record without leaving this transaction.""" + rows = await self.find(collection, query, limit=1) + return rows[0] if rows else None + + async def count( + self, collection: str, query: Optional[Dict[str, Any]] = None + ) -> int: + """Count matches in this transaction. + + Backends should override this with a native aggregate when available; + the default keeps correctness and transaction visibility for adapters + whose transaction API only supports ``find``. + """ + return len(await self.find(collection, query or {})) + + async def find_ids( + self, + collection: str, + query: Dict[str, Any], + *, + limit: Optional[int] = None, + sort: Optional[List[Tuple[str, int]]] = None, + offset: int = 0, + ) -> List[str]: + """Return a transaction-consistent ID page. + + Adapters should override this to project IDs natively. The fallback + preserves the Database contract and reads through this transaction. + """ + if offset < 0 or (limit is not None and limit < 0): + raise ValueError("offset and limit must be nonnegative") + if limit == 0: + return [] + from .database import _keyset_sort_fields + + rows = await self.find( + collection, + query, + sort=_keyset_sort_fields(sort), + limit=None if limit is None else offset + limit, + ) + return [str(row["id"]) for row in rows[offset:] if "id" in row] + + async def find_many( + self, collection: str, ids: List[str] + ) -> Dict[str, Dict[str, Any]]: + """Fetch records by ID inside this transaction. + + Adapters should override this when they can perform a set query. + """ + out: Dict[str, Dict[str, Any]] = {} + for record_id in dict.fromkeys(ids): + record = await self.get(collection, record_id) + if record is not None: + out[record_id] = record + return out + + async def find_iter( + self, + collection: str, + query: Dict[str, Any], + *, + sort: Optional[List[Tuple[str, int]]] = None, + batch_size: int = 100, + cursor: Optional[bytes] = None, + ) -> AsyncIterator[Dict[str, Any]]: + """Iterate transaction-visible results using deterministic keysets.""" + if ( + isinstance(batch_size, bool) + or not isinstance(batch_size, int) + or batch_size < 1 + ): + raise ValueError("batch_size must be a positive integer") + from .database import ( + _keyset_condition, + _keyset_payload, + _keyset_sort_fields, + decode_cursor, + ) + + fields = _keyset_sort_fields(sort) + payload = decode_cursor(cursor) + while True: + after = _keyset_condition(fields, payload) + if after is None: + page_query = dict(query) + elif query: + page_query = {"$and": [query, after]} + else: + page_query = after + rows = await self.find( + collection, page_query, sort=fields, limit=batch_size + ) + if not rows: + return + for row in rows: + payload = _keyset_payload(row, fields) + yield row + if len(rows) < batch_size: + return + + async def count_incident_edges( + self, collection: str, node_ids: List[str] + ) -> Dict[str, int]: + """Count requested edge endpoints without escaping this transaction.""" + counts = dict.fromkeys(node_ids, 0) + if not counts: + return counts + iterator = self.find_iter( + collection, + { + "$or": [ + {"source": {"$in": list(counts)}}, + {"target": {"$in": list(counts)}}, + ] + }, + ) + async for edge in iterator: + for node_id in { + value + for value in (edge.get("source"), edge.get("target")) + if isinstance(value, str) + }: + if node_id in counts: + counts[node_id] += 1 + return counts + @abstractmethod async def commit(self) -> None: """Commit this transaction.""" @@ -189,8 +319,8 @@ async def transaction_context(database, transaction_id: Optional[str] = None): """Context manager for database transactions. Yields a ``Transaction`` when the database supports it, or ``None`` - when transactions are unavailable (callers should fall back to - non-transactional writes). + when transactions are unavailable. Callers requiring atomic writes must + reject the unavailable case. Cancellation rolls back just like an error. Args: database: Database instance @@ -211,7 +341,7 @@ async def transaction_context(database, transaction_id: Optional[str] = None): try: yield transaction await database.commit_transaction(transaction) - except Exception: + except BaseException: await database.rollback_transaction(transaction) raise diff --git a/jvspatial/observability/README.md b/jvspatial/observability/README.md index 63b0664..8d58ed6 100644 --- a/jvspatial/observability/README.md +++ b/jvspatial/observability/README.md @@ -8,7 +8,7 @@ Metrics protocol, default no-op recorder, OpenTelemetry adapter. ## Purpose -`observability/` defines a tiny protocol (`MetricsRecorder`) that any metrics backend can implement. The library wires `ObservableDatabase` (internal, in `jvspatial/db/_observable.py`) to emit a structured log line and one metric per DB operation, with WARNING-level elevation past a configurable slow-query threshold. +`observability/` defines a tiny protocol (`MetricsRecorder`) that any metrics backend can implement. `ObservableDatabase` (internal, in `jvspatial/db/_observable.py`) records per-operation metrics and logs failures at ERROR. Successful operations, including slow calls, are silent; slow-call counts remain available through metrics. ## Layout @@ -35,7 +35,7 @@ From `jvspatial.observability.otel` (requires `pip install jvspatial[otel]`): - **The structured log field set from `ObservableDatabase` is public.** Schema: `backend`, `op`, `collection`, `duration_ms`, `success`, `result_count`. Breaking changes require a deprecation cycle. (See [docs/md/stability.md](../../docs/md/stability.md).) - **`NullMetricsRecorder` is zero-overhead.** Calls dispatch to no-op methods; do not add allocations or branching here. -- **Slow-query threshold elevates the log to WARNING.** Configurable per database via `create_database(..., slow_query_ms=N)`. +- **Only failures are logged.** The slow-query threshold is configurable per database via `create_database(..., slow_query_ms=N)` and applies to metrics, not successful-operation logging. - **Four metric names are emitted per DB op**: `jvspatial.db.op.duration_seconds`, `jvspatial.db.op.count`, `jvspatial.db.op.slow_count`, `jvspatial.db.op.result_count`. ## Modification patterns diff --git a/jvspatial/observability/__init__.py b/jvspatial/observability/__init__.py index 203e1eb..6e75d2f 100644 --- a/jvspatial/observability/__init__.py +++ b/jvspatial/observability/__init__.py @@ -7,8 +7,8 @@ The default implementation, :class:`NullMetricsRecorder`, has zero overhead beyond a method call. * :class:`ObservableDatabase` -- a :class:`Database` wrapper (see - ``jvspatial.db._observable``) that emits a structured log line and - one metric per database operation. Opt-in via + ``jvspatial.db._observable``) that records one metric per database + operation and emits a structured warning for slow operations. Opt-in via ``create_database(..., observe=True)``. Both surfaces are public. See ``docs/md/observability.md`` for the diff --git a/jvspatial/version.py b/jvspatial/version.py index 8a87c6c..67e6b74 100644 --- a/jvspatial/version.py +++ b/jvspatial/version.py @@ -9,4 +9,4 @@ # - MAJOR: Breaking changes # - MINOR: New features or breaking changes while pre-1.0 # - PATCH: Bug fixes, backward compatible -__version__ = "0.2.1" +__version__ = "0.2.2" diff --git a/tests/core/test_graph_transaction.py b/tests/core/test_graph_transaction.py index a808903..7ea3663 100644 --- a/tests/core/test_graph_transaction.py +++ b/tests/core/test_graph_transaction.py @@ -180,6 +180,39 @@ async def test_graph_transaction_rollback_does_not_leak_cached_mutation() -> Non end_request_identity_map(identity_token) +@pytest.mark.asyncio +async def test_graph_transaction_rehydrates_parent_cached_entities_on_its_handle() -> ( + None +): + """Preloaded graph objects must not retain the outer database binding.""" + db = _StoreDB() + parent = GraphContext(cast(Database, db)) + async with scoped_default_context_async(parent): + identity_token = begin_request_identity_map() + try: + original = await Folder.create(name="before") + preloaded = await Folder.get(original.id) + assert preloaded is original + + async with graph_transaction(db) as transaction_context: + transactional = await Folder.get(original.id) + assert transactional is not preloaded + assert await transactional.get_context() is transaction_context + transactional.name = "after" + await transactional.save() + # The parent object and shared process cache remain untouched + # until commit, so this reads the committed value. + assert preloaded.name == "before" + independent = GraphContext(cast(Database, db)) + assert ( + await independent.get_node(Folder, original.id) + ).name == "before" + + assert (await Folder.get(original.id)).name == "after" + finally: + end_request_identity_map(identity_token) + + @pytest.mark.asyncio async def test_index_setup_is_scoped_to_database_instance(monkeypatch) -> None: monkeypatch.setenv("JVSPATIAL_AUTO_CREATE_INDEXES", "true") @@ -208,3 +241,15 @@ async def test_save_batch_preserves_mixed_collection_input_order(tmp_path): child = Note(title="child") records = [parent, Object(), child] assert await ctx.save_batch(records) == records + + +@pytest.mark.asyncio +async def test_graph_transaction_rejects_unavailable_deployment(): + """A backend returning None must never bind a non-transactional context.""" + db = _FakeDB() + db.begin_transaction = AsyncMock(return_value=None) + with pytest.raises(TransactionUnavailable, match="native transaction"): + async with graph_transaction(db): + pytest.fail("Unprotected graph writes must not be allowed") + assert not db.committed + assert not db.rolled_back diff --git a/tests/db/test_bounded_query_execution.py b/tests/db/test_bounded_query_execution.py index 83bc2b4..470e314 100644 --- a/tests/db/test_bounded_query_execution.py +++ b/tests/db/test_bounded_query_execution.py @@ -188,27 +188,36 @@ def test_partial_regex_retains_flags(): assert pushed == {"$and": [{"context.label": {"$regex": "alpha", "$options": "i"}}]} -async def test_observed_fallback_reports_context_local_work(tmp_path, caplog): - import logging +async def test_observed_fallback_reports_context_local_work(tmp_path): + class Recorder: + def __init__(self): + self.values = [] - db = create_database("sqlite", db_path=tmp_path / "observe.db", observe=True) + def record_duration(self, *_args, **_kwargs): + pass + + def increment_counter(self, *_args, **_kwargs): + pass + + def record_value(self, name, value, /, **labels): + self.values.append((name, value, labels)) + + recorder = Recorder() + db = create_database( + "sqlite", db_path=tmp_path / "observe.db", observe=True, metrics=recorder + ) try: await seed(db) - with caplog.at_level(logging.DEBUG, logger="jvspatial.db.observable"): - await db.find( - "object", - {"context.group": 1, "context.label": {"$regex": "^match$"}}, - limit=1, - sort=[("id", 1)], - ) - record = next(r for r in caplog.records if getattr(r, "op", None) == "find") - assert record.execution_mode == "residual_scan" - assert record.scanned_rows == 3 - assert record.decoded_bytes > 0 - assert ( - _query_execution.get() is None - or _query_execution.get() is not record.__dict__ + await db.find( + "object", + {"context.group": 1, "context.label": {"$regex": "^match$"}}, + limit=1, + sort=[("id", 1)], ) + observed = {name: value for name, value, _labels in recorder.values} + assert observed["jvspatial.db.query.scanned_rows"] == 3 + assert observed["jvspatial.db.query.decoded_bytes"] > 0 + assert _query_execution.get() is None finally: await db.close() diff --git a/tests/db/test_mongo_async_driver.py b/tests/db/test_mongo_async_driver.py index 7a59a87..3f6b85b 100644 --- a/tests/db/test_mongo_async_driver.py +++ b/tests/db/test_mongo_async_driver.py @@ -60,7 +60,53 @@ async def test_live_driver_queries_bulk_cursor_and_degrees(driver): iterator = db.find_iter("object", {}) await anext(iterator) await iterator.aclose() - assert not await db.is_transactional() # This test service is standalone. + topology = await db._client.admin.command("hello") + expected = bool(topology.get("setName")) or topology.get("msg") == "isdbgrid" + assert await db.is_transactional() is expected + finally: + if db._client is not None: + await db._client.drop_database(db.db_name) + await db.close() + + +@pytest.mark.parametrize("driver", ["motor", "pymongo"]) +async def test_live_transaction_read_contract_commit_and_rollback(driver): + """Qualify shared reads on real sessions when a replica set is configured.""" + from jvspatial.db.transaction import transaction_context + + uri = os.getenv("JVSPATIAL_MONGODB_TEST_URI") + if not uri: + pytest.skip("JVSPATIAL_MONGODB_TEST_URI is not configured") + db = MongoDB( + uri=uri, db_name=f"transaction_audit_{uuid.uuid4().hex}", driver=driver + ) + try: + if not await db.is_transactional(): + pytest.skip("MongoDB transactions require a replica set or sharded cluster") + await db.save("object", {"id": "existing", "rank": 1}) + async with transaction_context(db) as tx: + assert tx is not None + await tx.save("object", {"id": "committed", "rank": 2}) + assert await tx.count("object") == 2 + assert await tx.find_ids("object", {}, sort=[("rank", 1)]) == [ + "existing", + "committed", + ] + assert set(await tx.find_many("object", ["existing", "committed"])) == { + "existing", + "committed", + } + rows = [r async for r in tx.find_iter("object", {}, batch_size=1)] + assert {r["id"] for r in rows} == {"existing", "committed"} + assert await db.get("object", "committed") is None + assert await db.get("object", "committed") is not None + with pytest.raises(RuntimeError, match="abort"): + async with transaction_context(db) as tx: + assert tx is not None + await tx.save("object", {"id": "rolled_back", "rank": 3}) + assert await tx.find_one("object", {"id": "rolled_back"}) is not None + raise RuntimeError("abort") + assert await db.get("object", "rolled_back") is None finally: if db._client is not None: await db._client.drop_database(db.db_name) diff --git a/tests/db/test_observable_database.py b/tests/db/test_observable_database.py index 5770505..4a66858 100644 --- a/tests/db/test_observable_database.py +++ b/tests/db/test_observable_database.py @@ -1,8 +1,8 @@ """Tests for the ObservableDatabase wrapper. Covers: -* structured log line emitted with the standard fields per op, -* slow-query elevation (INFO -> WARNING) at the threshold, +* successful calls, including slow calls, stay silent while failures are logged, +* slow-query metrics still increment at the configured threshold, * metrics recorder gets called with the right labels, * error path emits the log/metric with success=False before re-raising, * factory wires observe=True/slow_query_ms/metrics correctly, @@ -44,86 +44,51 @@ def jsondb() -> Iterator[JsonDB]: yield JsonDB(base_path=tmp) -# ---------------------- structured log line --------------------------- +# ---------------------- routine log suppression ---------------------- -class TestStructuredLog: - async def test_get_emits_log_with_standard_fields(self, jsondb, caplog): - wrapped = ObservableDatabase(jsondb) +class TestRoutineOperationsSilent: + async def test_fast_operations_record_metrics_without_logs(self, jsondb, caplog): + recorder = _CapturingRecorder() + wrapped = ObservableDatabase(jsondb, metrics=recorder) await jsondb.save("node", {"id": "x", "v": 1}) - with caplog.at_level(logging.DEBUG, logger="jvspatial.db.observable"): - await wrapped.get("node", "x") - assert any( - r.name == "jvspatial.db.observable" - and r.levelno == logging.DEBUG - and getattr(r, "op", None) == "get" - and getattr(r, "collection", None) == "node" - and getattr(r, "backend", None) == "JsonDB" - and getattr(r, "success", None) is True - and getattr(r, "result_count", None) == 1 - and getattr(r, "duration_ms", None) is not None - for r in caplog.records - ), [vars(r) for r in caplog.records] - - async def test_find_log_includes_result_count(self, jsondb, caplog): - wrapped = ObservableDatabase(jsondb) - await jsondb.save("node", {"id": "1", "v": 1}) - await jsondb.save("node", {"id": "2", "v": 2}) with caplog.at_level(logging.DEBUG, logger="jvspatial.db.observable"): await wrapped.find("node", {}) - find_records = [r for r in caplog.records if getattr(r, "op", None) == "find"] - assert find_records, "no find log record emitted" - assert getattr(find_records[-1], "result_count", None) == 2 - async def test_count_log_records_count_as_result_count(self, jsondb, caplog): - wrapped = ObservableDatabase(jsondb) - await jsondb.save("node", {"id": "a", "k": "x"}) - await jsondb.save("node", {"id": "b", "k": "x"}) - await jsondb.save("node", {"id": "c", "k": "y"}) - with caplog.at_level(logging.DEBUG, logger="jvspatial.db.observable"): - await wrapped.count("node", {"k": "x"}) - rec = [r for r in caplog.records if getattr(r, "op", None) == "count"] - assert rec - assert getattr(rec[-1], "result_count", None) == 2 + assert not [ + record + for record in caplog.records + if record.name == "jvspatial.db.observable" + ] + assert recorder.durations + assert recorder.counters -# ---------------------- slow query elevation -------------------------- +# ---------------------- slow query silence -------------------------- class TestSlowQuery: - async def test_slow_query_elevates_to_warning(self, jsondb, caplog): - # Force every op to count as slow by using a 0ms threshold. + async def test_slow_query_stays_silent(self, jsondb, caplog): + # Slow-call telemetry belongs in metrics, not application logs. wrapped = ObservableDatabase(jsondb, slow_query_ms=0.0) await jsondb.save("node", {"id": "x", "v": 1}) - with caplog.at_level(logging.WARNING, logger="jvspatial.db.observable"): + with caplog.at_level(logging.DEBUG, logger="jvspatial.db.observable"): await wrapped.get("node", "x") - warns = [ - r - for r in caplog.records - if r.levelno == logging.WARNING and r.name == "jvspatial.db.observable" - ] - assert warns, "expected a WARNING-level slow query log" - assert "SLOW" in warns[-1].message + records = [r for r in caplog.records if r.name == "jvspatial.db.observable"] + assert records == [] - @pytest.mark.parametrize("operation", ["get", "find"]) - async def test_fast_query_is_silent_at_info_with_metrics_intact( - self, jsondb, caplog, operation - ): + async def test_fast_query_stays_silent(self, jsondb, caplog): # Threshold high enough that nothing in this test triggers it. - rec = _CapturingRecorder() - wrapped = ObservableDatabase(jsondb, slow_query_ms=10_000.0, metrics=rec) + wrapped = ObservableDatabase(jsondb, slow_query_ms=10_000.0) await jsondb.save("node", {"id": "x", "v": 1}) with caplog.at_level(logging.INFO, logger="jvspatial.db.observable"): - argument = "x" if operation == "get" else {} - await getattr(wrapped, operation)("node", argument) + await wrapped.get("node", "x") warns = [ r for r in caplog.records if r.levelno == logging.WARNING and r.name == "jvspatial.db.observable" ] assert warns == [] - assert not [r for r in caplog.records if r.name == "jvspatial.db.observable"] - assert rec.durations and rec.counters # ---------------------- metrics --------------------------------------- @@ -206,7 +171,7 @@ async def _exploding_save(collection, data): # Monkey-patch the inner backend's save to raise. jsondb.save = _exploding_save # type: ignore[assignment] - with caplog.at_level(logging.WARNING, logger="jvspatial.db.observable"): + with caplog.at_level(logging.INFO, logger="jvspatial.db.observable"): with pytest.raises(_Bomb): await wrapped.save("node", {"id": "x"}) @@ -214,8 +179,7 @@ async def _exploding_save(collection, data): save_recs = [r for r in caplog.records if getattr(r, "op", None) == "save"] assert save_recs assert getattr(save_recs[-1], "success", None) is False - assert save_recs[-1].levelno == logging.WARNING - assert "FAILED" in save_recs[-1].message + assert save_recs[-1].levelno == logging.ERROR # Metrics also tagged success=False. assert any(d[2].get("success") is False for d in rec.durations) @@ -244,9 +208,8 @@ async def test_observe_false_no_wrap(self, tmp_path): async def test_compose_cache_then_observe(self, tmp_path): """Factory applies cache first, observability outside. - The structured log should report user-visible latency including - the cache. Verified by checking that two reads of the same id - produce two log lines AND only one underlying backend get.""" + Metrics should include user-visible latency across the cache wrapper. + Routine successful reads must remain silent.""" db = create_database( "json", base_path=str(tmp_path), diff --git a/tests/db/test_postgres_integration.py b/tests/db/test_postgres_integration.py index 2b1928c..fbaab19 100644 --- a/tests/db/test_postgres_integration.py +++ b/tests/db/test_postgres_integration.py @@ -374,6 +374,71 @@ async def test_find_one_and_update_upsert(self, pg_db: "PostgresDB") -> None: class TestPostgresTransactionAtomicOps: """CAS + insert_if_absent must share one public transaction handle.""" + async def test_transaction_query_methods_see_uncommitted_state( + self, pg_db: "PostgresDB" + ) -> None: + await pg_db.save( + "o", + {"id": "o.QueryPage.a", "entity": "QueryPage", "context": {"n": 1}}, + ) + txn = await pg_db.begin_transaction() + await txn.save( + "o", + {"id": "o.QueryPage.b", "entity": "QueryPage", "context": {"n": 2}}, + ) + try: + assert await txn.find_one("o", {"id": "o.QueryPage.b"}) == { + "id": "o.QueryPage.b", + "entity": "QueryPage", + "context": {"n": 2}, + } + assert await txn.count("o", {"entity": "QueryPage"}) == 2 + assert await txn.find_ids( + "o", {"entity": "QueryPage"}, sort=[("id", 1)], limit=1, offset=1 + ) == ["o.QueryPage.b"] + assert await txn.find_many("o", ["o.QueryPage.b", "missing"]) == { + "o.QueryPage.b": { + "id": "o.QueryPage.b", + "entity": "QueryPage", + "context": {"n": 2}, + } + } + assert await txn.delete("o", "o.QueryPage.a") is True + assert await txn.count("o", {"entity": "QueryPage"}) == 1 + finally: + await pg_db.rollback_transaction(txn) + + assert await pg_db.get("o", "o.QueryPage.b") is None + assert await pg_db.get("o", "o.QueryPage.a") is not None + + async def test_transaction_bootstrap_is_connection_local_and_rollback_safe( + self, pg_db: "PostgresDB" + ) -> None: + from jvspatial.db.postgres import PostgresDB + + db = PostgresDB( + dsn=pg_db.dsn, + schema_name=pg_db.schema_name, + min_size=1, + max_size=1, + ) + try: + txn = await db.begin_transaction() + await asyncio.wait_for( + txn.save( + "cold_collection", + {"id": "cold.1", "entity": "Cold", "context": {}}, + ), + timeout=5, + ) + await txn.rollback() + assert "cold_collection" not in db._collections_bootstrapped + assert ( + await asyncio.wait_for(db.find("cold_collection", {}), timeout=5) == [] + ) + finally: + await db.close() + async def test_transaction_find_one_and_update_inc( self, pg_db: "PostgresDB" ) -> None: @@ -711,6 +776,31 @@ async def test_no_tenant_scope_sees_nothing_when_required( # Without ``tenant(...)`` the GUC is empty; required-mode rejects all. assert await pg_db_rls.find("node", {}) == [] + async def test_transaction_sets_and_enforces_tenant_scope( + self, pg_db_rls: "PostgresDB" + ) -> None: + await pg_db_rls.enable_rls("node") + async with pg_db_rls.tenant("acme"): + txn = await pg_db_rls.begin_transaction() + await txn.save( + "node", + { + "id": "n.txn.acme", + "entity": "Node", + "tenant_id": "acme", + "context": {"name": "acme"}, + }, + ) + await txn.commit() + + async with pg_db_rls.tenant("beta"): + txn = await pg_db_rls.begin_transaction() + try: + assert await txn.get("node", "n.txn.acme") is None + assert await txn.find("node", {}) == [] + finally: + await txn.rollback() + # ---- pgvector + hybrid queries -------------------------------------------- diff --git a/tests/db/test_postgres_transaction_reads.py b/tests/db/test_postgres_transaction_reads.py new file mode 100644 index 0000000..3b5e417 --- /dev/null +++ b/tests/db/test_postgres_transaction_reads.py @@ -0,0 +1,97 @@ +"""Transaction query methods must retain Postgres semantics and one connection.""" + +from __future__ import annotations + +import asyncio +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +pytest.importorskip("asyncpg") + +from jvspatial.db.database import DatabaseError +from jvspatial.db.postgres import PostgresTransaction + + +def _transaction(connection: Any) -> PostgresTransaction: + db = MagicMock() + db.schema_name = "test_schema" + db._bootstrap_collection = AsyncMock() + db._record_from_row = lambda row: row["data"] + db._collections_bootstrapped = set() + db._collection_bootstrap_locks = {} + db._pool = None + db._find_ids_with_connection = AsyncMock(return_value=["n.entry.1"]) + db._count_with_connection = AsyncMock(return_value=1) + db._find_many_with_connection = AsyncMock( + return_value={"n.entry.1": {"id": "n.entry.1"}} + ) + transaction = MagicMock() + transaction.commit = AsyncMock() + transaction.rollback = AsyncMock() + return PostgresTransaction(db, connection, transaction=transaction) + + +@pytest.mark.asyncio +async def test_read_methods_use_transaction_connection_and_share_query_contract(): + class Connection: + statements = [] + + async def fetch(self, sql, *params): + self.statements.append((sql, params)) + if sql.startswith("SELECT id FROM"): + return [{"id": "n.entry.1"}] + return [{"id": "n.entry.1", "data": {"id": "n.entry.1"}}] + + async def fetchrow(self, sql, *params): + self.statements.append((sql, params)) + if "COUNT(*)" in sql: + return {"n": 1} + return {"data": {"id": "n.entry.1"}} + + conn = Connection() + tx = _transaction(conn) + + assert await tx.find_one("node", {"id": "n.entry.1"}) == {"id": "n.entry.1"} + assert await tx.find_ids("node", {}, limit=1, offset=0) == ["n.entry.1"] + assert await tx.count("node", {"entity": "Entry"}) == 1 + assert await tx.find_many("node", ["n.entry.1", "n.entry.1"]) == { + "n.entry.1": {"id": "n.entry.1"} + } + assert len(conn.statements) == 1 + assert "test_schema." in conn.statements[0][0] + assert tx._db._bootstrap_collection.await_count == 4 + assert all( + call.kwargs["connection"] is conn + for call in tx._db._bootstrap_collection.await_args_list + ) + assert tx._db._find_ids_with_connection.await_args.kwargs["limit"] == 1 + assert tx._db._count_with_connection.await_args.args[0] is conn + assert tx._db._find_many_with_connection.await_args.args[0] is conn + + +@pytest.mark.asyncio +async def test_transaction_serializes_fanout_and_refuses_reads_after_close(): + class Connection: + active = 0 + maximum = 0 + + async def fetchrow(self, *_args, **_kwargs): + self.active += 1 + self.maximum = max(self.maximum, self.active) + try: + await asyncio.sleep(0) + return {"data": {"id": "same"}} + finally: + self.active -= 1 + + connection = Connection() + tx = _transaction(connection) + results = await asyncio.gather(*(tx.get("object", "same") for _ in range(12))) + assert results == [{"id": "same"}] * 12 + assert connection.maximum == 1 + + await tx.rollback() + with pytest.raises(DatabaseError, match="no longer active"): + await tx.get("object", "same") diff --git a/tests/db/test_transaction_api.py b/tests/db/test_transaction_api.py new file mode 100644 index 0000000..dfcc11e --- /dev/null +++ b/tests/db/test_transaction_api.py @@ -0,0 +1,123 @@ +"""Shared transaction APIs retain reads on adapter transaction handles.""" + +from typing import Any + +import pytest + +from jvspatial.db.transaction import MongoDBTransaction + + +class _Cursor: + def __init__(self, rows): + self.rows = rows + + def sort(self, _sort): + return self + + def limit(self, limit): + self.rows = self.rows[:limit] + return self + + async def to_list(self, length=None): + return self.rows if length is None else self.rows[:length] + + +class _Collection: + def __init__(self): + self.rows = { + "one": {"_id": "one", "id": "one", "rank": 1}, + "two": {"_id": "two", "id": "two", "rank": 2}, + } + self.calls = [] + + def find(self, query, *, session): + self.calls.append(("find", query, session)) + rows = list(self.rows.values()) + for key, value in query.items(): + if key.startswith("$"): + continue + if isinstance(value, dict) and "$in" in value: + rows = [row for row in rows if row.get(key) in value["$in"]] + else: + rows = [row for row in rows if row.get(key) == value] + return _Cursor(rows) + + async def find_one(self, query, *, session): + self.calls.append(("find_one", query, session)) + return self.rows.get(query.get("_id")) + + +class _DB: + def __init__(self, collection): + self.collection = collection + + def __getitem__(self, _name): + return self.collection + + +@pytest.mark.asyncio +async def test_mongo_transaction_inherits_consistent_read_contract(): + session: Any = object() + collection = _Collection() + tx = MongoDBTransaction("txn", session, _DB(collection)) + + assert await tx.find_one("node", {"id": "one"}) == collection.rows["one"] + assert await tx.count("node") == 2 + before = len(collection.calls) + assert await tx.find_ids("node", {}, limit=0) == [] + assert len(collection.calls) == before + assert await tx.find_ids("node", {}, limit=1, sort=[("rank", 1)]) == ["one"] + assert await tx.find_many("node", ["two", "one", "two"]) == { + "two": collection.rows["two"], + "one": collection.rows["one"], + } + assert collection.calls + assert all(call[-1] is session for call in collection.calls) + + +@pytest.mark.asyncio +async def test_transaction_context_rolls_back_on_cancellation(): + """Cancellation aborts the native handle rather than leaving it open.""" + import asyncio + from types import SimpleNamespace + from unittest.mock import AsyncMock + + from jvspatial.db.transaction import transaction_context + + tx = object() + db = SimpleNamespace( + begin_transaction=AsyncMock(return_value=tx), + commit_transaction=AsyncMock(), + rollback_transaction=AsyncMock(), + ) + with pytest.raises(asyncio.CancelledError): + async with transaction_context(db) as opened: + assert opened is tx + raise asyncio.CancelledError + db.commit_transaction.assert_not_awaited() + db.rollback_transaction.assert_awaited_once_with(tx) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("action", ["commit", "rollback"]) +@pytest.mark.parametrize("async_end", [True, False]) +async def test_mongo_transaction_ends_session_for_both_driver_shapes(action, async_end): + """Both Motor and PyMongo asynchronous sessions actually finish cleanup.""" + from types import SimpleNamespace + from unittest.mock import AsyncMock, Mock + + from jvspatial.db.mongodb import MongoDB + + end = AsyncMock() if async_end else Mock() + tx = SimpleNamespace( + commit=AsyncMock(), + rollback=AsyncMock(), + session=SimpleNamespace(end_session=end), + ) + db = object.__new__(MongoDB) + await getattr(db, f"{action}_transaction")(tx) + getattr(tx, action).assert_awaited_once() + if async_end: + end.assert_awaited_once() + else: + end.assert_called_once()