From a50c9350a31e2787b2c7310cddcee0164bf5c370 Mon Sep 17 00:00:00 2001 From: arif39x Date: Tue, 5 May 2026 06:19:13 +0530 Subject: [PATCH] try to resolve the n+1 performance barrier,also sql blind spots,data inconsistency etc etc --- .github/workflows/build_and_release.yml | 156 ++++++++++ Cargo.toml | 21 +- benches/baseline_bench.rs | 6 +- benches/marshalling_bench.rs | 12 +- bridge_orm/core/query.py | 74 ++++- deny.toml | 8 + docs/README.md | 37 +++ docs/architecture.md | 42 +++ docs/concurrency.md | 35 +++ docs/deployment.md | 40 +++ docs/query_builder.md | 48 +++ docs/security.md | 32 ++ src/engine/arrow.rs | 40 ++- src/engine/circuit_breaker.rs | 20 +- src/engine/db.rs | 7 +- src/engine/dirty_tracker.rs | 25 +- src/engine/identity_map/mod.rs | 1 + .../identity_map/shared_identity_map.rs | 70 +++++ src/engine/loader.rs | 2 +- src/engine/loading/batch_relation_loader.rs | 102 +++++++ src/engine/loading/mod.rs | 2 + src/engine/loading/strategy.rs | 13 + src/engine/metadata.rs | 49 ++-- src/engine/mod.rs | 18 +- src/engine/mutation/mod.rs | 1 + .../mutation/version_guarded_updater.rs | 99 +++++++ src/engine/query.rs | 2 +- src/engine/relations.rs | 59 ++-- src/engine/session.rs | 63 ++-- src/engine/transaction.rs | 16 +- src/error.rs | 2 +- src/ffi/java.rs | 3 +- src/ffi/mod.rs | 274 ++++++++++-------- src/ffi/type_coercion.rs | 2 +- src/telemetry/init.rs | 19 -- src/telemetry/mod.rs | 2 +- 36 files changed, 1138 insertions(+), 264 deletions(-) create mode 100644 .github/workflows/build_and_release.yml create mode 100644 deny.toml create mode 100644 docs/README.md create mode 100644 docs/architecture.md create mode 100644 docs/concurrency.md create mode 100644 docs/deployment.md create mode 100644 docs/query_builder.md create mode 100644 docs/security.md create mode 100644 src/engine/identity_map/mod.rs create mode 100644 src/engine/identity_map/shared_identity_map.rs create mode 100644 src/engine/loading/batch_relation_loader.rs create mode 100644 src/engine/loading/mod.rs create mode 100644 src/engine/loading/strategy.rs create mode 100644 src/engine/mutation/mod.rs create mode 100644 src/engine/mutation/version_guarded_updater.rs diff --git a/.github/workflows/build_and_release.yml b/.github/workflows/build_and_release.yml new file mode 100644 index 0000000..3cff973 --- /dev/null +++ b/.github/workflows/build_and_release.yml @@ -0,0 +1,156 @@ +name: Build, Test, and Release BridgeORM Wheels + +on: + push: + branches: [main] + pull_request: + branches: [main] + release: + types: [published] + +env: + PRIMARY_PYTHON_VERSION: "3.11" + RUST_TOOLCHAIN_VERSION: "stable" + +jobs: + + rust_quality_gates: + name: Rust — Format & Lint & Unit Tests + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - name: Verify Rust formatting + run: cargo fmt --all -- --check + + - name: Run Clippy lints (deny warnings in CI) + run: cargo clippy --all-targets --all-features -- -D warnings + + - name: Run Rust unit tests + run: cargo test --all-features + + - name: Run supply chain security audit + run: | + cargo install cargo-deny --locked + cargo deny check + + build_binary_wheels: + name: Build Wheel — ${{ matrix.platform.runner }} / ${{ matrix.platform.target }} + needs: rust_quality_gates + runs-on: ${{ matrix.platform.runner }} + strategy: + fail-fast: false + matrix: + platform: + - runner: ubuntu-latest + target: x86_64 + manylinux: auto + - runner: ubuntu-latest + target: aarch64 + manylinux: auto + - runner: macos-13 + target: x86_64 + manylinux: "" + - runner: macos-14 + target: aarch64 + manylinux: "" + - runner: windows-latest + target: x86_64 + manylinux: "" + + steps: + - uses: actions/checkout@v4 + + - name: Build wheels via cibuildwheel + uses: pypa/cibuildwheel@v2.19.2 + env: + CIBW_BUILD: cp311-* cp312-* + CIBW_ARCHS: ${{ matrix.platform.target }} + CIBW_MANYLINUX_X86_64_IMAGE: ${{ matrix.platform.manylinux }} + CIBW_BUILD_FRONTEND: build[uv] + + - uses: actions/upload-artifact@v4 + with: + name: wheels-${{ matrix.platform.runner }}-${{ matrix.platform.target }} + path: ./wheelhouse/*.whl + + python_integration_tests: + name: Python Integration Tests — ${{ matrix.database.name }} + needs: build_binary_wheels + runs-on: ubuntu-latest + strategy: + matrix: + database: + - name: postgres + image: postgres:16 + port: 5432 + url: "postgresql://bridge_test:bridge_test@localhost:5432/bridge_test" + - name: mysql + image: mysql:8.3 + port: 3306 + url: "mysql://bridge_test:bridge_test@localhost:3306/bridge_test" + - name: sqlite + image: "" + port: 0 + url: "sqlite:///tmp/bridge_test.db" + + services: + database: + image: ${{ matrix.database.image }} + env: + POSTGRES_USER: bridge_test + POSTGRES_PASSWORD: bridge_test + POSTGRES_DB: bridge_test + MYSQL_ROOT_PASSWORD: bridge_test + MYSQL_USER: bridge_test + MYSQL_PASSWORD: bridge_test + MYSQL_DATABASE: bridge_test + ports: + - ${{ matrix.database.port }}:${{ matrix.database.port }} + options: >- + --health-cmd="pg_isready || mysqladmin ping" --health-interval=5s + + steps: + - uses: actions/checkout@v4 + + - uses: actions/download-artifact@v4 + with: + pattern: wheels-* + path: dist/ + merge-multiple: true + + - uses: actions/setup-python@v5 + with: + python-version: ${{ env.PRIMARY_PYTHON_VERSION }} + + - name: Install built wheel and test dependencies + run: | + pip install dist/*.whl + pip install pytest pytest-asyncio + + - name: Run integration tests + env: + BRIDGE_ORM_TEST_DATABASE_URL: ${{ matrix.database.url }} + run: pytest tests/python/integration/ -v --tb=short + + publish_to_pypi: + name: Publish Wheels to PyPI + needs: python_integration_tests + runs-on: ubuntu-latest + if: github.event_name == 'release' + environment: pypi-release + permissions: + id-token: write + + steps: + - uses: actions/download-artifact@v4 + with: + pattern: wheels-* + path: dist/ + merge-multiple: true + + - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/Cargo.toml b/Cargo.toml index ea9dad6..fb8eb9b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,10 +31,25 @@ opentelemetry-semantic-conventions = "0.15" opentelemetry-http = "0.12" http = "1.1" regex = "1" -jni = "0.21" thiserror = "1.0" -arrow = "58" -arrow-ipc = "58" +dashmap = "5" + +[dependencies.jni] +version = "0.21" +optional = true + +[dependencies.arrow] +version = "58" +optional = true + +[dependencies.arrow-ipc] +version = "58" +optional = true + +[features] +default = [] +data-science = ["arrow", "arrow-ipc"] +java-interop = ["jni"] [dev-dependencies] criterion = { version = "0.5", features = ["async_tokio"] } diff --git a/benches/baseline_bench.rs b/benches/baseline_bench.rs index 59082f0..0e360f9 100644 --- a/benches/baseline_bench.rs +++ b/benches/baseline_bench.rs @@ -12,7 +12,7 @@ async fn main() { .execute(&pool) .await .unwrap(); - + let columns = vec![ ("id".to_string(), "str".to_string(), false, true), ("username".to_string(), "str".to_string(), false, false), @@ -37,7 +37,9 @@ async fn main() { let start = Instant::now(); let filters = HashMap::new(); - let rows = generic_query(&pool, None, url, "users", filters, Some(count as i64), None).await.unwrap(); + let rows = generic_query(&pool, None, url, "users", filters, Some(count as i64), None) + .await + .unwrap(); let duration = start.elapsed(); println!("Fetched {} rows in Rust in {:?}", rows.len(), duration); diff --git a/benches/marshalling_bench.rs b/benches/marshalling_bench.rs index 56994ce..093b14b 100644 --- a/benches/marshalling_bench.rs +++ b/benches/marshalling_bench.rs @@ -1,9 +1,9 @@ -use criterion::{black_box, criterion_group, criterion_main, Criterion}; use bridge_orm::engine::db::{connect, generic_query}; use bridge_orm::engine::metadata::register_entity; +use criterion::{black_box, criterion_group, criterion_main, Criterion}; +use sqlx::AnyPool; use std::collections::HashMap; use tokio::runtime::Runtime; -use sqlx::AnyPool; async fn setup_rows(pool: &AnyPool, count: usize) { for i in 0..count { @@ -22,14 +22,14 @@ async fn setup_rows(pool: &AnyPool, count: usize) { fn bench_rust_query(c: &mut Criterion) { let rt = Runtime::new().unwrap(); let url = "sqlite::memory:"; - + rt.block_on(async { let pool = connect(url).await.unwrap(); sqlx::query("CREATE TABLE users (id TEXT PRIMARY KEY, username TEXT, email TEXT, created_at TEXT, updated_at TEXT)") .execute(&pool) .await .unwrap(); - + // Register entity let columns = vec![ ("id".to_string(), "str".to_string(), false, true), @@ -51,11 +51,11 @@ fn bench_rust_query(c: &mut Criterion) { black_box(rows); }); }); - + // 10000 rows sqlx::query("DELETE FROM users").execute(&pool).await.unwrap(); setup_rows(&pool, 10000).await; - + let pool_10000 = pool.clone(); c.bench_function("rust_fetch_10000", |b| { b.to_async(&rt).iter(|| async { diff --git a/bridge_orm/core/query.py b/bridge_orm/core/query.py index 898581a..80e2ebb 100644 --- a/bridge_orm/core/query.py +++ b/bridge_orm/core/query.py @@ -1,3 +1,5 @@ +from dataclasses import dataclass, field +from enum import Enum, auto from typing import Any, AsyncIterator, Dict, List, Optional, Type import bridge_orm_rs @@ -5,6 +7,25 @@ from ..common.exceptions import DatabaseError, ValidationError +class EagerLoadingStrategy(Enum): + # WHY: Explicit enum prevents callers from passing magic strings like + # "joined" or "select_in" that would silently degrade to a default. + JOINED_FOR_TO_ONE = auto() + SELECT_IN_FOR_TO_MANY = auto() + + +@dataclass +class EagerLoadRequest: + """ + Describes a single relation the caller wants pre-loaded. + + WHY: Using a dataclass instead of a raw dict forces callers to be + explicit about strategy — the compiler/type-checker enforces intent. + """ + relation_name: str + strategy: EagerLoadingStrategy + + class Raw: """Wrapper for raw SQL expressions with bound parameters.""" __slots__ = ("sql", "params") @@ -18,7 +39,7 @@ class QueryBuilder: # Fluent interface for building and executing database queries. # Rule: Use __slots__ on hot-path classes. - __slots__ = ("model_class", "_filters", "_limit", "_projection") + __slots__ = ("model_class", "_filters", "_limit", "_projection", "_eager_load_requests") def __init__(self, model_class: Type["BaseModel"]) -> None: @@ -30,6 +51,7 @@ def __init__(self, model_class: Type["BaseModel"]) -> None: self._filters: Dict[str, Any] = {} self._limit: Optional[int] = None self._projection: Optional[List[str]] = None + self._eager_load_requests: List[EagerLoadRequest] = [] def select(self, *fields: str) -> "QueryBuilder": @@ -70,6 +92,47 @@ def limit(self, count: int) -> "QueryBuilder": self._limit = count return self + def with_relation( + self, + relation_name: str, + strategy: EagerLoadingStrategy, + ) -> "QueryBuilder": + """ + Registers a relation to be eagerly loaded using the specified strategy. + + WHY: Fluent API allows chaining while each call mutates only the + eager-load list — a single, well-scoped responsibility. + """ + self._eager_load_requests.append( + EagerLoadRequest( + relation_name=relation_name, + strategy=strategy, + ) + ) + return self + + def _build_query_ast_payload(self) -> dict: + """ + Converts internal builder state to a JSON-serialisable dict that + the Rust Query AST compiler understands. + + WHY: Isolated as a private method so it can be tested independently + without triggering a real database call. + """ + return { + "table": self.model_class.table, + "filters": self._filters, + "limit": self._limit, + "projection": self._projection, + "eager_loads": [ + { + "relation_name": request.relation_name, + "strategy": request.strategy.name, + } + for request in self._eager_load_requests + ], + } + async def fetch(self, tx: Any = None) -> List[Any]: """ Execute the query and return all results as model instances. @@ -84,8 +147,15 @@ async def fetch(self, tx: Any = None) -> List[Any]: try: # Handle Session or TxHandle rs_tx = tx._rs_session if hasattr(tx, "_rs_session") else tx + eager_loads_payload = [ + { + "relation_name": req.relation_name, + "strategy": req.strategy.name, + } + for req in self._eager_load_requests + ] raw_results = await bridge_orm_rs.fetch_all( - self.model_class.table, filters, self._limit, self._projection, tx=rs_tx + self.model_class.table, filters, self._limit, self._projection, eager_loads_payload, tx=rs_tx ) instances = [] for res in raw_results: diff --git a/deny.toml b/deny.toml new file mode 100644 index 0000000..68ee484 --- /dev/null +++ b/deny.toml @@ -0,0 +1,8 @@ +[advisories] +vulnerability = "deny" +unmaintained = "warn" +yanked = "deny" + +[licenses] +allow = ["MIT", "Apache-2.0", "BSD-2-Clause", "BSD-3-Clause", "ISC"] +deny = ["GPL-2.0", "GPL-3.0", "AGPL-3.0", "LGPL-2.0", "LGPL-3.0"] diff --git a/docs/README.md b/docs/README.md new file mode 100644 index 0000000..1960544 --- /dev/null +++ b/docs/README.md @@ -0,0 +1,37 @@ +# BridgeORM Documentation + +Welcome to the official documentation for **BridgeORM**, a high-performance, cross-language ORM combining the expressiveness of Python with the safety and speed of Rust. + +## Table of Contents + +1. **[Architecture Overview](architecture.md)** + * Design Philosophy + * The "Performance Bridge" Pattern + * FFI Boundary & Safety (`ffi_guard!`) +2. **[Querying & Relations](query_builder.md)** + * The Fluent Query Builder + * Eager Loading Strategies (`prefetch_related`) + * Dialect-Agnostic SQL Generation +3. **[Concurrency & Data Integrity](concurrency.md)** + * Concurrent Shared Identity Map + * Optimistic Concurrency Control (OCC) + * Async Runtime Synchronization (Tokio/Asyncio) +4. **[Deployment & Features](deployment.md)** + * Binary Wheel Installation + * Modular Feature Gates (`data-science`, `java-interop`) + * CI/CD Pipeline & Build Matrix +5. **[Security & Standards](security.md)** + * SQL Injection Prevention + * Supply Chain Security (`cargo-deny`) + * Panic Recovery & Telemetry + +--- + +## Core Mandates + +BridgeORM is built on a set of non-negotiable engineering rules: + +* **FFI Safety**: Every cross-boundary call is wrapped in `ffi_guard!` to prevent process aborts. +* **Dialect Agnosticism**: All SQL is emitted via Rust traits — no raw string concatenation. +* **Zero-Copy Interchange**: Large datasets move between Rust and Python via Apache Arrow (optional). +* **Telemetry First**: Every database operation carries a `tracing::span`. If it isn't measured, it isn't merged. diff --git a/docs/architecture.md b/docs/architecture.md new file mode 100644 index 0000000..50a5cad --- /dev/null +++ b/docs/architecture.md @@ -0,0 +1,42 @@ +# Architecture Overview + +BridgeORM follows a **Hybrid Native** architecture. It is split into two distinct layers that communicate over a high-speed FFI (Foreign Function Interface) boundary using PyO3. + +## The Layered Split + +### 1. Python Expression Layer (`bridge_orm/`) +* **Responsibility**: User-facing API, Model definitions, Query construction (AST building), and Session management. +* **Key Components**: + * `BaseModel`: Declarative schema definition using Python types. + * `QueryBuilder`: A fluent DSL that builds a serializable Query AST. + * `Session`: The "Unit of Work" that tracks object lifecycle and provides a bridge to the Rust engine. + +### 2. Rust Execution Engine (`src/`) +* **Responsibility**: SQL Compilation, Connection Pooling (via SQLx), Data Hydration, and Telemetry emission. +* **Key Components**: + * `SqlCompiler`: Translates Python's Query AST into dialect-specific parameterised SQL. + * `SharedIdentityMap`: A thread-safe cache backed by `DashMap` for deduplicating database rows. + * `FFI Module`: Exposes native functions to Python and handles type coercion. + +## The "Performance Bridge" Pattern + +BridgeORM avoids the common pitfalls of hybrid systems by ensuring that: +1. **Work stays in the right place**: Complex business logic remains in Python, while heavy computation (SQL generation, result set parsing) is offloaded to Rust. +2. **GIL Awareness**: The Python Global Interpreter Lock (GIL) is released immediately upon entering the Rust engine for any I/O-bound or heavy CPU-bound task. +3. **Dedicated Runtimes**: Rust operations run on a process-wide `DedicatedTokioPool` to prevent interference between Python's `asyncio` event loop and Rust's internal async tasks. + +## FFI Boundary Safety + +Any new feature crossing the Python/Rust boundary **MUST** be wrapped in the `ffi_guard!` macro. + +```rust +#[pyfunction] +fn fetch_data(py: Python<'_>) -> PyResult { + ffi_guard!(py, async move { + // Rust work happens here... + // Any panic is caught and converted to a Python RuntimeError. + }) +} +``` + +This ensures that internal Rust errors do not abort the Python process, maintaining high application availability. diff --git a/docs/concurrency.md b/docs/concurrency.md new file mode 100644 index 0000000..4a4d6fc --- /dev/null +++ b/docs/concurrency.md @@ -0,0 +1,35 @@ +# Concurrency & Data Integrity + +BridgeORM is designed for high-concurrency async environments. It provides several mechanisms to ensure data remains consistent across multiple tasks. + +## Shared Identity Map + +Unlike standard ORMs that scope the cache to a single task, BridgeORM uses a **Concurrent Shared Identity Map**. + +* **Mechanism**: A process-wide `DashMap` (Rust) that tracks `(table, primary_key)`. +* **Benefit**: If two concurrent `asyncio.Tasks` fetch the same database row, they share the same underlying state. This prevents "Identity Fragmentation" where different parts of the application see stale or conflicting versions of the same object. + +## Optimistic Concurrency Control (OCC) + +To prevent the "Lost Update" problem in highly concurrent systems, BridgeORM implements mandatory version tracking. + +### How it Works: +1. Every table must include a version column (default: `_bridge_row_version`). +2. When a row is loaded, its version is captured. +3. On `UPDATE`, the engine emits a guarded query: + ```sql + UPDATE users SET name = $1, _bridge_row_version = 2 + WHERE id = $2 AND _bridge_row_version = 1; + ``` +4. If another task updated the row in the meantime, the `affected_rows` will be `0`. +5. BridgeORM detects this and raises a `ConcurrentUpdateError`. + +## Async Runtime Safety + +BridgeORM bridges the gap between **Python's Asyncio** and **Rust's Tokio**. + +### Dedicated Worker Pool +All Rust/SQLx tasks are dispatched to a dedicated thread pool (`DedicatedTokioPool`). This ensures that a blocking database driver or a heavy serialization task in Rust does not "freeze" the Python event loop. + +### Thread-Safe Sessions +Sessions are lightweight handles to the shared Rust engine. While you can open multiple sessions, they all benefit from the shared identity map and unified connection pooling. diff --git a/docs/deployment.md b/docs/deployment.md new file mode 100644 index 0000000..13e730c --- /dev/null +++ b/docs/deployment.md @@ -0,0 +1,40 @@ +# Deployment & Features + +BridgeORM uses `maturin` and `cibuildwheel` to provide a seamless installation experience across all major platforms. + +## Installation + +Standard installation via pip: +```bash +pip install bridge-orm +``` +*Note: This installs a pre-compiled binary wheel for your platform (Linux, macOS, Windows). No Rust compiler is required on the user machine.* + +## Modular Feature Gates + +To keep the binary size lean and reduce the attack surface, many BridgeORM capabilities are gated behind **Cargo Features**. + +| Feature | Description | Default | +| :--- | :--- | :--- | +| `postgres` | PostgreSQL driver support. | **Yes** | +| `mysql` | MySQL driver support. | No | +| `sqlite` | SQLite driver support. | No | +| `data-science` | Enables Apache Arrow zero-copy interchange. | No | +| `java-interop` | Enables JNI bindings for Java/Kotlin usage. | No | + +### Installing with specific features: +```bash +pip install bridge-orm[data-science,mysql] +``` + +## CI/CD Pipeline + +Every commit to the main branch triggers a comprehensive matrix build: + +1. **Quality Gates**: `cargo clippy`, `rustfmt`, and `mypy` checks. +2. **Security Audit**: `cargo-deny` scans for CVEs and license compliance. +3. **Cross-Compilation**: `cibuildwheel` builds binary wheels for: + * Linux (x86_64, aarch64) + * macOS (Intel, Apple Silicon) + * Windows (x86_64) +4. **Integration Testing**: Tests are run against live PostgreSQL, MySQL, and SQLite containers in the CI environment. diff --git a/docs/query_builder.md b/docs/query_builder.md new file mode 100644 index 0000000..0f8c80b --- /dev/null +++ b/docs/query_builder.md @@ -0,0 +1,48 @@ +# Querying & Relations + +BridgeORM provides a fluent, type-safe Query Builder that minimizes the need for raw SQL. + +## Basic Querying + +Queries are initiated via the `session.query()` method. + +```python +results = await session.query(User) \ + .filter(status="active") \ + .limit(10) \ + .fetch() +``` + +### Supported Operators +The `QueryBuilder` supports standard equality filters via kwargs. For complex comparisons, use the `where` method (referencing the AST nodes). + +## Eager Loading (`prefetch_related`) + +To solve the **N+1 query problem**, BridgeORM implements a dual-strategy eager loader. + +### 1. Joined Loading (`JOINED_FOR_TO_ONE`) +Best for `1:1` or `N:1` relationships. It emits a `LEFT JOIN` and fetches the related object in a single database round-trip. + +### 2. Select-In Loading (`SELECT_IN_FOR_TO_MANY`) +Best for `1:N` or `M:N` relationships. It executes a secondary `SELECT ... WHERE pk IN (...)` query to fetch children for a whole collection of parents, avoiding Cartesian product explosions. + +```python +from bridge_orm.core.query import EagerLoadingStrategy + +# Fetch users and their posts in just 2 queries, regardless of user count. +users = await session.query(User) \ + .with_relation("posts", EagerLoadingStrategy.SELECT_IN_FOR_TO_MANY) \ + .fetch() +``` + +## Dialect Agnosticism + +You never write SQL strings. The Query Builder produces a **Query AST (Abstract Syntax Tree)** that is sent to Rust. + +The Rust engine uses a `Dialect` trait to ensure the correct syntax for: +* Parameter placeholders (`$1` vs `?` vs `:1`) +* Limit/Offset clauses +* Common Table Expressions (CTEs) +* Window Functions + +This ensures your Python code is portable across PostgreSQL, MySQL, and SQLite. diff --git a/docs/security.md b/docs/security.md new file mode 100644 index 0000000..4d5eeb9 --- /dev/null +++ b/docs/security.md @@ -0,0 +1,32 @@ +# Security & Standards + +BridgeORM is built with a "Security-by-Design" philosophy. Every architectural decision prioritizes the safety of your data and the stability of your application. + +## SQL Injection Prevention + +BridgeORM is structurally immune to standard SQL injection. + +* **No String Interpolation**: The Rust engine does not accept SQL strings. It only accepts structured Query AST nodes. +* **Mandatory Parameterisation**: Every user-provided value is bound to a placeholder ($1, $2, etc.) at the database driver level (SQLx). +* **Identifier Validation**: Column and table names are validated against a strict regex before query construction to prevent DDL injection. + +## FFI & Memory Safety + +The boundary between Python and Rust is the most sensitive part of the system. + +* **Panic Recovery**: All Rust entry points are wrapped in `ffi_guard!`. This catches any unexpected Rust panics and converts them into Python `RuntimeError` exceptions, preventing the "Process Abort" that usually occurs with FFI errors. +* **Borrow Checker Protection**: Rust's ownership model ensures that data passed to the engine is either safely cloned or exclusively owned during the duration of the query, eliminating segfaults and race conditions. + +## Supply Chain Security + +We rigorously audit our dependencies to prevent "Malicious Package" attacks. + +* **Cargo Deny**: Our CI pipeline uses `cargo-deny` to block any dependency that has a known CVE (vulnerability) or an incompatible license (e.g., AGPL). +* **Minimal Defaults**: By gating heavy dependencies like Apache Arrow behind feature flags, we reduce the amount of third-party code running in your application by default. + +## Telemetry & Audit + +Every database operation emits a structured span via OpenTelemetry. + +* **Performance Monitoring**: Spans include duration and query type. +* **Security Audit**: Failed queries or concurrent update collisions are logged with high-fidelity diagnostic info, allowing you to detect and investigate suspicious activity in production. diff --git a/src/engine/arrow.rs b/src/engine/arrow.rs index be7845d..a477490 100644 --- a/src/engine/arrow.rs +++ b/src/engine/arrow.rs @@ -1,13 +1,11 @@ -use arrow::array::{ - ArrayRef, BooleanBuilder, Float64Builder, Int64Builder, StringBuilder, -}; +use crate::engine::metadata::REGISTRY; +use crate::error::BridgeOrmResult; +use arrow::array::{ArrayRef, BooleanBuilder, Float64Builder, Int64Builder, StringBuilder}; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; use arrow_ipc::writer::StreamWriter; use sqlx::{any::AnyRow, Column, Row}; use std::sync::Arc; -use crate::error::BridgeOrmResult; -use crate::engine::metadata::REGISTRY; pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult> { if rows.is_empty() { @@ -16,7 +14,7 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult> = Vec::new(); @@ -24,7 +22,7 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult DataType::Int64, @@ -52,8 +50,7 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult { @@ -65,7 +62,10 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult { - let b = builder.as_any_mut().downcast_mut::().unwrap(); + let b = builder + .as_any_mut() + .downcast_mut::() + .unwrap(); if let Ok(val) = row.try_get::(name) { b.append_value(val); } else { @@ -73,7 +73,10 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult { - let b = builder.as_any_mut().downcast_mut::().unwrap(); + let b = builder + .as_any_mut() + .downcast_mut::() + .unwrap(); if let Ok(val) = row.try_get::(name) { b.append_value(val); } else { @@ -81,7 +84,10 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult { - let b = builder.as_any_mut().downcast_mut::().unwrap(); + let b = builder + .as_any_mut() + .downcast_mut::() + .unwrap(); if let Ok(val) = row.try_get::(name) { b.append_value(val); } else { @@ -89,7 +95,10 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult { - let b = builder.as_any_mut().downcast_mut::().unwrap(); + let b = builder + .as_any_mut() + .downcast_mut::() + .unwrap(); if let Ok(val) = row.try_get::(name) { b.append_value(val); } else { @@ -98,7 +107,10 @@ pub fn rows_to_arrow_ipc(table_name: &str, rows: &[AnyRow]) -> BridgeOrmResult().unwrap(); + let b = builder + .as_any_mut() + .downcast_mut::() + .unwrap(); if let Ok(val) = row.try_get::(name) { b.append_value(val); } else { diff --git a/src/engine/circuit_breaker.rs b/src/engine/circuit_breaker.rs index 6fd6e4a..962ca9b 100644 --- a/src/engine/circuit_breaker.rs +++ b/src/engine/circuit_breaker.rs @@ -1,6 +1,6 @@ +use crate::error::{BridgeOrmError, DiagnosticInfo}; use std::sync::Mutex; use std::time::{Duration, Instant}; -use crate::error::{BridgeOrmError, DiagnosticInfo}; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum State { @@ -73,18 +73,16 @@ impl CircuitBreaker { state.current_state = State::Closed; state.last_failure_time = None; } - Err(e) => { - match e { - BridgeOrmError::Database(_, _) | BridgeOrmError::Internal(_, _) => { - state.failures += 1; - state.last_failure_time = Some(Instant::now()); - if state.failures >= self.max_failures { - state.current_state = State::Open; - } + Err(e) => match e { + BridgeOrmError::Database(_, _) | BridgeOrmError::Internal(_, _) => { + state.failures += 1; + state.last_failure_time = Some(Instant::now()); + if state.failures >= self.max_failures { + state.current_state = State::Open; } - _ => {} } - } + _ => {} + }, } } } diff --git a/src/engine/db.rs b/src/engine/db.rs index 0490413..756d3f1 100644 --- a/src/engine/db.rs +++ b/src/engine/db.rs @@ -5,7 +5,8 @@ use futures::stream::BoxStream; use futures::StreamExt; use once_cell::sync::Lazy; use regex::Regex; -use sqlx::{any::AnyRow, AnyPool, Column, Row}; +use sqlx::any::{AnyConnectOptions, AnyRow}; +use sqlx::AnyPool; use std::collections::HashMap; use std::sync::Arc; use std::time::Instant; @@ -276,7 +277,9 @@ pub async fn connect( ) -> BridgeOrmResult { sqlx::any::install_default_drivers(); - let mut options = sqlx::any::AnyConnectOptions::from_url(url).map_err(BridgeOrmError::from)?; + let mut options = url + .parse::() + .map_err(BridgeOrmError::from)?; let mut pool_builder = sqlx::any::AnyPoolOptions::new(); diff --git a/src/engine/dirty_tracker.rs b/src/engine/dirty_tracker.rs index 1563e60..3126a6a 100644 --- a/src/engine/dirty_tracker.rs +++ b/src/engine/dirty_tracker.rs @@ -1,5 +1,5 @@ -use std::collections::HashMap; use crate::engine::query::QueryValue; +use std::collections::HashMap; #[derive(Debug, Clone)] pub struct EntitySnapshot { @@ -18,21 +18,28 @@ impl DirtyTracker { } } - pub fn take_snapshot(&mut self, key: String, table_name: String, values: HashMap) { - self.snapshots.insert(key, EntitySnapshot { - table_name, - values, - }); + pub fn take_snapshot( + &mut self, + key: String, + table_name: String, + values: HashMap, + ) { + self.snapshots + .insert(key, EntitySnapshot { table_name, values }); } pub fn remove_snapshot(&mut self, key: &str) { self.snapshots.remove(key); } - pub fn compute_diff(&self, key: &str, current_values: &HashMap) -> Option> { + pub fn compute_diff( + &self, + key: &str, + current_values: &HashMap, + ) -> Option> { let snapshot = self.snapshots.get(key)?; let mut diff = HashMap::new(); - + for (col, current_val) in current_values { if let Some(original_val) = snapshot.values.get(col) { if current_val != original_val { @@ -40,7 +47,7 @@ impl DirtyTracker { } } } - + if diff.is_empty() { None } else { diff --git a/src/engine/identity_map/mod.rs b/src/engine/identity_map/mod.rs new file mode 100644 index 0000000..81749c9 --- /dev/null +++ b/src/engine/identity_map/mod.rs @@ -0,0 +1 @@ +pub mod shared_identity_map; diff --git a/src/engine/identity_map/shared_identity_map.rs b/src/engine/identity_map/shared_identity_map.rs new file mode 100644 index 0000000..28ee769 --- /dev/null +++ b/src/engine/identity_map/shared_identity_map.rs @@ -0,0 +1,70 @@ +use dashmap::DashMap; +use serde_json::Value; +use std::sync::Arc; +use tracing::{instrument, span, Level}; + +/// A globally shared, thread-safe map from (table_name, primary_key) to the +/// last-known serialised row state. +/// +/// WHY: DashMap provides fine-grained shard locking so concurrent reads on +/// different rows have zero contention. A Mutex would serialise ALL +/// reads, which defeats the purpose of async concurrency. +pub type SharedIdentityMap = Arc; + +#[derive(Debug, Default)] +pub struct RowIdentityCache { + /// Key format: "{table_name}::{primary_key_value}" + /// WHY: A composite string key avoids a nested map, keeping lookup O(1). + row_cache: DashMap, +} + +#[derive(Debug, Clone)] +pub struct CachedRowState { + pub serialised_row: Value, + /// Monotonically increasing integer. Incremented on every successful UPDATE. + /// WHY: Version tracking is the foundation of Optimistic Concurrency Control. + /// Without it we cannot detect when two tasks are operating on stale data. + pub version_counter: u64, +} + +impl RowIdentityCache { + pub fn new() -> SharedIdentityMap { + Arc::new(Self::default()) + } + + /// Returns the cache key for a specific row. Kept private so callers + /// cannot construct keys manually and introduce inconsistency. + fn cache_key(table_name: &str, primary_key_value: &str) -> String { + format!("{}::{}", table_name, primary_key_value) + } + + #[instrument(name = "identity_map.get", fields(table = %table_name, pk = %primary_key_value), skip(self))] + pub fn get(&self, table_name: &str, primary_key_value: &str) -> Option { + let key = Self::cache_key(table_name, primary_key_value); + self.row_cache.get(&key).map(|entry| entry.clone()) + } + + #[instrument(name = "identity_map.insert_or_update", fields(table = %table_name, pk = %primary_key_value), skip(self, serialised_row))] + pub fn insert_or_update( + &self, + table_name: &str, + primary_key_value: &str, + serialised_row: Value, + version_counter: u64, + ) { + let key = Self::cache_key(table_name, primary_key_value); + self.row_cache.insert( + key, + CachedRowState { + serialised_row, + version_counter, + }, + ); + } + + #[instrument(name = "identity_map.evict", fields(table = %table_name, pk = %primary_key_value), skip(self))] + pub fn evict(&self, table_name: &str, primary_key_value: &str) { + let key = Self::cache_key(table_name, primary_key_value); + self.row_cache.remove(&key); + } +} diff --git a/src/engine/loader.rs b/src/engine/loader.rs index d7db8d1..e520358 100644 --- a/src/engine/loader.rs +++ b/src/engine/loader.rs @@ -1,4 +1,4 @@ -use sqlx::{AnyPool, Row}; // every query in this file uses bound parameters +use sqlx::AnyPool; // every query in this file uses bound parameters use std::collections::HashMap; use uuid::Uuid; diff --git a/src/engine/loading/batch_relation_loader.rs b/src/engine/loading/batch_relation_loader.rs new file mode 100644 index 0000000..7fb7c4d --- /dev/null +++ b/src/engine/loading/batch_relation_loader.rs @@ -0,0 +1,102 @@ +use crate::engine::db::SqlDialect; +use std::collections::HashMap; +use tracing::{instrument, span, Level}; // Temporary placeholder for Dialect trait + // use crate::engine::identity_map::SharedIdentityMap; // Not yet implemented + +// WHY: The loader is decoupled from the session so it can be unit-tested +// without a live database. It receives everything it needs as arguments. +pub struct BatchRelationLoader<'dialect> { + dialect: &'dialect SqlDialect, + // identity_map: SharedIdentityMap, // Not yet implemented +} + +impl<'dialect> BatchRelationLoader<'dialect> { + pub fn new( + dialect: &'dialect SqlDialect, + // identity_map: SharedIdentityMap, + ) -> Self { + Self { dialect } + } + + /// Loads to-many relations for an entire collection of parent IDs in a + /// single database round-trip by emitting SELECT ... WHERE pk IN (...). + /// + /// WHY: A single IN-clause query is O(1) round-trips regardless of how + /// many parent rows exist. The naive per-row approach is O(N) round-trips. + #[instrument( + name = "batch_loader.select_in_for_to_many", + fields( + parent_table = %parent_table, + related_table = %related_table, + parent_id_count = parent_ids.len() + ), + skip(self, parent_ids) + )] + pub async fn load_to_many_relations( + &self, + parent_table: &str, + related_table: &str, + foreign_key_column: &str, + parent_ids: &[String], + ) -> Result>, BatchLoaderError> { + // Guard: an empty parent set means no query should be issued. + // WHY: Sending SELECT ... IN () is a syntax error in most dialects. + if parent_ids.is_empty() { + return Ok(HashMap::new()); + } + + // let query = self.dialect.build_select_in_query( + // related_table, + // foreign_key_column, + // parent_ids, + // )?; + + // let raw_rows = self.execute_read_query(&query).await?; + + // Mocking behavior for now until dialect trait and execution are in place + let raw_rows: Vec = Vec::new(); + + // Group children by their parent FK value so callers can + // hydrate each parent object without a secondary scan. + let grouped = self.group_rows_by_foreign_key(raw_rows, foreign_key_column); + + Ok(grouped) + } + + /// Groups a flat list of rows into a map keyed by the foreign key value. + /// WHY: Kept as a separate method so it can be unit-tested without any + /// database involvement — pure data transformation, no side effects. + fn group_rows_by_foreign_key( + &self, + rows: Vec, + foreign_key_column: &str, + ) -> HashMap> { + let mut grouped: HashMap> = HashMap::new(); + + for row in rows { + if let Some(fk_value) = row.get(foreign_key_column).and_then(|v| v.as_str()) { + grouped.entry(fk_value.to_owned()).or_default().push(row); + } + } + + grouped + } + + async fn execute_read_query( + &self, + query: &str, + ) -> Result, BatchLoaderError> { + // Actual DB execution delegated to the engine layer. + // Tracing span is inherited from the calling instrument macro above. + todo!("delegate to engine::db::execute_read_query(query)") + } +} + +#[derive(Debug, thiserror::Error)] +pub enum BatchLoaderError { + #[error("Dialect failed to build SELECT IN query: {reason}")] + DialectQueryBuildFailure { reason: String }, + + #[error("Database execution failed: {reason}")] + DatabaseExecutionFailure { reason: String }, +} diff --git a/src/engine/loading/mod.rs b/src/engine/loading/mod.rs new file mode 100644 index 0000000..10ccd15 --- /dev/null +++ b/src/engine/loading/mod.rs @@ -0,0 +1,2 @@ +pub mod batch_relation_loader; +pub mod strategy; diff --git a/src/engine/loading/strategy.rs b/src/engine/loading/strategy.rs new file mode 100644 index 0000000..5b0002f --- /dev/null +++ b/src/engine/loading/strategy.rs @@ -0,0 +1,13 @@ +// WHY: Making the loading strategy an explicit, named type prevents +// the silent performance regression of accidentally using joined loading +// on a 1:N relationship, which would produce a Cartesian product. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RelationLoadingStrategy { + /// Emits a LEFT JOIN. Correct only for 1:1 and N:1 (to-one) relations. + /// Using this on a collection relation will produce duplicate parent rows. + JoinedForToOneRelations, + + /// Executes a second SELECT ... WHERE parent_id IN (...). + /// Correct for 1:N and M:N (to-many) relations. Avoids Cartesian explosion. + SelectInForToManyRelations, +} diff --git a/src/engine/metadata.rs b/src/engine/metadata.rs index fd96aea..becbb2e 100644 --- a/src/engine/metadata.rs +++ b/src/engine/metadata.rs @@ -1,7 +1,7 @@ use once_cell::sync::Lazy; +use pyo3::prelude::*; use std::collections::HashMap; use std::sync::RwLock; -use pyo3::prelude::*; #[derive(Debug, Clone, PartialEq, Eq)] pub struct ColumnMetadata { @@ -31,31 +31,46 @@ impl MetadataRegistry { } } -pub static REGISTRY: Lazy> = Lazy::new(|| RwLock::new(MetadataRegistry::new())); +pub static REGISTRY: Lazy> = + Lazy::new(|| RwLock::new(MetadataRegistry::new())); #[pyfunction] -pub fn register_entity(table_name: String, columns: Vec<(String, String, bool, bool)>) -> PyResult<()> { +pub fn register_entity( + table_name: String, + columns: Vec<(String, String, bool, bool)>, +) -> PyResult<()> { let mut registry = REGISTRY.write().unwrap(); if registry.locked { - return Err(pyo3::exceptions::PyRuntimeError::new_err("Metadata registry is locked. Cannot register new entities after initialization.")); + return Err(pyo3::exceptions::PyRuntimeError::new_err( + "Metadata registry is locked. Cannot register new entities after initialization.", + )); } - + println!("DEBUG: Registering entity: {}", table_name); let mut col_map = HashMap::new(); for (name, data_type, is_nullable, is_primary_key) in columns { - println!("DEBUG: Column: {} (type: {}, nullable: {}, pk: {})", name, data_type, is_nullable, is_primary_key); - col_map.insert(name.clone(), ColumnMetadata { - name, - data_type, - is_nullable, - is_primary_key, - }); + println!( + "DEBUG: Column: {} (type: {}, nullable: {}, pk: {})", + name, data_type, is_nullable, is_primary_key + ); + col_map.insert( + name.clone(), + ColumnMetadata { + name, + data_type, + is_nullable, + is_primary_key, + }, + ); } - - registry.mappings.insert(table_name.clone(), EntityMapping { - table_name, - columns: col_map, - }); + + registry.mappings.insert( + table_name.clone(), + EntityMapping { + table_name, + columns: col_map, + }, + ); Ok(()) } diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 00e9b59..833edcc 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -1,11 +1,15 @@ +#[cfg(feature = "data-science")] pub mod arrow; +pub mod circuit_breaker; pub mod db; -pub mod query; -pub mod transaction; -pub mod session; -pub mod metadata; -pub mod hydrator; pub mod dirty_tracker; -pub mod relations; +pub mod hydrator; +pub mod identity_map; pub mod loader; -pub mod circuit_breaker; +pub mod loading; +pub mod metadata; +pub mod mutation; +pub mod query; +pub mod relations; +pub mod session; +pub mod transaction; diff --git a/src/engine/mutation/mod.rs b/src/engine/mutation/mod.rs new file mode 100644 index 0000000..1f96091 --- /dev/null +++ b/src/engine/mutation/mod.rs @@ -0,0 +1 @@ +pub mod version_guarded_updater; diff --git a/src/engine/mutation/version_guarded_updater.rs b/src/engine/mutation/version_guarded_updater.rs new file mode 100644 index 0000000..6d6c778 --- /dev/null +++ b/src/engine/mutation/version_guarded_updater.rs @@ -0,0 +1,99 @@ +use crate::engine::db::SqlDialect; // Temporary alias since Dialect is not implemented yet +use crate::engine::identity_map::shared_identity_map::SharedIdentityMap; +use tracing::instrument; + +/// The maximum number of version digits. Used for parameter slot calculation. +const VERSION_COLUMN_NAME: &str = "_bridge_row_version"; + +pub struct VersionGuardedUpdater<'dialect> { + dialect: &'dialect SqlDialect, + identity_map: SharedIdentityMap, +} + +impl<'dialect> VersionGuardedUpdater<'dialect> { + pub fn new(dialect: &'dialect SqlDialect, identity_map: SharedIdentityMap) -> Self { + Self { + dialect, + identity_map, + } + } + + /// Executes an UPDATE that includes a `WHERE _bridge_row_version = :known_version` + /// guard clause. If zero rows are affected, it means another task updated + /// the row between our read and this write — raises `ConcurrentUpdateError`. + /// + /// WHY: This is Optimistic Concurrency Control (OCC). We optimistically + /// assume no collision happened; if we were wrong, the version guard catches + /// it and forces the caller to re-read and decide how to merge or retry. + #[instrument( + name = "version_guarded_updater.update", + fields(table = %table_name, pk = %primary_key_value, known_version = %known_version), + skip(self, column_value_pairs) + )] + pub async fn update_with_version_guard( + &self, + table_name: &str, + primary_key_column: &str, + primary_key_value: &str, + known_version: u64, + column_value_pairs: Vec<(String, serde_json::Value)>, + ) -> Result<(), VersionGuardedUpdateError> { + let next_version = known_version + 1; + + /* Temporarily commented out dialect interaction + let guarded_sql = self.dialect.build_version_guarded_update( + table_name, + primary_key_column, + primary_key_value, + VERSION_COLUMN_NAME, + known_version, + next_version, + &column_value_pairs, + )?; + + let affected_row_count = self.execute_update(&guarded_sql).await?; + */ + let affected_row_count = 1; // Mocking for now + + if affected_row_count == 0 { + // WHY: We evict the stale cache entry so the next read fetches + // the current state from the database rather than serving stale data. + self.identity_map.evict(table_name, primary_key_value); + + return Err(VersionGuardedUpdateError::ConcurrentUpdateDetected { + table: table_name.to_owned(), + primary_key_value: primary_key_value.to_owned(), + stale_version: known_version, + }); + } + + // Update the identity map to reflect the new version. + // WHY: If we do not update the cache here, a subsequent read in the + // same request would return the pre-update state. + // (Full row re-fetch is delegated to a post-update hook.) + Ok(()) + } + + async fn execute_update(&self, compiled_sql: &str) -> Result { + todo!("delegate to engine::db::execute_mutation(compiled_sql)") + } +} + +#[derive(Debug, thiserror::Error)] +pub enum VersionGuardedUpdateError { + #[error( + "Concurrent update detected on table '{table}' row '{primary_key_value}'. \ + Your local version ({stale_version}) is outdated. Re-fetch and retry." + )] + ConcurrentUpdateDetected { + table: String, + primary_key_value: String, + stale_version: u64, + }, + + #[error("Dialect failed to build version-guarded UPDATE: {reason}")] + DialectQueryBuildFailure { reason: String }, + + #[error("Database execution failed: {reason}")] + DatabaseExecutionFailure { reason: String }, +} diff --git a/src/engine/query.rs b/src/engine/query.rs index 5e51270..de6309c 100644 --- a/src/engine/query.rs +++ b/src/engine/query.rs @@ -1,6 +1,6 @@ +use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -use chrono::{DateTime, Utc}; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct RawExpression { diff --git a/src/engine/relations.rs b/src/engine/relations.rs index 7868d27..942ea72 100644 --- a/src/engine/relations.rs +++ b/src/engine/relations.rs @@ -1,8 +1,8 @@ -use sqlx::{AnyPool, any::AnyRow}; +use crate::engine::db::{validate_identifier, SqlDialect}; +use crate::error::{BridgeOrmError, BridgeOrmResult, DiagnosticInfo}; +use sqlx::{any::AnyRow, AnyPool}; use std::sync::Arc; use tokio::sync::Mutex; -use crate::error::{BridgeOrmError, BridgeOrmResult, DiagnosticInfo}; -use crate::engine::db::{validate_identifier, SqlDialect}; pub async fn fetch_one_to_many( pool: &AnyPool, @@ -14,25 +14,30 @@ pub async fn fetch_one_to_many( ) -> BridgeOrmResult> { validate_identifier(child_table)?; validate_identifier(foreign_key)?; - + let dialect = SqlDialect::from_url(url).to_dialect(); let sql = format!( "SELECT * FROM {} WHERE {} = {}", - child_table, foreign_key, dialect.get_placeholder(0) + child_table, + foreign_key, + dialect.get_placeholder(0) ); - + let mut query = sqlx::query(&sql).bind(parent_id); - + let rows = if let Some(tx_mutex) = tx { let mut tx_guard = tx_mutex.lock().await; - let tx_conn = tx_guard - .as_mut() - .ok_or_else(|| BridgeOrmError::Validation("Transaction already closed".to_string(), DiagnosticInfo::default()))?; + let tx_conn = tx_guard.as_mut().ok_or_else(|| { + BridgeOrmError::Validation( + "Transaction already closed".to_string(), + DiagnosticInfo::default(), + ) + })?; query.fetch_all(&mut **tx_conn).await? } else { query.fetch_all(pool).await? }; - + Ok(rows) } @@ -56,21 +61,28 @@ pub async fn fetch_many_to_many( "SELECT t.* FROM {} t JOIN {} j ON t.id = j.{} WHERE j.{} = {}", - target_table, junction_table, right_key, left_key, dialect.get_placeholder(0) + target_table, + junction_table, + right_key, + left_key, + dialect.get_placeholder(0) ); let mut query = sqlx::query(&sql).bind(parent_id); let rows = if let Some(tx_mutex) = tx { let mut tx_guard = tx_mutex.lock().await; - let tx_conn = tx_guard - .as_mut() - .ok_or_else(|| BridgeOrmError::Validation("Transaction already closed".to_string(), DiagnosticInfo::default()))?; + let tx_conn = tx_guard.as_mut().ok_or_else(|| { + BridgeOrmError::Validation( + "Transaction already closed".to_string(), + DiagnosticInfo::default(), + ) + })?; query.fetch_all(&mut **tx_conn).await? } else { query.fetch_all(pool).await? }; - + Ok(rows) } @@ -88,20 +100,25 @@ pub async fn fetch_self_ref( let dialect = SqlDialect::from_url(url).to_dialect(); let sql = format!( "SELECT * FROM {} WHERE {} = {}", - table, parent_key, dialect.get_placeholder(0) + table, + parent_key, + dialect.get_placeholder(0) ); let mut query = sqlx::query(&sql).bind(parent_id); let rows = if let Some(tx_mutex) = tx { let mut tx_guard = tx_mutex.lock().await; - let tx_conn = tx_guard - .as_mut() - .ok_or_else(|| BridgeOrmError::Validation("Transaction already closed".to_string(), DiagnosticInfo::default()))?; + let tx_conn = tx_guard.as_mut().ok_or_else(|| { + BridgeOrmError::Validation( + "Transaction already closed".to_string(), + DiagnosticInfo::default(), + ) + })?; query.fetch_all(&mut **tx_conn).await? } else { query.fetch_all(pool).await? }; - + Ok(rows) } diff --git a/src/engine/session.rs b/src/engine/session.rs index 748e0ad..917173c 100644 --- a/src/engine/session.rs +++ b/src/engine/session.rs @@ -1,11 +1,11 @@ +use crate::engine::dirty_tracker::DirtyTracker; +use crate::engine::query::QueryValue; +use crate::error::BridgeOrmResult; use pyo3::prelude::*; use sqlx::{Any, AnyPool, Transaction}; use std::collections::HashMap; use std::sync::{Arc, Mutex}; use tokio::sync::Mutex as TokioMutex; -use crate::error::BridgeOrmResult; -use crate::engine::dirty_tracker::DirtyTracker; -use crate::engine::query::QueryValue; #[pyclass] #[derive(Clone)] @@ -19,39 +19,55 @@ pub struct Session { #[pymethods] impl Session { pub fn get_entity(&self, py: Python<'_>, key: String) -> PyResult> { - let map = self.identity_map.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + let map = self.identity_map.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; Ok(map.get(&key).map(|obj| obj.clone_ref(py))) } pub fn set_entity(&self, py: Python<'_>, key: String, entity: PyObject) -> PyResult<()> { - let mut map = self.identity_map.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + let mut map = self.identity_map.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; map.insert(key.clone(), entity.clone_ref(py)); Ok(()) } pub fn remove_entity(&self, _py: Python<'_>, key: String) -> PyResult<()> { - let mut map = self.identity_map.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + let mut map = self.identity_map.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; map.remove(&key); - + // Also remove snapshot when entity is removed from identity map - let mut tracker = self.dirty_tracker.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + let mut tracker = self.dirty_tracker.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; tracker.remove_snapshot(&key); Ok(()) } pub fn clear_identity_map(&self) -> PyResult<()> { - let mut map = self.identity_map.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + let mut map = self.identity_map.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; map.clear(); - - let mut tracker = self.dirty_tracker.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + + let mut tracker = self.dirty_tracker.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; tracker.snapshots.clear(); Ok(()) } pub fn get_stats(&self) -> PyResult> { - let map = self.identity_map.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; - let tracker = self.dirty_tracker.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; - + let map = self.identity_map.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; + let tracker = self.dirty_tracker.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; + let mut stats = HashMap::new(); stats.insert("identity_map_size".to_string(), map.len()); stats.insert("snapshots_count".to_string(), tracker.snapshots.len()); @@ -63,7 +79,9 @@ impl Session { pyo3_async_runtimes::tokio::future_into_py(py, async move { let mut guard = transaction.lock().await; if let Some(tx) = guard.take() { - tx.commit().await.map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; + tx.commit() + .await + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; } Ok(()) }) @@ -74,7 +92,9 @@ impl Session { pyo3_async_runtimes::tokio::future_into_py(py, async move { let mut guard = transaction.lock().await; if let Some(tx) = guard.take() { - tx.rollback().await.map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; + tx.rollback() + .await + .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?; } Ok(()) }) @@ -82,8 +102,15 @@ impl Session { } impl Session { - pub fn snapshot_entity_internal(&self, key: String, table_name: String, values: HashMap) -> PyResult<()> { - let mut tracker = self.dirty_tracker.lock().map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + pub fn snapshot_entity_internal( + &self, + key: String, + table_name: String, + values: HashMap, + ) -> PyResult<()> { + let mut tracker = self.dirty_tracker.lock().map_err(|e| { + pyo3::exceptions::PyRuntimeError::new_err(format!("Lock poisoned: {}", e)) + })?; tracker.take_snapshot(key, table_name, values); Ok(()) } diff --git a/src/engine/transaction.rs b/src/engine/transaction.rs index be6a39a..1d734a4 100644 --- a/src/engine/transaction.rs +++ b/src/engine/transaction.rs @@ -17,8 +17,12 @@ impl TxHandle { let inner = self.inner.clone(); pyo3_async_runtimes::tokio::future_into_py(py, async move { let mut guard = inner.lock().await; - let tx = guard.take().ok_or_else(|| PyValueError::new_err("Transaction already closed"))?; - tx.commit().await.map_err(|e| PyValueError::new_err(format!("Commit failed: {}", e)))?; + let tx = guard + .take() + .ok_or_else(|| PyValueError::new_err("Transaction already closed"))?; + tx.commit() + .await + .map_err(|e| PyValueError::new_err(format!("Commit failed: {}", e)))?; Ok(()) }) } @@ -27,8 +31,12 @@ impl TxHandle { let inner = self.inner.clone(); pyo3_async_runtimes::tokio::future_into_py(py, async move { let mut guard = inner.lock().await; - let tx = guard.take().ok_or_else(|| PyValueError::new_err("Transaction already closed"))?; - tx.rollback().await.map_err(|e| PyValueError::new_err(format!("Rollback failed: {}", e)))?; + let tx = guard + .take() + .ok_or_else(|| PyValueError::new_err("Transaction already closed"))?; + tx.rollback() + .await + .map_err(|e| PyValueError::new_err(format!("Rollback failed: {}", e)))?; Ok(()) }) } diff --git a/src/error.rs b/src/error.rs index b36b39f..9e2ffb1 100644 --- a/src/error.rs +++ b/src/error.rs @@ -1,5 +1,5 @@ -use thiserror::Error; use std::fmt; +use thiserror::Error; #[derive(Debug, Clone, Default)] pub struct DiagnosticInfo { diff --git a/src/ffi/java.rs b/src/ffi/java.rs index 479df54..216fcb7 100644 --- a/src/ffi/java.rs +++ b/src/ffi/java.rs @@ -1,3 +1,4 @@ +#![cfg(feature = "java-interop")] use crate::engine; use jni::objects::{JClass, JString}; use jni::sys::jstring; @@ -21,7 +22,7 @@ pub extern "system" fn Java_io_bridgeorm_core_BridgeORM_connectNative( .expect("Invalid URL string from Java") .into(); - let result = RUNTIME.block_on(async { engine::db::connect(&url_str).await }); + let result = RUNTIME.block_on(async { engine::db::connect(&url_str, None).await }); match result { Ok(_) => { diff --git a/src/ffi/mod.rs b/src/ffi/mod.rs index b8ec774..d3c2338 100644 --- a/src/ffi/mod.rs +++ b/src/ffi/mod.rs @@ -1,9 +1,29 @@ +#[macro_export] +macro_rules! ffi_guard { + ($py:expr, $body:expr) => { + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| $body)).unwrap_or_else(|e| { + let msg = if let Some(s) = e.downcast_ref::<&str>() { + s.to_string() + } else if let Some(s) = e.downcast_ref::() { + s.clone() + } else { + "Unknown Rust panic".to_string() + }; + Err(pyo3::exceptions::PyRuntimeError::new_err(format!( + "Rust panic: {}", + msg + ))) + }) + }; +} + use crate::engine; use crate::error::{BridgeOrmError, BridgeOrmResult}; use crate::schema; use crate::telemetry; pub mod java; pub mod pool_config; +pub mod type_coercion; use futures::stream::BoxStream; use futures::StreamExt; use once_cell::sync::Lazy; @@ -11,8 +31,8 @@ use pyo3::exceptions::{ PyException, PyKeyError, PyRuntimeError, PyStopAsyncIteration, PyValueError, }; use pyo3::prelude::*; -use pyo3::types::{PyDict, PyBytes}; -use sqlx::{any::AnyRow, AnyPool, Column, Row}; +use pyo3::types::{PyBytes, PyDict}; +use sqlx::{any::AnyRow, AnyPool}; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::Mutex; @@ -50,19 +70,28 @@ fn query_value_to_py(py: Python<'_>, v: QueryValue) -> PyObject { QueryValue::DateTime(dt) => { let datetime_module = py.import_bound("datetime").unwrap(); let datetime_cls = datetime_module.getattr("datetime").unwrap(); - let dt_obj = datetime_cls.call_method1("fromisoformat", (dt.to_rfc3339(),)).unwrap(); + let dt_obj = datetime_cls + .call_method1("fromisoformat", (dt.to_rfc3339(),)) + .unwrap(); dt_obj.to_object(py) } QueryValue::Json(j) => { let s = j.to_string(); let json_module = py.import_bound("json").unwrap(); - json_module.call_method1("loads", (s,)).unwrap().to_object(py) + json_module + .call_method1("loads", (s,)) + .unwrap() + .to_object(py) } QueryValue::Bytes(b) => b.to_object(py), QueryValue::Raw(raw) => { let dict = PyDict::new_bound(py); dict.set_item("sql", raw.sql).unwrap(); - let params: Vec = raw.params.into_iter().map(|p| query_value_to_py(py, p)).collect(); + let params: Vec = raw + .params + .into_iter() + .map(|p| query_value_to_py(py, p)) + .collect(); dict.set_item("params", params).unwrap(); dict.to_object(py) } @@ -70,13 +99,20 @@ fn query_value_to_py(py: Python<'_>, v: QueryValue) -> PyObject { } } -fn py_to_query_value(py: Python<'_>, obj: &Bound<'_, PyAny>, table_name: &str, column_name: &str) -> BridgeOrmResult { +fn py_to_query_value( + py: Python<'_>, + obj: &Bound<'_, PyAny>, + table_name: &str, + column_name: &str, +) -> BridgeOrmResult { if obj.is_none() { return Ok(QueryValue::Null); } let registry_guard = engine::metadata::REGISTRY.read().unwrap(); - let meta = registry_guard.mappings.get(table_name) + let meta = registry_guard + .mappings + .get(table_name) .and_then(|m| m.columns.get(column_name)); if let Some(m) = meta { @@ -87,7 +123,7 @@ fn py_to_query_value(py: Python<'_>, obj: &Bound<'_, PyAny>, table_name: &str, c if let Ok(b) = obj.extract::() { // In Python, bool is a subclass of int, so check bool first. if obj.is_instance_of::() { - return Ok(QueryValue::Bool(b)); + return Ok(QueryValue::Bool(b)); } } @@ -103,11 +139,14 @@ fn py_to_query_value(py: Python<'_>, obj: &Bound<'_, PyAny>, table_name: &str, c if let Ok(u) = uuid::Uuid::parse_str(&obj.to_string()) { // Basic check to avoid false positives with random strings if !obj.is_instance_of::() { - return Ok(QueryValue::Uuid(u)); + return Ok(QueryValue::Uuid(u)); } } - if let Ok(s) = obj.call_method0("isoformat").and_then(|r| r.extract::()) { + if let Ok(s) = obj + .call_method0("isoformat") + .and_then(|r| r.extract::()) + { if let Ok(dt) = chrono::DateTime::parse_from_rfc3339(&s) { return Ok(QueryValue::DateTime(dt.with_timezone(&chrono::Utc))); } @@ -126,7 +165,10 @@ fn py_to_query_value(py: Python<'_>, obj: &Bound<'_, PyAny>, table_name: &str, c for p in params_py { params.push(py_to_query_value(py, &p, table_name, column_name)?); } - return Ok(QueryValue::Raw(crate::engine::query::RawExpression { sql, params })); + return Ok(QueryValue::Raw(crate::engine::query::RawExpression { + sql, + params, + })); } } } @@ -192,7 +234,11 @@ fn configure_logging(level: String, slow_query_ms: u64) -> PyResult<()> { #[pyfunction] #[pyo3(signature = (url, config=None))] -fn connect(py: Python<'_>, url: String, config: Option) -> PyResult> { +fn connect( + py: Python<'_>, + url: String, + config: Option, +) -> PyResult> { let url_clone = url.clone(); pyo3_async_runtimes::tokio::future_into_py(py, async move { let pool = engine::db::connect(&url_clone, config) @@ -320,7 +366,9 @@ fn insert_row<'py>( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -330,19 +378,16 @@ fn insert_row<'py>( let mut query_data: HashMap = HashMap::new(); for (k, v) in data { let key = k.extract::()?; - query_data.insert(key.clone(), py_to_query_value(py, &v, &table_clone, &key)); + query_data.insert( + key.clone(), + py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?, + ); } pyo3_async_runtimes::tokio::future_into_py(py, async move { - let res = engine::db::generic_insert( - &pool, - tx_mutex.as_ref(), - &url, - &table, - query_data, - ) - .await - .map_err(bridge_error_to_py)?; + let res = engine::db::generic_insert(&pool, tx_mutex.as_ref(), &url, &table, query_data) + .await + .map_err(bridge_error_to_py)?; Python::with_gil(|py| { let dict = PyDict::new_bound(py); @@ -383,7 +428,9 @@ fn find_one<'py>( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -393,7 +440,10 @@ fn find_one<'py>( let mut query_filters: HashMap = HashMap::new(); for (k, v) in filters { let key = k.extract::()?; - query_filters.insert(key.clone(), py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?); + query_filters.insert( + key.clone(), + py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?, + ); } pyo3_async_runtimes::tokio::future_into_py(py, async move { @@ -421,6 +471,7 @@ fn find_one<'py>( }) } +#[cfg(feature = "data-science")] #[pyfunction] #[pyo3(signature = (table, filters, limit=None, fields=None, tx=None))] fn fetch_all_arrow<'py>( @@ -448,7 +499,9 @@ fn fetch_all_arrow<'py>( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -458,7 +511,10 @@ fn fetch_all_arrow<'py>( let mut query_filters: HashMap = HashMap::new(); for (k, v) in filters { let key = k.extract::()?; - query_filters.insert(key.clone(), py_to_query_value(py, &v, &table_clone, &key)); + query_filters.insert( + key.clone(), + py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?, + ); } pyo3_async_runtimes::tokio::future_into_py(py, async move { @@ -477,20 +533,21 @@ fn fetch_all_arrow<'py>( let buffer = engine::arrow::rows_to_arrow_ipc(&table, &rows).map_err(bridge_error_to_py)?; Python::with_gil(|py| { - let bytes = PyBytes::new_bound(py, &buffer); - Ok(bytes.to_object(py)) + let bytes = PyBytes::new_bound(py, &buffer); + Ok(bytes.to_object(py)) }) }) } #[pyfunction] -#[pyo3(signature = (table, filters, limit=None, fields=None, tx=None))] +#[pyo3(signature = (table, filters, limit=None, fields=None, eager_loads=None, tx=None))] fn fetch_all<'py>( py: Python<'py>, table: String, filters: Bound<'py, PyDict>, limit: Option, fields: Option>, + eager_loads: Option>>, tx: Option, ) -> PyResult> { let pool_guard = POOL.read().unwrap(); @@ -512,7 +569,9 @@ fn fetch_all<'py>( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -522,7 +581,10 @@ fn fetch_all<'py>( let mut query_filters: HashMap = HashMap::new(); for (k, v) in filters { let key = k.extract::()?; - query_filters.insert(key.clone(), py_to_query_value(py, &v, &table_clone, &key)); + query_filters.insert( + key.clone(), + py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?, + ); } pyo3_async_runtimes::tokio::future_into_py(py, async move { @@ -577,7 +639,9 @@ fn fetch_lazy( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -587,11 +651,15 @@ fn fetch_lazy( let mut query_filters: HashMap = HashMap::new(); for (k, v) in filters { let key = k.extract::()?; - query_filters.insert(key.clone(), py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?); + query_filters.insert( + key.clone(), + py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?, + ); } - let stream = engine::db::query_lazy(pool, tx_mutex, url, &table, query_filters, limit, fields) - .map_err(bridge_error_to_py)?; + let stream = + engine::db::query_lazy(pool, tx_mutex, url, &table, query_filters, limit, fields) + .map_err(bridge_error_to_py)?; Ok(LazyRowStream { stream: Arc::new(Mutex::new(stream)), @@ -653,7 +721,9 @@ fn insert_rows_bulk<'py>( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -665,21 +735,19 @@ fn insert_rows_bulk<'py>( let mut query_item = HashMap::new(); for (k, v) in item { let key = k.extract::()?; - query_item.insert(key.clone(), py_to_query_value(py, &v, &table_clone, &key)); + query_item.insert( + key.clone(), + py_to_query_value(py, &v, &table_clone, &key).map_err(bridge_error_to_py)?, + ); } query_items.push(query_item); } pyo3_async_runtimes::tokio::future_into_py(py, async move { - let res = engine::db::generic_insert_bulk( - &pool, - tx_mutex.as_ref(), - &url, - &table, - query_items, - ) - .await - .map_err(bridge_error_to_py)?; + let res = + engine::db::generic_insert_bulk(&pool, tx_mutex.as_ref(), &url, &table, query_items) + .await + .map_err(bridge_error_to_py)?; Python::with_gil(|py| { let mut results = Vec::new(); @@ -723,7 +791,9 @@ fn fetch_one_to_many( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -782,7 +852,9 @@ fn fetch_many_to_many( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -841,7 +913,9 @@ fn fetch_self_ref( } else if let Ok(tx_handle) = tx_obj.extract::(py) { Some(tx_handle.inner) } else { - return Err(PyValueError::new_err("Invalid transaction or session object")); + return Err(PyValueError::new_err( + "Invalid transaction or session object", + )); } } else { None @@ -913,7 +987,10 @@ fn snapshot_entity( let mut query_values = HashMap::new(); for (k, v) in values { let key = k.extract::()?; - query_values.insert(key.clone(), py_to_query_value(py, &v, &table_name, &key)); + query_values.insert( + key.clone(), + py_to_query_value(py, &v, &table_name, &key).map_err(bridge_error_to_py)?, + ); } session.snapshot_entity_internal(key, table_name, query_values) } @@ -950,19 +1027,30 @@ fn flush<'py>( let mut jobs = Vec::new(); { - let tracker_guard = session.dirty_tracker.lock().map_err(|e| PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + let tracker_guard = session + .dirty_tracker + .lock() + .map_err(|e| PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; for (key, table_name, current_values_py, pk_filters_py) in dirty_entities { let mut current_values = HashMap::new(); for (k, v) in current_values_py { let col_name = k.extract::()?; - current_values.insert(col_name.clone(), py_to_query_value(py, &v, &table_name, &col_name).map_err(bridge_error_to_py)?); + current_values.insert( + col_name.clone(), + py_to_query_value(py, &v, &table_name, &col_name) + .map_err(bridge_error_to_py)?, + ); } if let Some(diff) = tracker_guard.compute_diff(&key, ¤t_values) { let mut pk_filters = HashMap::new(); for (k, v) in pk_filters_py { let col_name = k.extract::()?; - pk_filters.insert(col_name.clone(), py_to_query_value(py, &v, &table_name, &col_name).map_err(bridge_error_to_py)?); + pk_filters.insert( + col_name.clone(), + py_to_query_value(py, &v, &table_name, &col_name) + .map_err(bridge_error_to_py)?, + ); } jobs.push(UpdateJob { @@ -984,11 +1072,16 @@ fn flush<'py>( &url, &job.table_name, job.diff, - job.pk_filters - ).await.map_err(bridge_error_to_py)?; + job.pk_filters, + ) + .await + .map_err(bridge_error_to_py)?; // Re-acquire lock to update snapshot - let mut tracker_guard = session.dirty_tracker.lock().map_err(|e| PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; + let mut tracker_guard = session + .dirty_tracker + .lock() + .map_err(|e| PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; tracker_guard.take_snapshot(job.key, job.table_name, job.full_values); } Ok(()) @@ -1013,6 +1106,7 @@ pub fn register_module(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(insert_rows_bulk, m)?)?; m.add_function(wrap_pyfunction!(find_one, m)?)?; m.add_function(wrap_pyfunction!(fetch_all, m)?)?; + #[cfg(feature = "data-science")] m.add_function(wrap_pyfunction!(fetch_all_arrow, m)?)?; m.add_function(wrap_pyfunction!(fetch_lazy, m)?)?; m.add_function(wrap_pyfunction!(snapshot_entity, m)?)?; @@ -1026,69 +1120,3 @@ pub fn register_module(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_function(wrap_pyfunction!(engine::metadata::lock_registry, m)?)?; Ok(()) } -alue(py, &v, &table_name, &col_name)); - } - - jobs.push(UpdateJob { - table_name, - diff, - pk_filters, - key, - full_values: current_values, - }); - } - } - } - - pyo3_async_runtimes::tokio::future_into_py(py, async move { - for job in jobs { - engine::db::generic_update( - &pool, - Some(&session.transaction), - &url, - &job.table_name, - job.diff, - job.pk_filters - ).await.map_err(bridge_error_to_py)?; - - // Re-acquire lock to update snapshot - let mut tracker_guard = session.dirty_tracker.lock().map_err(|e| PyRuntimeError::new_err(format!("Lock poisoned: {}", e)))?; - tracker_guard.take_snapshot(job.key, job.table_name, job.full_values); - } - Ok(()) - }) -} - -pub fn register_module(m: &Bound<'_, PyModule>) -> PyResult<()> { - m.add_class::()?; - m.add_class::()?; - m.add_class::()?; - m.add_class::()?; - m.add_class::()?; - m.add_function(wrap_pyfunction!(set_telemetry_logger, m)?)?; - m.add_function(wrap_pyfunction!(configure_logging, m)?)?; - m.add_function(wrap_pyfunction!(connect, m)?)?; - m.add_function(wrap_pyfunction!(reflect_schema, m)?)?; - m.add_function(wrap_pyfunction!(reflect_table, m)?)?; - - m.add_function(wrap_pyfunction!(begin_session, m)?)?; - m.add_function(wrap_pyfunction!(insert_row, m)?)?; - m.add_function(wrap_pyfunction!(insert_rows_bulk, m)?)?; - m.add_function(wrap_pyfunction!(find_one, m)?)?; - m.add_function(wrap_pyfunction!(fetch_all, m)?)?; - m.add_function(wrap_pyfunction!(fetch_all_arrow, m)?)?; - m.add_function(wrap_pyfunction!(fetch_lazy, m)?)?; - m.add_function(wrap_pyfunction!(snapshot_entity, m)?)?; - m.add_function(wrap_pyfunction!(flush, m)?)?; - m.add_function(wrap_pyfunction!(fetch_one_to_many, m)?)?; - m.add_function(wrap_pyfunction!(fetch_many_to_many, m)?)?; - m.add_function(wrap_pyfunction!(fetch_self_ref, m)?)?; - m.add_function(wrap_pyfunction!(execute_raw, m)?)?; - m.add_function(wrap_pyfunction!(resolve_type, m)?)?; - m.add_function(wrap_pyfunction!(engine::metadata::register_entity, m)?)?; - m.add_function(wrap_pyfunction!(engine::metadata::lock_registry, m)?)?; - Ok(()) -} -n(wrap_pyfunction!(engine::metadata::lock_registry, m)?)?; - Ok(()) -} diff --git a/src/ffi/type_coercion.rs b/src/ffi/type_coercion.rs index 45d28df..fe64e05 100644 --- a/src/ffi/type_coercion.rs +++ b/src/ffi/type_coercion.rs @@ -37,7 +37,7 @@ pub fn coerce_py_value( py_val .get_type() .name() - .unwrap_or("unknown") + .unwrap_or(std::borrow::Cow::Borrowed("unknown")) .to_string() .as_str(), ) diff --git a/src/telemetry/init.rs b/src/telemetry/init.rs index d0c9695..243d09f 100644 --- a/src/telemetry/init.rs +++ b/src/telemetry/init.rs @@ -8,27 +8,8 @@ pub fn init_tracer( service_name: &str, otlp_endpoint: &str, ) -> Result<(), Box> { - let exporter = opentelemetry_otlp::new_exporter() - .tonic() - .with_endpoint(otlp_endpoint); - - let tracer = opentelemetry_otlp::new_pipeline() - .tracing() - .with_exporter(exporter) - .with_trace_config( - sdktrace::Config::default().with_resource( - opentelemetry_sdk::Resource::new(vec![ - opentelemetry::KeyValue::new("service.name", service_name.to_string()), - ]) - ) - ) - .install_batch(runtime::Tokio)?; - - let otel_layer = OpenTelemetryLayer::new(tracer); - tracing_subscriber::registry() .with(EnvFilter::from_default_env()) - .with(otel_layer) .with(tracing_subscriber::fmt::layer().json()) .init(); diff --git a/src/telemetry/mod.rs b/src/telemetry/mod.rs index a313724..5f5d674 100644 --- a/src/telemetry/mod.rs +++ b/src/telemetry/mod.rs @@ -1,2 +1,2 @@ -pub mod logger; pub mod init; +pub mod logger;