diff --git a/README.md b/README.md index e806af4..c288804 100644 --- a/README.md +++ b/README.md @@ -3,8 +3,8 @@

- A composable architectural foundation for building scalable applications on top of FastAPI.
- Standardize your structure, manage atomic transactions, and protect sensitive data β€” without sacrificing FastAPI's flexibility. + An architectural chassis for building modular applications on top of FastAPI.
+ Structuring transactional boundaries, constructor dependency injection, and field-level access controls while preserving native FastAPI semantics.

@@ -29,51 +29,49 @@ ## 🧭 Overview -**FastAPI is an exceptional microframework for HTTP routing, request parsing, and asynchronous I/O.** +FastAPI provides an efficient, ergonomic foundation for HTTP handling, asynchronous I/O, and data validation using Pydantic. However, as codebases grow beyond simple APIs into larger services or modular monoliths, engineering teams frequently face common structural challenges: -However, as applications grow into medium and large systems, teams inevitably encounter an architectural gap: -* Database sessions leak across asynchronous tasks or background jobs. -* Business workflows spanning multiple repositories lack reliable, all-or-nothing transaction boundaries. -* Teams end up writing dozens of nearly identical DTO schemas just to hide internal fields (like profit margins, supplier terms, or sensitive IDs) from different user roles. -* Domain modules become tightly coupled through circular imports and tangled dependency trees. +* **Session Lifetime & Leaks:** Database sessions can inadvertently leak across concurrent coroutines or terminate prematurely during background tasks. +* **Transaction Coordination:** Workflows that update multiple repositories often lack a unified, atomic transaction boundary, risking partial writes when a middle step fails. +* **Schema Proliferation:** Writing and maintaining separate Pydantic models (`UserPublic`, `UserInternal`, `UserAdminResponse`) solely to conceal specific fields across user roles introduces repetitive boilerplate. +* **Module Coupling:** Large domain modules frequently develop tangled dependency graphs and circular imports when interacting with shared databases. -Heavy enterprise frameworks solve these challenges by enforcing rigid conventions, proprietary ORMs, and heavy abstractions that eliminate your freedom. +Full-stack monolithic frameworks solve these issues by introducing comprehensive conventions, coupled ORMs, and centralized registries. While effective, this approach can reduce architectural flexibility. -**ZCore is the middle ground:** -ZCore provides a structured, modular layer that sits directly on top of standard FastAPI, SQLAlchemy 2.0, and Pydantic V2. It introduces proven architectural patterns (Unit of Work, IoC Dependency Injection, Field-Level Security, and Domain Event Dispatching) as **composable building blocks**. +**ZCore is designed as an intermediate layer.** It does not hide FastAPI or replace its router. Instead, it provides a set of decoupled architectural primitivesβ€”such as a re-entrant Unit of Work, constructor-based Inversion of Control (IoC), context-aware schema filtering (`Zchema`), and lifecycle pluginsβ€”built on top of standard **FastAPI**, **SQLAlchemy 2.0 (Async)**, and **Pydantic V2**. -> **Your code remains 100% standard FastAPI.** Adopt ZCore piece by piece where it solves real problems, or use the full architectural stack β€” with zero vendor lock-in. +You can adopt individual components where they solve concrete architectural needs or use the structured chassis as a cohesive pattern for modular application design. --- -## πŸ›οΈ Reference Showcase: ZShop +## πŸ›οΈ Reference Implementation: ZShop -To see ZCore's architectural concepts in action before writing a single line of code, explore **[ZShop](https://github.com/Baseryn/zshop)** β€” the official Modular Monolith e-commerce reference implementation: +To examine how these architectural patterns interact in an end-to-end codebase, refer to **[ZShop](https://github.com/Baseryn/zshop)**, a modular monolith reference application: -* **Interactive Live Inspector Dock:** Monitor active request context, correlation IDs (`x-request-id`), execution latencies, and real-time SSE stream events. -* **1-Click RBAC Persona Switcher:** Switch between SuperAdmin, Store Manager, Customer, and Guest to watch confidential data fields automatically vanish from responses via `Zchema`. -* **Atomic Rollback Demonstration:** An interactive deficit checkout button verifying that stock adjustments and order persistence roll back atomically when any invariant fails. -* **Decoupled Architecture:** Strict inter-module communication using Python Protocols (`ProductContract`) without cross-domain database coupling. +* **Context & Security Inspection:** Demonstrates active correlation IDs (`x-request-id`), execution profiling, and role-based data projection. +* **Role-Based Schema Pruning:** Displays how confidential attributes (such as supplier costs) are dynamically excluded based on the authenticated context via `Zchema`. +* **Atomic Rollback Demonstration:** Validates multi-repository rollback guarantees by simulating invariant violations during order checkout. +* **Contract-Driven Decoupling:** Demonstrates cross-domain communication via Python `Protocols` (`ProductContract`) rather than direct database model imports. -πŸ‘‰ **[Explore the ZShop Showcase Repository](https://github.com/Baseryn/zshop)** +πŸ‘‰ **[Explore the ZShop Reference Codebase](https://github.com/Baseryn/zshop)** --- -## βš–οΈ Why ZCore? (An Honest Comparison) +## βš–οΈ Architectural Comparison -| Architectural Concern | Plain FastAPI | Full-Stack Monoliths (e.g. Django) | With ZCore on FastAPI | +| Area | Plain FastAPI | Heavy Monoliths (e.g., Django) | FastAPI with ZCore | | :--- | :--- | :--- | :--- | -| **Application Philosophy** | Minimalist microframework; no opinion on database, folders, or architecture. | Opinionated, batteries-included; tightly coupled to its own ORM and template engine. | **Composable layer;** provides enterprise architecture patterns while keeping FastAPI completely standard. | -| **Role-Based Data Exposure** | Writing multiple Pydantic models per role (`UserPublic`, `UserAdmin`) or manual `if/else` checks. | Model-level permissions or custom serializer logic. | **Single-schema context shielding (`Zchema`);** forbidden fields are pruned dynamically from validation, serialization, and OpenAPI. | -| **Multi-Repository Operations** | Scattered `session.commit()` calls; high risk of partial writes on failure. | Managed transaction blocks (`transaction.atomic()`). | **Re-entrant `UnitOfWork`;** coordinated all-or-nothing boundaries with domain events deferred until physical commit succeeds. | -| **Background Processing** | Passing request sessions to background tasks often leads to `Session is closed` errors. | Requires external queue workers (Celery, RQ) even for simple background jobs. | **Dedicated `@background_task`;** automatically isolates DB sessions and IoC lifecycles from completed HTTP requests. | -| **Domain Modularity** | Standard APIRouters; easily slips into spaghetti code and circular imports at scale. | Django Apps with rigid settings and global app registries. | **DAG-driven `Plugin` system;** topological startup sorting and pure Python Protocol contracts for zero-coupling. | -| **Dataset Pagination** | Manual `offset / limit` SQL slicing that degrades on large tables. | Built-in offset paginators. | **Keyset Cursor Pagination (Base64);** zero-drift pagination and dynamic AST-like JSON search engine. | -| **Testing Isolation** | Manual DB cleanup routines or slow schema recreation per test file. | TransactionTestCase rolling back transactions per test. | **Zero-side-effect `ZTestClient`;** automated savepoint rollbacks, context mocking, and container sandboxing. | +| **Philosophy** | Minimalist microframework; architectural organization is left entirely to the developer. | Batteries-included, opinionated; coupled to built-in ORM and global application registries. | **Composable layer;** provides structured architectural patterns while keeping routers and schemas standard. | +| **Field-Level Security** | Manual `if/else` checks or maintaining distinct Pydantic DTOs for each access level. | Model-level field permissions or specialized serializer layers. | **Context-aware pruning (`Zchema`);** dynamically filters restricted fields across validation, serialization, and dynamic schema introspection. | +| **Transaction Management** | Manual `session.commit()` calls across routes and services; potential for partial writes. | Built-in transaction blocks (`transaction.atomic()`). | **Re-entrant `UnitOfWork`;** depth-aware coordination with domain events buffered until root commit succeeds. | +| **Background Processing** | Request-scoped sessions passed to `BackgroundTasks` often encounter closed session errors. | Relies on external task queue workers (e.g., Celery) even for simple background steps. | **Scoped `@background_task`;** isolates session allocation and IoC lifecycles from completed request scopes. | +| **Modular Domain Design** | Standard `APIRouter` instances; vulnerable to circular imports as features expand. | Strict app layout and global settings configuration. | **DAG-driven `Plugin` system;** topological startup sorting and pure Python Protocol contracts for interface decoupling. | +| **Dataset Pagination** | Manual `offset / limit` SQL slicing, which may suffer performance degradation on large tables. | Built-in offset paginators. | **Keyset Cursor Pagination;** compound keyset traversal with URL-safe base64 cursors, alongside dynamic JSON filtering. | +| **Test Isolation** | Manual database cleanup or recreating schema fixtures between test runs. | Transactional test cases rolling back writes per test method. | **Rollback harness via `ZTestClient`;** automated savepoint rollbacks, container isolation, and context mocking. | --- -## πŸš€ Quick Start (in Under 60 Seconds) +## πŸš€ Quick Start ### 1. Installation @@ -81,21 +79,21 @@ To see ZCore's architectural concepts in action before writing a single line of pip install "fastapi-zcore-framework[all]" ``` -### 2. Scaffold a New Project +### 2. Initialize a Project -The interactive `zc` CLI configures your environment, sets up a virtual environment (`uv` or `pip`), and provisions your database driver: +The interactive `zc` CLI configures your workspace, sets up a virtual environment (`uv` or `pip`), and bootstraps the appropriate asynchronous database driver: ```bash -# Interactive mode (recommended) +# Interactive setup zc init core_api && cd core_api -# Generate your first domain module +# Scaffold a domain module zc startapp tasks ``` ### 3. Define Your Domain Model -Open `tasks/models.py` (standard SQLAlchemy 2.0): +In `tasks/models.py` (standard SQLAlchemy 2.0): ```python import uuid @@ -116,33 +114,35 @@ class Task(Base, SoftDeleteMixin): zc run ``` -Your API is running at `http://127.0.0.1:8000` with **8 production-ready endpoints** out of the box: -* `POST /tasks/` β€” Create entity -* `GET /tasks/{id}` β€” Fetch single record (with automatic PK type inspection) -* `GET /tasks/` β€” List records (with cursor or offset pagination) -* `POST /tasks/search` β€” Dynamic nested JSON filtering engine -* `POST /tasks/lookup` β€” Lightweight field-projected endpoint for dropdowns -* `PUT /tasks/{id}` & `PATCH /tasks/{id}` β€” Full and partial updates -* `DELETE /tasks/{id}` β€” Soft delete (or permanent purge via `?force=true`) -* `GET /tasks/?schema=true` β€” Dynamic role-aware frontend form schema +The application starts at `http://127.0.0.1:8000` with 8 standard endpoints managed by `BaseRouter`: +* `POST /tasks/` β€” Entity creation +* `GET /tasks/{id}` β€” Single-record retrieval with primary key inspection +* `GET /tasks/` β€” List retrieval supporting offset or keyset cursor pagination +* `POST /tasks/search` β€” Dynamic nested JSON filtering and sorting +* `POST /tasks/lookup` β€” Lightweight, column-projected endpoint for relational references +* `PUT /tasks/{id}` & `PATCH /tasks/{id}` β€” Complete and partial updates +* `DELETE /tasks/{id}` β€” Soft delete (or physical purge via `?force=true`) +* `GET /tasks/_schema` β€” Role-aware schema introspection (all actions or filtered via `?action=create|update|view|lookup`) --- -## πŸ›οΈ Core Architectural Patterns +## πŸ›οΈ Core Architectural Components

-1. Context Shielding & Role-Based Field Pruning (Zchema) +1. Context Shielding & Field-Level Access Control (Zchema) -### The Problem -In enterprise applications, different roles require different views of the same data. A Customer must not see wholesale cost prices or internal supplier notes, while a Store Manager requires full financial transparency. Maintaining separate DTOs (`ProductPublic`, `ProductManager`, `ProductCreate`, `ProductAdminUpdate`) creates boilerplate and maintenance overhead. +### Problem +Managing multiple versions of the same schema (`ProductCustomerView`, `ProductInternalView`, `ProductAdminUpdate`) to hide sensitive attributes (such as purchase cost or internal remarks) introduces redundancy and increases the risk of accidental data leaks. -### The ZCore Solution -Define your model once inheriting from `Zchema`. When a user's security context marks a field as restricted, ZCore automatically prunes it across **input parsing** (preventing mass assignment), **response serialization** (preventing data leakage), and **OpenAPI schema generation**: +### Solution +Define the schema once by inheriting from `Zchema`. When fields are marked as restricted within the current request context (`ctx.restricted_fields`), ZCore prunes them across: +1. **Input validation:** Silently discards restricted fields to prevent mass assignment. +2. **Output serialization:** Removes restricted attributes before JSON encoding. +3. **Schema introspection:** Omits restricted properties dynamically when querying the `/_schema` endpoint. ```python import uuid from decimal import Decimal -from pydantic import Field from zcore import Zchema class ProductResponse(Zchema): @@ -151,51 +151,50 @@ class ProductResponse(Zchema): id: uuid.UUID name: str price: Decimal - cost_price: Decimal | None = None # πŸ”’ Sensitive business margin - supplier_notes: str | None = None # πŸ”’ Internal agreement details + cost_price: Decimal | None = None # Internal margin + supplier_notes: str | None = None # Vendor agreement notes ``` -* **When a Customer views the product:** `cost_price` and `supplier_notes` are completely omitted from the JSON output. ZCore automatically attaches a `Vary: Authorization, Cookie` header to protect downstream CDNs from cache poisoning. -* **When a Manager views the product:** All fields are returned transparently. +When restricted fields are pruned from a response, ZCore appends a `Vary: Authorization, Cookie` header to prevent shared HTTP caches or CDNs from caching privileged data representations for unprivileged clients.
2. Atomic Unit of Work & Deferred Domain Events (UnitOfWork) -### The Problem -In distributed or multi-repository architectures, business actions involve multiple persistence steps (e.g., deducting inventory, recording a purchase order, and generating an invoice). If step 2 fails, step 1 must be reverted. Furthermore, if you send an email or emit a domain event *before* the database commit succeeds, a subsequent failure leaves external systems out of sync. +### Problem +When business operations touch multiple repositories, issuing separate commits introduces the risk of partial database writes if an intermediate step fails. Additionally, publishing domain events (or sending notifications) before a database transaction physically commits can lead to state inconsistency if the transaction ultimately rolls back. -### The ZCore Solution -ZCore implements a re-entrant, depth-aware `UnitOfWork`. Nested transactions safely flush changes to savepoints without prematurely committing, while domain events are buffered and dispatched **only after the root commit succeeds**: +### Solution +ZCore implements a depth-aware, re-entrant `UnitOfWork`. Nested scopes flush changes safely without premature commits, while domain events remain buffered and are dispatched only after the root commit succeeds: ```python from zcore import UnitOfWork async with UnitOfWork(session=self.session, dispatcher=self.dispatcher) as uow: - # 1. Adjust inventory + # 1. Update inventory await self.inventory_service.adjust_stock(product_id, -quantity) - # 2. Persist order + # 2. Persist order entity self.order_repo.add(new_order) await self.session.flush() # 3. Buffer domain event - # If the transaction rolls back, this event is discarded immediately! + # If any error occurs before commit completes, this event is discarded. uow.register_event("order.created", {"order_id": str(new_order.id)}) ```
-3. Contract-Driven Decoupling via Inversion of Control (IoC Container) +3. Constructor-Based Dependency Injection (IoC Container) -### The Problem -Modular monoliths frequently suffer from circular imports when domain A (e.g., Orders) needs data from domain B (e.g., Catalog). Importing repositories or models across module boundaries creates tight database coupling. +### Problem +In complex applications, deep chains of FastAPI `Depends(...)` can become verbose. Furthermore, domain modules risk circular imports when cross-referencing concrete repository implementations. -### The ZCore Solution -Domains communicate strictly through pure Python `Protocols` (Contracts). The IoC Container auto-wires dependencies at runtime using constructor type hints with signature caching: +### Solution +The built-in IoC container supports constructor-based auto-wiring via type annotations, caching constructor signatures to minimize reflection overhead. It supports Singleton, Scoped, and Transient lifecycles, and automatically skips parameters with default values unless explicitly registered: ```python -# contracts/catalog.py (No database imports!) +# Pure Python contract (contracts/catalog.py) from typing import Protocol, runtime_checkable @runtime_checkable @@ -204,11 +203,11 @@ class ProductContract(Protocol): ``` ```python -# apps/orders/services.py +# Domain implementation (apps/orders/services.py) from contracts.catalog import ProductContract class OrderService(BaseService[Orders]): - # Pure Python constructor: Auto-wired by ZCore's IoC container! + # Dependencies are resolved by constructor type hints def __init__(self, repository: OrderRepository, inventory: ProductContract): super().__init__(model=Orders, repository=repository) self.inventory = inventory @@ -216,35 +215,45 @@ class OrderService(BaseService[Orders]):
-4. Isolated Background Scopes (@background_task) +4. Isolated Background Execution (@background_task) -### The Problem -FastAPI's built-in `BackgroundTasks` executes callables after sending the HTTP response. If you pass an active request-scoped database session to a background task, the session is closed by the time the worker executes, causing `InterfaceError: Session is closed`. +### Problem +FastAPI executes background tasks after completing the HTTP response. If a task receives an `AsyncSession` bound to the finished request lifecycle, the session is often closed by the time execution begins, causing connection errors. -### The ZCore Solution -The `@background_task` decorator creates an independent execution scope, provisions a dedicated database session, binds an isolated logging task ID, and auto-resolves annotated dependencies from the container: +### Solution +The `@background_task` decorator creates an independent execution scope. It provisions a dedicated database session, isolates context variables, and auto-wires type-annotated dependencies directly from the container: ```python +import uuid from zcore import background_task @background_task async def generate_invoice_task( order_id: uuid.UUID, - order_repo: OrderRepository, # Automatically injected with an isolated session! + order_repo: OrderRepository, # Injected with an independent session ) -> None: order = await order_repo.get(id=order_id) - # Safely executes without relying on completed HTTP request lifecycles + # Operates safely outside of the HTTP request lifecycle ```
-5. Zero-Side-Effect Testing Engine (ZTestClient) +5. Composite Primary Keys & Keyset Pagination + +### Capabilities +* **Composite Primary Keys:** `BaseRepository` natively handles composite models and PostgreSQL partitioned tables (e.g., `(id, operation_date)`). Operations use tuple-based lookups (`tuple_(*pks).in_(...)`), allowing PostgreSQL to perform partition pruning without full table scans. +* **Compound Keyset Pagination:** `CursorPagination` supports non-primary-key sorting columns using compound keyset conditions `(col < val) OR (col == val AND pk < last_pk)`. It handles type coercion for `date`, `datetime`, `UUID`, and numeric types. +* **Mutual Exclusivity:** `SearchRequest` validates against conflicting pagination modes, raising a 422 error if both offset (`page > 1`) and keyset (`cursor`) parameters are supplied simultaneously. +
+ +
+6. Isolated Testing Harness (ZTestClient) -### The Problem -Integration tests that write to a database require tedious teardown scripts or slow table recreation between test cases, leading to flaky test suites. +### Problem +Integration tests that write directly to a database typically require slow schema teardowns or manual cleanup scripts to prevent test state leakage. -### The ZCore Solution -`ZTestClient` executes tests inside **transactional savepoint rollbacks**. Changes are rolled back immediately upon test completion, keeping test runs fast and deterministic: +### Solution +`ZTestClient` coordinates tests within transactional savepoints. Database modifications are automatically rolled back upon test completion, while container bindings and event listeners are restored from snapshots: ```python import pytest @@ -255,25 +264,25 @@ from main import app async def test_order_placement_atomicity(): async with ZTestClient( app, - user_id="customer-uuid", + user_id="test-user-id", scopes=["orders:create", "products:view"] ) as client: response = await client.post("/checkout/orders/", json={...}) assert response.status_code == 201 - # Database is automatically rolled back to its exact pre-test state! + # Transaction changes are rolled back automatically upon context exit ```
--- -## πŸ› οΈ Interactive Developer Tooling (`zc` CLI) +## πŸ› οΈ Developer Tooling (`zc` CLI) -ZCore includes an interactive terminal user interface (TUI) powered by Questionary and Rich: +ZCore provides an interactive command-line interface powered by Questionary and Rich: ```text $ zc -⚑ ZCore Framework v0.1.0-rc.2 β€’ Modern Modular Monolith +⚑ ZCore Framework v0.1.0-rc.3 β€’ Modern Modular Monolith FastAPI β€’ SQLAlchemy 2.0 β€’ Pydantic V2 ? What framework task would you like to perform? @@ -285,43 +294,43 @@ $ zc πŸšͺ exit β€” Exit CLI ``` -### CLI Command Summary +### Command Reference -| Command | Key Flags | Purpose | +| Command | Key Options | Purpose | | :--- | :--- | :--- | -| `zc` | β€” | Open interactive terminal dashboard | -| `zc init [name]` | `--db [sqlite\|postgres\|mysql]`, `-y` | Bootstrap project with `.env`, `.venv`, and driver dependencies | -| `zc startapp [name]` | `--template / --no-template`, `--test`, `-y` | Generate domain module layers (models, schemas, repos, services, routers) | -| `zc run [app]` | `--host`, `--port`, `--reload`, `--workers`, `--env-file` | Run server with cascading config hierarchy (`CLI > .env > Defaults`) | -| `zc gensecret` | β€” | Generate a 64-character cryptographic `SECRET_KEY` | -| `zc genenv` | `-o `, `-f / --force` | Introspect active Pydantic `Settings` and scaffold `.env.example` | +| `zc` | β€” | Launches interactive dashboard | +| `zc init [name]` | `--db [sqlite\|postgres\|mysql]`, `-y` | Scaffolds project files, `.env`, and initializes virtual environment | +| `zc startapp [name]` | `--template / --no-template`, `--test`, `-y` | Generates architectural layers (models, schemas, repos, services, routers) | +| `zc run [app]` | `--host`, `--port`, `--reload`, `--workers`, `--env-file` | Starts Uvicorn with hierarchical configuration (`CLI > .env > Defaults`) | +| `zc gensecret` | β€” | Generates a 64-character cryptographic `SECRET_KEY` | +| `zc genenv` | `-o `, `-f / --force` | Introspects active Pydantic `Settings` and outputs a template `.env` | --- -## πŸ“’ Community Testing & Feedback (Release Candidate `rc.2`) +## πŸ“’ Testing & Community Validation (Release Candidate `rc.3`) -We are currently validating **`v0.1.0-rc.2`** before tagging the official `v1.0.0` stable release. +We are currently evaluating **`v0.1.0-rc.3`** across real-world workloads prior to finalizing the `v1.0.0` stable release. -We built ZCore to solve architectural pain points we faced in production. Before declaring the API stable, **we need your critical engineering feedback:** +ZCore was created to address architectural friction points encountered in production environments. We welcome technical feedback, bug reports, and discussion on edge cases: -* **Try the Showcase:** Clone [ZShop](https://github.com/Baseryn/zshop), run the seed script, and test the endpoints. -* **Test the Edge Cases:** Push the boundaries of `UnitOfWork`, test complex nested queries with `SearchEngine`, or inspect field pruning with `Zchema`. -* **Break It & Report:** If an abstraction feels awkward, if a type hint breaks, or if an async session leaks, please open an issue! +* **Evaluate the Showcase:** Review [ZShop](https://github.com/Baseryn/zshop) to observe how components interact in a concrete project. +* **Test Architectural Boundaries:** Evaluate `UnitOfWork` behavior under nested operations, verify complex queries with `SearchEngine`, or test composite key models. +* **Report Issues:** If an abstraction feels rigid, if a type hint fails strict static analysis, or if an edge case is unhandled, please open a detailed issue on GitHub. -πŸ‘‰ **[Submit Issues & Architectural Feedback on GitHub](https://github.com/Baseryn/zcore/issues)** +πŸ‘‰ **[Submit Issues & Technical Feedback on GitHub](https://github.com/Baseryn/zcore/issues)** --- -## πŸ“š Documentation Index +## πŸ“š Documentation -Comprehensive guides, architectural deep-dives, and complete API references are available at our official documentation portal: +Detailed guides, design explanations, and API references are available at our documentation portal: -* **[Quick Start Tutorial](https://zcore.baseryn.com/docs/quick-start)** β€” From zero to a working API in 60 seconds. -* **[10-Step Quick Learn](https://zcore.baseryn.com/docs/quick-learn/step-1)** β€” Systematic layer-by-layer walkthrough. -* **[Architectural Comparisons](https://zcore.baseryn.com/docs/comparisons)** β€” Honest comparisons with plain FastAPI, Django, and other frameworks. -* **[How-To Guides](https://zcore.baseryn.com/docs/how-to)** β€” Practical recipes for Multi-tenancy, Keyset Pagination, File Uploads, and Caching. +* **[Quick Start Tutorial](https://zcore.baseryn.com/docs/quick-start)** β€” Basic project initialization and endpoint setup. +* **[Layer-by-Layer Guide](https://zcore.baseryn.com/docs/quick-learn/step-1)** β€” Walkthrough of model, repository, service, and router separation. +* **[Architectural Comparisons](https://zcore.baseryn.com/docs/comparisons)** β€” Technical comparisons with plain FastAPI and Django. +* **[How-To Guides](https://zcore.baseryn.com/docs/how-to)** β€” Practical patterns for Keyset Pagination, File Storage, and Caching. * **[Core Concepts](https://zcore.baseryn.com/docs/core-concepts/context)** β€” Deep dives into DI auto-wiring, Unit of Work, Kernel DAG, and Context shielding. -* **[API Reference](https://zcore.baseryn.com/docs/api-reference/repository)** β€” Detailed signatures and class specifications. +* **[API Reference](https://zcore.baseryn.com/docs/api-reference/repository)** β€” Method signatures and parameter details. --- @@ -333,5 +342,5 @@ See the [LICENSE](https://github.com/Baseryn/zcore/blob/master/LICENSE) file for ---

- Developed with architectural rigor and care by Ali Alf Ostovar / Baseryn. + Maintained by Ali Alf Ostovar and the Baseryn open-source community.

\ No newline at end of file diff --git a/src/zcore/__init__.py b/src/zcore/__init__.py index f3f13c2..fd39ec2 100644 --- a/src/zcore/__init__.py +++ b/src/zcore/__init__.py @@ -1,5 +1,7 @@ """ZCore Framework Core Package Entrypoint.""" +__version__ = "0.1.0-rc.3" + from typing import TYPE_CHECKING, Any from zcore.config import settings @@ -113,6 +115,7 @@ "ZCoreRequest", "ZDateTime", "Zchema", + "__version__", "app_exception_handler", "background_scope", "background_task", diff --git a/src/zcore/cache/base.py b/src/zcore/cache/base.py index c09433d..681e354 100644 --- a/src/zcore/cache/base.py +++ b/src/zcore/cache/base.py @@ -8,6 +8,7 @@ import asyncio import contextlib +import threading from typing import Any, Generic, TypeVar import structlog @@ -29,6 +30,7 @@ _shared_redis_client: Any | None = None _eviction_task: asyncio.Task | None = None +_eviction_lock = threading.Lock() def _ensure_eviction_task(interval: int | None = None) -> None: @@ -42,10 +44,13 @@ def _ensure_eviction_task(interval: int | None = None) -> None: effective_interval: int = int(raw_interval) if raw_interval is not None else 60 try: loop = asyncio.get_running_loop() - if _eviction_task is None or _eviction_task.done(): - _eviction_task = loop.create_task(_start_eviction_loop(interval=effective_interval)) except RuntimeError: - pass + return + + if _eviction_task is None or _eviction_task.done(): + with _eviction_lock: + if _eviction_task is None or _eviction_task.done(): + _eviction_task = loop.create_task(_start_eviction_loop(interval=effective_interval)) def init_cache(redis_url: str | None = None, **kwargs: Any) -> None: @@ -89,11 +94,14 @@ async def close_cache() -> None: """Cancel background loops and cleanly close distributed connections.""" global _shared_redis_client, _eviction_task - if _eviction_task and not _eviction_task.done(): - _eviction_task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await _eviction_task + with _eviction_lock: + task = _eviction_task _eviction_task = None + + if task and not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task logger.debug("Cache memory eviction loop stopped.") if _shared_redis_client: diff --git a/src/zcore/cli/__init__.py b/src/zcore/cli/__init__.py index 743b1d8..b411a86 100644 --- a/src/zcore/cli/__init__.py +++ b/src/zcore/cli/__init__.py @@ -8,6 +8,7 @@ import questionary from questionary import Choice +from zcore import __version__ from zcore.cli.commands import ( DB_DRIVERS, find_zcore_projects, @@ -366,7 +367,7 @@ def main() -> None: if args.version: console.print( - f"[bold {ZCORE_PRIMARY}]ZCore Framework[/bold {ZCORE_PRIMARY}] - Version [bold white]0.1.0-rc.2[/bold white]" + f"[bold {ZCORE_PRIMARY}]ZCore Framework[/bold {ZCORE_PRIMARY}] - Version [bold white]{__version__}[/bold white]" ) sys.exit(0) diff --git a/src/zcore/cli/templates.py b/src/zcore/cli/templates.py index 9b165b6..38ee86a 100644 --- a/src/zcore/cli/templates.py +++ b/src/zcore/cli/templates.py @@ -5,7 +5,14 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) from fastapi import FastAPI -from zcore import Kernel, db_manager, register_db_event_dispatcher, register_exception_handlers, settings +from zcore import ( + Kernel, + __version__, + db_manager, + register_db_event_dispatcher, + register_exception_handlers, + settings, +) from zcore.logging import setup_logging from zcore.web import RequestLogMiddleware, ScopedDependencyMiddleware @@ -33,7 +40,7 @@ async def root(): return { "status": "healthy", "framework": "ZCore", - "version": "0.1.0-rc.2", + "version": __version__, "debug": settings.DEBUG } """ diff --git a/src/zcore/cli/ui.py b/src/zcore/cli/ui.py index 9b60510..8a50726 100644 --- a/src/zcore/cli/ui.py +++ b/src/zcore/cli/ui.py @@ -3,6 +3,8 @@ from questionary import Style from rich.console import Console +from zcore import __version__ + console = Console() ZCORE_PRIMARY = "#10b981" @@ -29,7 +31,7 @@ def print_banner() -> None: console.print() console.print( f"[bold {ZCORE_PRIMARY}]⚑ ZCore Framework[/bold {ZCORE_PRIMARY}] " - f"[dim white]v0.1.0-rc.2[/dim white] " + f"[dim white]v{__version__}[/dim white] " f"[{ZCORE_MUTED}]β€’ Modern Modular Monolith[/{ZCORE_MUTED}]" ) console.print(f" [dim {ZCORE_MUTED}]FastAPI β€’ SQLAlchemy 2.0 β€’ Pydantic V2[/dim {ZCORE_MUTED}]") diff --git a/src/zcore/config.py b/src/zcore/config.py index d70fbd3..380a39c 100644 --- a/src/zcore/config.py +++ b/src/zcore/config.py @@ -13,7 +13,7 @@ from pydantic import AliasChoices, BaseModel, Field, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict -from zcore.kernel.di import container +from zcore.kernel.di import DIException, container T = TypeVar("T", bound="Settings") @@ -174,6 +174,8 @@ def get_settings(settings_class: type[T] = Settings) -> T: # type: ignore[assig The resolved settings instance of type `T`. """ try: + if not container.is_registered(settings_class): + raise DIException(f"Settings class {settings_class.__name__} is not initialized.") return cast(T, container.resolve(settings_class)) except Exception: settings_inst = settings_class() diff --git a/src/zcore/db/pagination.py b/src/zcore/db/pagination.py index 6bcac5f..921dc4b 100644 --- a/src/zcore/db/pagination.py +++ b/src/zcore/db/pagination.py @@ -11,14 +11,15 @@ import math import uuid from collections.abc import Sequence -from datetime import UTC, datetime +from datetime import UTC, date, datetime +from decimal import Decimal from typing import Any, Generic, Literal, TypeVar from pydantic import BaseModel, Field, model_validator from sqlalchemy import and_, func, inspect, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import Select -from sqlalchemy.sql.sqltypes import DateTime +from sqlalchemy.sql.sqltypes import Date, DateTime from zcore.config import settings from zcore.exceptions.base import ValidationError @@ -87,10 +88,14 @@ class CursorParams(BaseModel): cursor: The base64-encoded keyset cursor string specifying position context. If None, fetches from the beginning. Defaults to None. size: The maximum quantity of records to yield in the dataset. + cursor_field: Target column name used for keyset boundary evaluation. Defaults to None. + order: Target sorting order applied during keyset boundary traversal. Defaults to None. """ cursor: str | None = None size: int | None = None + cursor_field: str | None = None + order: Literal["asc", "desc"] | str | None = None @model_validator(mode="after") def _validate_and_bound_size(self) -> "CursorParams": @@ -227,7 +232,7 @@ def __init__(self, cursor_field: str | None = None, order: str = "desc"): order: The sorting order ('asc' or 'desc'). Defaults to "desc". """ self.cursor_field = cursor_field - self.order = order.lower() + self.order = order.lower() if order else "desc" def _encode_cursor(self, last_item: Any, cursor_field: str, pk_name: str) -> str: """Serialize and encode an entity's coordinate parameters to base64. @@ -244,6 +249,10 @@ def _encode_cursor(self, last_item: Any, cursor_field: str, pk_name: str) -> str if isinstance(value, datetime): value = value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) value = value.isoformat() + elif isinstance(value, date): + value = value.isoformat() + elif isinstance(value, (uuid.UUID, Decimal)): + value = str(value) payload = {"value": value, "pk": str(getattr(last_item, pk_name, ""))} @@ -310,7 +319,9 @@ async def paginate( mapper = inspect(model) pk_col = mapper.primary_key[0] pk_name = pk_col.name - effective_cursor_field = self.cursor_field or pk_name + + effective_cursor_field = params.cursor_field or self.cursor_field or pk_name + effective_order = (params.order or self.order or "desc").lower() valid_columns = {col.key for col in mapper.columns} if effective_cursor_field not in valid_columns: @@ -336,22 +347,41 @@ async def paginate( with contextlib.suppress(ValueError): last_pk = uuid.UUID(last_pk) - if isinstance(col.type, DateTime) and isinstance(val, str): + try: + col_type = getattr(col.type, "python_type", None) + except (NotImplementedError, AttributeError): + col_type = None + + if (col_type is datetime or isinstance(col.type, DateTime)) and isinstance(val, str): with contextlib.suppress(ValueError): val = datetime.fromisoformat(val) + elif (col_type is date or isinstance(col.type, Date)) and isinstance(val, str): + with contextlib.suppress(ValueError): + val = date.fromisoformat(val.split("T")[0]) + elif col_type is uuid.UUID and isinstance(val, str): + with contextlib.suppress(ValueError): + val = uuid.UUID(val) + elif col_type is int and isinstance(val, str): + with contextlib.suppress(ValueError): + val = int(val) + elif col_type is float and isinstance(val, str): + with contextlib.suppress(ValueError): + val = float(val) if effective_cursor_field == pk_name: - query = query.where(col < val if self.order == "desc" else col > val) + query = query.where(col < val if effective_order == "desc" else col > val) else: - if self.order == "desc": + if effective_order == "desc": query = query.where(or_(col < val, and_(col == val, pk_col < last_pk))) else: query = query.where(or_(col > val, and_(col == val, pk_col > last_pk))) + query = query.order_by(None) + if effective_cursor_field == pk_name: - query = query.order_by(col.desc() if self.order == "desc" else col.asc()) + query = query.order_by(col.desc() if effective_order == "desc" else col.asc()) else: - if self.order == "desc": + if effective_order == "desc": query = query.order_by(col.desc(), pk_col.desc()) else: query = query.order_by(col.asc(), pk_col.asc()) diff --git a/src/zcore/db/repository.py b/src/zcore/db/repository.py index 4d99c02..964c599 100644 --- a/src/zcore/db/repository.py +++ b/src/zcore/db/repository.py @@ -3,14 +3,15 @@ This module provides a Repository pattern implementation for SQLAlchemy 2.0 and Pydantic V2. It decouples business logic from database interactions by exposing highly specialized, reusable interfaces divided into read, write, and search capabilities. -It supports dynamic pagination, eager load optimization, field pruning, and full soft-delete lifecycles. +It supports dynamic pagination, eager load optimization, field pruning, composite primary keys, +and full soft-delete lifecycles. """ from collections.abc import Sequence from typing import TYPE_CHECKING, Any, Generic, TypeVar from pydantic import BaseModel -from sqlalchemy import Select, delete, func, insert, inspect, select, update +from sqlalchemy import Select, and_, delete, func, insert, inspect, select, tuple_, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import load_only from sqlalchemy.orm.interfaces import ExecutableOption @@ -34,16 +35,70 @@ class AbstractRepository(Generic[ModelType]): Attributes: db: The active asynchronous database session. model: The SQLAlchemy model class bound to the repository instance. - pk: The SQLAlchemy primary key column representation of the bound model. - pk_name: The string identifier name of the primary key field. - cursor_field: The field name to utilize during keyset pagination lookups. + pks: Tuple of primary key column representations of the bound model. + pk_names: Tuple of primary key column names. + is_composite_pk: Flag indicating if the model has a composite primary key. """ db: AsyncSession model: type[ModelType] - pk: Any - pk_name: str - cursor_field: str + pks: tuple[Any, ...] + pk_names: tuple[str, ...] + is_composite_pk: bool + _cursor_field: str | None = None + + @property + def pk(self) -> Any: + """Convenience property for single primary key models. + + Raises: + AttributeError: If accessed on a composite primary key model. + """ + if self.is_composite_pk: + raise AttributeError( + f"Model '{self.model.__name__}' has a composite primary key {self.pk_names}. " + "Access 'self.pks' instead of 'self.pk'." + ) + return self.pks[0] + + @property + def pk_name(self) -> str: + """Convenience property for single primary key column name. + + Raises: + AttributeError: If accessed on a composite primary key model. + """ + if self.is_composite_pk: + raise AttributeError( + f"Model '{self.model.__name__}' has a composite primary key {self.pk_names}. " + "Access 'self.pk_names' instead of 'self.pk_name'." + ) + return self.pk_names[0] + + @property + def cursor_field(self) -> str: + """Resolve the default cursor coordinate field. + + Returns: + The explicitly configured cursor field or the single primary key column name. + + Raises: + AttributeError: If invoked on a composite model without an explicit cursor field. + """ + if self._cursor_field is not None: + return self._cursor_field + + if self.is_composite_pk: + raise AttributeError( + f"Model '{self.model.__name__}' has a composite primary key {self.pk_names}. " + "Explicitly set 'self.cursor_field' on the repository or provide sorting criteria in requests." + ) + + return self.pk_names[0] + + @cursor_field.setter + def cursor_field(self, value: str | None) -> None: + self._cursor_field = value def _get_base_query(self) -> Select: """Construct the initial select statement for query operations. @@ -77,6 +132,115 @@ def _apply_filters(self, query: Select, *criterion: Any, **filters: Any) -> Sele query = query.filter_by(**filters) return query + def _build_pk_condition(self, target: Any) -> Any: + """Build equality filter condition for single or composite primary key target. + + Args: + target: Model instance, dict mapping column names to values, tuple/list of values, or scalar. + + Returns: + SQLAlchemy binary expression or compound `and_` expression matching target primary key. + + Raises: + ValueError: If target values do not conform to expected primary key structures. + """ + if isinstance(target, self.model): + return and_(*[col == getattr(target, col.key) for col in self.pks]) + + if isinstance(target, dict): + if self.is_composite_pk: + missing = [ + col.key for col in self.pks if col.key not in target and col.name not in target + ] + if missing: + raise ValueError( + f"Incomplete composite primary key mapping for '{self.model.__name__}'. " + f"Missing required key fields: {missing}" + ) + return and_(*[col == target.get(col.key, target.get(col.name)) for col in self.pks]) + + if self.pk_names[0] in target: + return self.pks[0] == target[self.pk_names[0]] + if self.pks[0].key in target: + return self.pks[0] == target[self.pks[0].key] + raise ValueError( + f"Missing required primary key '{self.pk_names[0]}' in mapping for '{self.model.__name__}'." + ) + + if isinstance(target, (tuple, list)): + if self.is_composite_pk: + if len(target) != len(self.pks): + raise ValueError( + f"Composite primary key tuple for '{self.model.__name__}' requires " + f"{len(self.pks)} values ({self.pk_names}), received {len(target)}." + ) + return and_(*[col == val for col, val in zip(self.pks, target, strict=True)]) + return self.pks[0] == target[0] + + if self.is_composite_pk: + raise ValueError( + f"Model '{self.model.__name__}' has composite primary keys {self.pk_names}. " + f"Target lookup must be a model instance, mapping, or tuple/list, received scalar: {target!r}" + ) + + return self.pks[0] == target + + def _build_ids_condition(self, ids: Sequence[Any]) -> Any: + """Construct matching expression for sequences of scalar or composite primary keys. + + Args: + ids: Sequence of scalars, tuples, dicts, or model instances. + + Returns: + SQLAlchemy filter expression resolving scalar `in_` or composite `tuple_.in_`. + + Raises: + ValueError: If items in ids do not conform to the model's primary key signature. + """ + if not self.is_composite_pk: + normalized_scalars = [ + getattr(item, self.pks[0].key) + if isinstance(item, self.model) + else item.get(self.pk_names[0], item.get(self.pks[0].key)) + if isinstance(item, dict) + else item[0] + if isinstance(item, (tuple, list)) + else item + for item in ids + ] + return self.pks[0].in_(normalized_scalars) + + normalized_tuples = [] + for item in ids: + if isinstance(item, self.model): + normalized_tuples.append(tuple(getattr(item, col.key) for col in self.pks)) + elif isinstance(item, dict): + missing = [ + col.key for col in self.pks if col.key not in item and col.name not in item + ] + if missing: + raise ValueError( + f"Incomplete composite key item in batch for '{self.model.__name__}'. " + f"Missing required fields: {missing}" + ) + normalized_tuples.append( + tuple(item.get(col.key, item.get(col.name)) for col in self.pks) + ) + elif isinstance(item, (tuple, list)): + if len(item) != len(self.pks): + raise ValueError( + f"Composite primary key item requires {len(self.pks)} values ({self.pk_names}), " + f"received {len(item)}: {item}." + ) + normalized_tuples.append(tuple(item)) + else: + raise ValueError( + f"Model '{self.model.__name__}' has composite primary keys {self.pk_names}. " + f"Batch items must be tuples, dicts, or model instances, received scalar: {item!r}" + ) + + return tuple_(*self.pks).in_(normalized_tuples) + def _supports_soft_delete(self) -> bool: """Determine whether the underlying model supports soft deletion. @@ -103,7 +267,7 @@ async def exist(self, *criterion: Any, **filters: Any) -> bool: Returns: True if matching records are found, False otherwise. """ - query = select(self.pk) + query = select(*self.pks) scoper = getattr(self.model, "scope_query", None) if scoper: query = scoper(query) @@ -147,7 +311,7 @@ async def get_by_ids( """Fetch a sequence of records matching a list of primary keys. Args: - ids: A list of primary key values to look up. + ids: A list of scalar primary keys or composite key tuples/mappings. fields: Specific entity fields to selectively load. Defaults to None. options: Additional SQLAlchemy executable options. Defaults to None. @@ -157,7 +321,7 @@ async def get_by_ids( if not ids: return [] - query = self._get_base_query().where(self.pk.in_(ids)) + query = self._get_base_query().where(self._build_ids_condition(ids)) if fields: query = query.options(load_only(*fields)) if options: @@ -197,7 +361,10 @@ async def get_list( return result.scalars().all() paginator = ( - CursorPagination(self.cursor_field) + CursorPagination( + cursor_field=getattr(pagination, "cursor_field", None) or self.cursor_field, + order=getattr(pagination, "order", None) or "desc", + ) if isinstance(pagination, CursorParams) else PageNumberPagination() ) @@ -287,10 +454,10 @@ async def update( partial: bool = False, **extra_data: Any, ) -> ModelType | None: - """Update an existing database record from a model instance or primary key. + """Update an existing database record from a model instance, primary key, or composite mapping. Args: - target: The model instance or primary key identifier of the record to update. + target: Model instance, primary key identifier, composite tuple, or composite mapping. schema: The Pydantic update schema containing modified parameters. partial: If True, applies modifications as a partial patch (ignoring unset fields). If False, updates the record using all fields. Defaults to False. @@ -302,7 +469,8 @@ async def update( if isinstance(target, self.model): record = target else: - record = await self.get(**{self.pk_name: target}) + condition = self._build_pk_condition(target) + record = await self.get(condition) if not record: return None @@ -320,10 +488,10 @@ async def update_multi( partial: bool = False, refresh: bool = False, ) -> Sequence[ModelType]: - """Bulk update multiple database records using DBAPI executemany. + """Bulk update multiple database records supporting single or composite keys. Args: - data: A mapping of model instances or primary keys to their update schemas. + data: A mapping of model instances, scalar keys, tuples, or dicts to their update schemas. partial: If True, ignores unset schema fields. Defaults to False. refresh: Parameter maintained for interface consistency. Defaults to False. @@ -336,14 +504,52 @@ async def update_multi( payloads = [] target_ids = [] for key, schema in data.items(): - pk_val = getattr(key, self.pk_name, key) - target_ids.append(pk_val) - payloads.append( - { - **schema.model_dump(exclude_unset=partial), - self.pk_name: pk_val, - } - ) + row_payload = schema.model_dump(exclude_unset=partial) + if isinstance(key, self.model): + target_id = ( + tuple(getattr(key, col.key) for col in self.pks) + if self.is_composite_pk + else getattr(key, self.pk_names[0]) + ) + for col in self.pks: + row_payload[col.key] = getattr(key, col.key) + elif isinstance(key, dict): + if self.is_composite_pk: + missing = [ + col.key for col in self.pks if col.key not in key and col.name not in key + ] + if missing: + raise ValueError( + f"Incomplete composite primary key mapping in update_multi for '{self.model.__name__}'. " + f"Missing fields: {missing}" + ) + target_id = tuple(key.get(col.key, key.get(col.name)) for col in self.pks) + for col in self.pks: + row_payload[col.key] = key.get(col.key, key.get(col.name)) + else: + pk_k = self.pk_names[0] if self.pk_names[0] in key else self.pks[0].key + target_id = key.get(pk_k) + row_payload[self.pk_names[0]] = target_id + elif isinstance(key, (tuple, list)) and self.is_composite_pk: + if len(key) != len(self.pks): + raise ValueError( + f"Composite primary key tuple in update_multi requires {len(self.pks)} values ({self.pk_names}), " + f"received {len(key)}." + ) + target_id = tuple(key) + for col, val in zip(self.pks, key, strict=True): + row_payload[col.key] = val + else: + if self.is_composite_pk: + raise ValueError( + f"Model '{self.model.__name__}' has composite primary keys {self.pk_names}. " + f"Update target must be a tuple, dict, or model instance, received scalar: {key!r}" + ) + target_id = key + row_payload[self.pk_names[0]] = key + + target_ids.append(target_id) + payloads.append(row_payload) await self.db.execute( update(self.model), @@ -354,10 +560,10 @@ async def update_multi( return await self.get_by_ids(ids=target_ids) async def delete(self, target: ModelType | Any, force: bool = False) -> ModelType | None: - """Delete a single record, applying soft deletion if supported unless forced. + """Delete a single record supporting single and composite primary keys. Args: - target: The model instance or primary key identifier of the record to delete. + target: Model instance, primary key identifier, composite tuple, or composite mapping. force: If True, executes a permanent physical hard deletion. Defaults to False. Returns: @@ -366,12 +572,13 @@ async def delete(self, target: ModelType | Any, force: bool = False) -> ModelTyp if isinstance(target, self.model): record = target else: + condition = self._build_pk_condition(target) if force: - query = select(self.model).where(getattr(self.model, self.pk_name) == target) + query = select(self.model).where(condition) result = await self.db.execute(query) record = result.scalars().first() else: - record = await self.get(**{self.pk_name: target}) + record = await self.get(condition) if not record: return None @@ -387,10 +594,10 @@ async def delete(self, target: ModelType | Any, force: bool = False) -> ModelTyp return record async def delete_multi(self, ids: list[Any], force: bool = False) -> Sequence[ModelType]: - """Delete multiple records by primary keys, using atomic soft update or hard delete. + """Delete multiple records by primary keys supporting single and composite entities. Args: - ids: A list of primary key values of records to delete. + ids: List of primary keys, composite tuples, mappings, or model instances. force: If True, executes permanent physical hard deletions. Defaults to False. Returns: @@ -401,13 +608,14 @@ async def delete_multi(self, ids: list[Any], force: bool = False) -> Sequence[Mo dialect = getattr(getattr(self.db, "bind", None), "dialect", None) supports_returning = bool(getattr(dialect, "delete_returning", False)) + ids_condition = self._build_ids_condition(ids) if not force and self._supports_soft_delete(): supports_update_returning = bool(getattr(dialect, "update_returning", False)) deletion_time = now() stmt = ( update(self.model) - .where(self.pk.in_(ids), self.model.deleted_at.is_(None)) + .where(ids_condition, self.model.deleted_at.is_(None)) .values(deleted_at=deletion_time) ) @@ -419,30 +627,30 @@ async def delete_multi(self, ids: list[Any], force: bool = False) -> Sequence[Mo await self.db.execute(stmt) await self.db.flush() - query = select(self.model).where(self.pk.in_(ids)) + query = select(self.model).where(ids_condition) res = await self.db.execute(query) return list(res.scalars().all()) if supports_returning: - stmt = delete(self.model).where(self.pk.in_(ids)).returning(self.model) + stmt = delete(self.model).where(ids_condition).returning(self.model) result = await self.db.scalars(stmt) await self.db.flush() return list(result.all()) else: - query = select(self.model).where(self.pk.in_(ids)) + query = select(self.model).where(ids_condition) records_res = await self.db.execute(query) records = list(records_res.scalars().all()) if records: - stmt = delete(self.model).where(self.pk.in_(ids)) + stmt = delete(self.model).where(ids_condition) await self.db.execute(stmt) await self.db.flush() return records async def restore(self, target: ModelType | Any) -> ModelType | None: - """Restore a previously soft-deleted record. + """Restore a previously soft-deleted record supporting single and composite entities. Args: - target: The model instance or primary key identifier of the record to restore. + target: Model instance, primary key identifier, composite tuple, or composite mapping. Returns: The restored database model instance, or None if the record was not found or unsupported. @@ -453,7 +661,8 @@ async def restore(self, target: ModelType | Any) -> ModelType | None: if isinstance(target, self.model): record = target else: - query = select(self.model).where(getattr(self.model, self.pk_name) == target) + condition = self._build_pk_condition(target) + query = select(self.model).where(condition) result = await self.db.execute(query) record = result.scalars().first() if not record: @@ -465,10 +674,10 @@ async def restore(self, target: ModelType | Any) -> ModelType | None: return record async def restore_multi(self, ids: list[Any]) -> Sequence[ModelType]: - """Restore multiple soft-deleted records using an atomic batch update. + """Restore multiple soft-deleted records supporting single and composite entities. Args: - ids: A list of primary key values of records to restore. + ids: List of primary keys, composite tuples, mappings, or model instances. Returns: A sequence of restored database model instances. @@ -478,10 +687,11 @@ async def restore_multi(self, ids: list[Any]) -> Sequence[ModelType]: dialect = getattr(getattr(self.db, "bind", None), "dialect", None) supports_update_returning = bool(getattr(dialect, "update_returning", False)) + ids_condition = self._build_ids_condition(ids) stmt = ( update(self.model) - .where(self.pk.in_(ids), self.model.deleted_at.is_not(None)) + .where(ids_condition, self.model.deleted_at.is_not(None)) .values(deleted_at=None) ) @@ -493,7 +703,7 @@ async def restore_multi(self, ids: list[Any]) -> Sequence[ModelType]: await self.db.execute(stmt) await self.db.flush() - query = select(self.model).where(self.pk.in_(ids)) + query = select(self.model).where(ids_condition) res = await self.db.execute(query) return list(res.scalars().all()) @@ -540,11 +750,16 @@ async def search( result = await self.db.execute(query) return result.scalars().all() - paginator = ( - CursorPagination(self.cursor_field) - if isinstance(pagination, CursorParams) - else PageNumberPagination() - ) + if isinstance(pagination, CursorParams): + primary_sort = search_in.sort[0] if search_in.sort else None + sort_field = pagination.cursor_field or ( + primary_sort.field if primary_sort else self.cursor_field + ) + sort_order = pagination.order or (primary_sort.order if primary_sort else "desc") + paginator = CursorPagination(cursor_field=sort_field, order=sort_order) + else: + paginator = PageNumberPagination() + return await paginator.paginate(self.db, query, pagination, self.model) @@ -557,7 +772,7 @@ class BaseRepository( """The default implementation combining Read, Write, and Search capabilities.""" def __init__(self, model: type[ModelType], db: AsyncSession): - """Initialize the BaseRepository. + """Initialize the BaseRepository supporting single and composite primary keys. Args: model: The SQLAlchemy declarative model class linked to this repository. @@ -565,6 +780,8 @@ def __init__(self, model: type[ModelType], db: AsyncSession): """ self.model = model self.db = db - self.pk = inspect(self.model).primary_key[0] - self.pk_name = self.pk.name - self.cursor_field = self.pk_name + mapper = inspect(self.model) + self.pks = tuple(mapper.primary_key) + self.pk_names = tuple(col.name for col in self.pks) + self.is_composite_pk = len(self.pks) > 1 + self._cursor_field = None diff --git a/src/zcore/db/search.py b/src/zcore/db/search.py index 4cac0b0..d958158 100644 --- a/src/zcore/db/search.py +++ b/src/zcore/db/search.py @@ -109,6 +109,11 @@ class SearchRequest(BaseModel): @model_validator(mode="after") def _validate_and_bound_size(self) -> SearchRequest: + if self.cursor and self.cursor.strip() and self.page > 1: + raise ValueError( + "Cannot combine 'cursor' with 'page' > 1. Choose either keyset (cursor) or offset (page) pagination." + ) + default_size = getattr(settings, "PAGINATION_DEFAULT_SIZE", 20) max_size = getattr(settings, "PAGINATION_MAX_SIZE", 100) diff --git a/src/zcore/kernel/di.py b/src/zcore/kernel/di.py index aa3192a..c085c25 100644 --- a/src/zcore/kernel/di.py +++ b/src/zcore/kernel/di.py @@ -8,6 +8,7 @@ import functools import inspect +import types import uuid from collections.abc import AsyncGenerator, Callable from contextlib import asynccontextmanager @@ -48,6 +49,14 @@ class CircularDependencyError(DIException): pass +class _DependencySignature(list[type[Any]]): + """Signature mapping preserving list-like inspection while caching parameter names.""" + + def __init__(self, kwargs_map: dict[str, type[Any]]) -> None: + super().__init__(kwargs_map.values()) + self.kwargs_map: dict[str, type[Any]] = kwargs_map + + class IoCContainer: """Central Inversion of Control (IoC) container for dependency management. @@ -72,6 +81,50 @@ def __init__(self) -> None: self._constructor_cache: dict[type[Any], Callable[..., Any] | None] = {} self._dependency_signature_cache: dict[type[Any], list[type[Any]]] = {} + def is_registered(self, interface: type[Any]) -> bool: + """Verify if a given interface or class is registered across any lifecycle store. + + Args: + interface: The interface or class type to inspect. + + Returns: + True if the interface has an active registration, False otherwise. + """ + unwrapped = self._unwrap_type(interface) + current_instances = _scoped_instances.get() + in_active_scope = current_instances is not None and unwrapped in current_instances + return ( + unwrapped in self._singletons + or unwrapped in self._scoped_definitions + or unwrapped in self._factories + or unwrapped is AsyncSession + or in_active_scope + ) + + @staticmethod + def _unwrap_type(annotation: Any) -> Any: + """Recursively unwrap Annotated, Optional, and Union types to extract the target type. + + Args: + annotation: Raw type annotation to normalize. + + Returns: + The extracted concrete dependency type. + """ + origin = get_origin(annotation) + if origin is Annotated: + return IoCContainer._unwrap_type(get_args(annotation)[0]) + + is_union = origin is not None and ( + origin is types.UnionType or str(origin) in ("typing.Union", "") + ) + if is_union: + args = [arg for arg in get_args(annotation) if arg is not type(None)] + if len(args) == 1: + return IoCContainer._unwrap_type(args[0]) + + return annotation + def register_singleton(self, interface: type[Any], instance: Any) -> None: """Register a pre-constructed instance as a global singleton. @@ -138,27 +191,36 @@ def resolve(self, interface: type[T], _stack: set[type[Any]] | None = None) -> T The fully constructed instance of type `T`. Raises: + DIException: If a scoped dependency is requested outside an active scope. CircularDependencyError: If a cyclic loop is detected during dependency tree assembly. """ - if interface in self._singletons: - return self._singletons[interface] + target_type = self._unwrap_type(interface) + + if target_type in self._singletons: + return self._singletons[target_type] scope_id = _current_scope_id.get() if scope_id is not None: current_instances = _scoped_instances.get() - if current_instances is not None and interface in current_instances: - return current_instances[interface] + if current_instances is not None and target_type in current_instances: + return current_instances[target_type] - if interface in self._scoped_definitions: - resolved_instance = self._scoped_definitions[interface](_stack) + if target_type in self._scoped_definitions: + resolved_instance = self._scoped_definitions[target_type](_stack) if current_instances is not None: - current_instances[interface] = resolved_instance + current_instances[target_type] = resolved_instance return resolved_instance + else: + if target_type in self._scoped_definitions or target_type is AsyncSession: + type_name = getattr(target_type, "__name__", str(target_type)) + raise DIException( + f"Cannot resolve scoped dependency '{type_name}' outside of an active scope." + ) - if interface in self._factories: - return self._factories[interface](_stack) + if target_type in self._factories: + return self._factories[target_type](_stack) - return self._auto_wire(interface, _stack) + return self._auto_wire(target_type, _stack) def _auto_wire(self, target_class: type[T], _stack: set[type[Any]] | None = None) -> T: """Analyze, resolve parameters, and construct a class instance. @@ -189,8 +251,14 @@ def _auto_wire(self, target_class: type[T], _stack: set[type[Any]] | None = None try: if target_class in self._dependency_signature_cache: - dependencies = self._dependency_signature_cache[target_class] - resolved_args = [self.resolve(dep, _stack.copy()) for dep in dependencies] + cached_sig = self._dependency_signature_cache[target_class] + if isinstance(cached_sig, _DependencySignature): + resolved_kwargs = { + name: self.resolve(dep, _stack.copy()) + for name, dep in cached_sig.kwargs_map.items() + } + return target_class(**resolved_kwargs) + resolved_args = [self.resolve(dep, _stack.copy()) for dep in cached_sig] return target_class(*resolved_args) if target_class not in self._constructor_cache: @@ -200,7 +268,7 @@ def _auto_wire(self, target_class: type[T], _stack: set[type[Any]] | None = None constructor = self._constructor_cache[target_class] if constructor is None or constructor is object.__init__: - self._dependency_signature_cache[target_class] = [] + self._dependency_signature_cache[target_class] = _DependencySignature({}) return target_class() try: @@ -209,24 +277,33 @@ def _auto_wire(self, target_class: type[T], _stack: set[type[Any]] | None = None type_hints = {} sig = inspect.signature(constructor) - dependencies = [] + dependencies_map: dict[str, type[Any]] = {} for name, param in sig.parameters.items(): if name == "self": continue - annotation = type_hints.get(name, param.annotation) - if annotation is inspect.Parameter.empty: + if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD): + continue + + raw_annotation = type_hints.get(name, param.annotation) + if raw_annotation is inspect.Parameter.empty: continue - if get_origin(annotation) is Annotated: - annotation = get_args(annotation)[0] + annotation = self._unwrap_type(raw_annotation) + has_default = param.default is not inspect.Parameter.empty + + if has_default and not self.is_registered(annotation): + continue - dependencies.append(annotation) + dependencies_map[name] = annotation - self._dependency_signature_cache[target_class] = dependencies + sig_entry = _DependencySignature(dependencies_map) + self._dependency_signature_cache[target_class] = sig_entry - resolved_args = [self.resolve(dep, _stack.copy()) for dep in dependencies] - return target_class(*resolved_args) + resolved_kwargs = { + name: self.resolve(dep, _stack.copy()) for name, dep in sig_entry.kwargs_map.items() + } + return target_class(**resolved_kwargs) finally: _stack.remove(target_class) @@ -380,9 +457,7 @@ def _resolve_injections(args: tuple[Any, ...], kwargs: dict[str, Any]) -> dict[s if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD): continue if param.annotation is not inspect.Parameter.empty: - annotation = param.annotation - if get_origin(annotation) is Annotated: - annotation = get_args(annotation)[0] + annotation = container._unwrap_type(param.annotation) resolved_kwargs[param_name] = container.resolve(annotation) return resolved_kwargs diff --git a/src/zcore/testing/fixtures.py b/src/zcore/testing/fixtures.py index 408c57e..c39ce2a 100644 --- a/src/zcore/testing/fixtures.py +++ b/src/zcore/testing/fixtures.py @@ -234,14 +234,20 @@ def __init__(self, app: FastAPI, stub: Any, override_func: Any) -> None: self.app = app self.stub = stub self.override_func = override_func + self._had_override: bool = False + self._original_override: Any = None async def setUp(self) -> None: """Register the dependency substitution.""" + self._had_override = self.stub in self.app.dependency_overrides + self._original_override = self.app.dependency_overrides.get(self.stub) self.app.dependency_overrides[self.stub] = self.override_func async def tearDown(self) -> None: - """Remove the dependency substitution.""" - if self.stub in self.app.dependency_overrides: + """Restore the dependency substitution to its previous state.""" + if self._had_override: + self.app.dependency_overrides[self.stub] = self._original_override + elif self.stub in self.app.dependency_overrides: del self.app.dependency_overrides[self.stub] diff --git a/src/zcore/web/api_router.py b/src/zcore/web/api_router.py index 23a0e5c..61f5b17 100644 --- a/src/zcore/web/api_router.py +++ b/src/zcore/web/api_router.py @@ -1,21 +1,19 @@ """Custom API Route and Serialization. This module provides specialized request, response, and routing behaviors for the ZCore -web layer. It delegates dynamic schema pruning and input validation to the bock-end +web layer. It delegates dynamic schema pruning and input validation to the back-end Zchema infrastructure. """ from collections.abc import Callable, Coroutine -from typing import Any, get_args, get_origin +from typing import Any from fastapi import Request, Response from fastapi.responses import JSONResponse from fastapi.routing import APIRoute -from pydantic import BaseModel from zcore.context.context import ctx from zcore.utils.helpers import json_dumps -from zcore.web.response import ResponseWrapper class ZCoreRequest(Request): @@ -50,58 +48,17 @@ def render(self, content: Any) -> bytes: return json_dumps(content).encode("utf-8") -def find_input_schema(dependant: Any) -> type[BaseModel] | None: - """Analyze FastAPI dependency models to find the body input Pydantic schema.""" - for param in getattr(dependant, "body_params", []): - annotation = getattr(param.field_info, "annotation", None) or getattr(param, "type_", None) - if isinstance(annotation, type) and issubclass(annotation, BaseModel): - return annotation - return None - - -def find_output_schema(response_model: Any) -> type[BaseModel] | None: - """Recursively analyze route response models to extract the target Pydantic schema.""" - if response_model is None: - return None - origin = get_origin(response_model) - args = get_args(response_model) - if origin is None: - if isinstance(response_model, type) and issubclass(response_model, BaseModel): - return response_model - return None - for arg in args: - schema = find_output_schema(arg) - if schema: - return schema - return None - - class ZCoreAPIRoute(APIRoute): """Custom API Route extending standard FastAPI route executions. - Configures routes to use the `ZCoreJSONResponse` class and dynamically generates - security-pruned schema definitions on demand when `?schema=true` is requested. + Configures routes to use the `ZCoreJSONResponse` class, dynamically binds + operational action context, and manages response caching headers based on + security restrictions. """ - expose_schema: bool - target_model: type[BaseModel] | None - def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) - self.expose_schema = False - if self.openapi_extra and isinstance(self.openapi_extra, dict): - if self.openapi_extra.get("expose_schema"): - self.expose_schema = True - - self.target_model = None - - if self.expose_schema: - if self.methods and any(m in ["POST", "PUT", "PATCH"] for m in self.methods): - self.target_model = find_input_schema(self.dependant) - else: - self.target_model = find_output_schema(self.response_model) - current_class = self.response_class if hasattr(current_class, "value"): current_class = current_class.value @@ -132,27 +89,6 @@ async def custom_route_handler(request: Request) -> Response: ctx.set("action", custom_action) - if self.expose_schema and zcore_request.query_params.get("schema") == "true": - if not self.target_model: - return JSONResponse( - status_code=400, - content={ - "success": False, - "message": "No schema defined for this endpoint.", - }, - ) - - # Dynamic generation dynamically invokes __get_pydantic_json_schema__ under active context - schema_dict = self.target_model.model_json_schema() - - response_payload: Any = ResponseWrapper( - success=True, - message="Schema generated successfully", - data=schema_dict, - meta={"restricted_fields": list(ctx.restricted_fields)}, - ) - return JSONResponse(content=response_payload.model_dump()) - response: Response = await original_route_handler(zcore_request) hidden_fields = ctx.restricted_fields diff --git a/src/zcore/web/base_router.py b/src/zcore/web/base_router.py index 338ba18..85bafc8 100644 --- a/src/zcore/web/base_router.py +++ b/src/zcore/web/base_router.py @@ -1,9 +1,11 @@ """Automated Web Router Scaffolding. This module provides the generic `BaseRouter` interface, which scaffolds -standard security-aware CRUD and lookup endpoints (POST, GET, GET_ALL, SEARCH, LOOKUP, UPDATE, PATCH, DELETE) +standard security-aware CRUD, lookup, and schema introspection endpoints +(POST, GET, GET_ALL, SEARCH, LOOKUP, UPDATE, PATCH, DELETE, SCHEMA) and integrates them with services, schemas, dependency requirements, and pagination handlers, -with clean declarative syntax, primary key auto-detection, and weighted route specificity sorting. +with clean declarative syntax, primary key auto-detection, optional authentication, +and weighted route specificity sorting. """ import uuid @@ -18,9 +20,11 @@ from sqlalchemy.orm import joinedload, selectinload from zcore.config import settings +from zcore.context.context import ctx from zcore.db.search import SearchRequest from zcore.exceptions.base import ValidationError from zcore.kernel.di import Injector +from zcore.security.dependencies import get_optional_user_stub from zcore.security.permissions import HasScopes from zcore.service.base import BaseService from zcore.web.api_router import ZCoreAPIRoute @@ -44,14 +48,15 @@ class RouteKey(StrEnum): UPDATE = "UPDATE" PATCH = "PATCH" DELETE = "DELETE" + SCHEMA = "SCHEMA" class BaseRouter(Generic[CreateSchemaType, UpdateSchemaType]): """Declarative web router orchestrator. - Automatically maps CRUD and lookup operations to matching database model dependencies and handles - dependency injections, schema validations, dynamic primary key route resolution, - and hierarchical route ordering based on path specificity. + Automatically maps CRUD, lookup, and schema introspection operations to matching database + model dependencies and handles dependency injections, schema validations, dynamic primary + key route resolution, and hierarchical route ordering based on path specificity. Attributes: model: The database declarative model class. @@ -66,11 +71,11 @@ class BaseRouter(Generic[CreateSchemaType, UpdateSchemaType]): service: Business service callable class. pk_type: Optional explicit primary key type. If None, auto-resolved from model metadata. prefix: Path prefix representing the route. + schema_path: Sub-path reserved for resource schema introspection. Defaults to "/_schema". tags: Endpoint group classification tags. exclude: Explicit endpoints to bypass during route scaffolding. pagination_class: Pagination engine class to construct list queries. route_class: Custom routing processing class. Defaults to ZCoreAPIRoute. - expose_schemas: Exposes target endpoint schemas dynamically. """ model: type[Any] @@ -86,17 +91,17 @@ class BaseRouter(Generic[CreateSchemaType, UpdateSchemaType]): pk_type: type[Any] | None = None prefix: str = "" + schema_path: str = "/_schema" tags: list[str] | None = None exclude: set[RouteKey] | None = None pagination_class: type["BasePagination"] | None = None route_class: type[APIRoute] = ZCoreAPIRoute - expose_schemas: set[RouteKey] | bool = False - def __init__(self) -> None: """Initialize the BaseRouter. - Performs fail-fast configuration checks and registers configured CRUD endpoints. + Performs fail-fast configuration checks, registers configured CRUD endpoints, + and binds auto-sorting hooks for subsequent custom routes. Raises: ValueError: If the required service parameter is not configured. @@ -110,6 +115,25 @@ def __init__(self) -> None: self.exclude = self.exclude or set() self._validate_schema_configurations() self._register_routes() + self._bind_router_sorting_hooks() + + def _bind_router_sorting_hooks(self) -> None: + """Bind hooks to auto-sort routes whenever new endpoints are registered.""" + original_add_api_route = self.router.add_api_route + original_add_route = self.router.add_route + + def wrapped_add_api_route(*args: Any, **kwargs: Any) -> Any: + result = original_add_api_route(*args, **kwargs) + self._sort_routes() + return result + + def wrapped_add_route(*args: Any, **kwargs: Any) -> Any: + result = original_add_route(*args, **kwargs) + self._sort_routes() + return result + + self.router.add_api_route = wrapped_add_api_route # type: ignore[method-assign] + self.router.add_route = wrapped_add_route # type: ignore[method-assign] def _resolve_pk_type(self) -> type[Any]: """Resolve the effective primary key type from explicit definition or model reflection. @@ -172,24 +196,6 @@ def _validate_schema_configurations(self) -> None: f"Model class must be defined in '{self.__class__.__name__}' to resolve route actions." ) - def _get_openapi_extra(self, route_key: RouteKey) -> dict[str, Any] | None: - """Construct OpenAPI specifications for dynamic schema endpoints. - - Args: - route_key: Target operational key to check. - - Returns: - An openapi metadata dictionary, or None. - """ - if isinstance(self.expose_schemas, bool): - expose = self.expose_schemas - else: - expose = route_key in (self.expose_schemas or set()) - - if expose: - return {"expose_schema": True} - return None - def _normalize_dependencies(self, raw_deps: list[Any] | Any) -> list[DependsClass]: """Normalize raw classes or parameters into FastAPI Depends structures. @@ -220,7 +226,7 @@ def get_route_action(self, route_key: RouteKey) -> str: route_key: The target operational key. Returns: - The calculated permission action identifier string. + The calculated permission action identifier string, or an empty string. """ if not self.model: raise ValueError( @@ -237,10 +243,10 @@ def get_route_action(self, route_key: RouteKey) -> str: RouteKey.PATCH: self.model.actions().UPDATE, RouteKey.DELETE: self.model.actions().DELETE, } - return action_map[route_key] + return action_map.get(route_key, "") def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: - """Generate default route dependencies. + """Generate route dependencies supporting mandatory or optional authorization. Args: route_key: The target operational route key. @@ -249,6 +255,9 @@ def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: Returns: A list of dependencies. """ + if route_key == RouteKey.SCHEMA: + return [get_optional_user_stub] + if not action: return [] return [HasScopes(action)] @@ -366,6 +375,10 @@ def _check_filters(items: list[Any] | None) -> None: message=f"Sort field '{s.field}' is not permitted in lookup queries." ) + def sort_routes(self) -> None: + """Sort routes using hierarchical specificity scoring to avoid path shadowing.""" + self._sort_routes() + def _sort_routes(self) -> None: """Sort routes using hierarchical specificity scoring to avoid path shadowing.""" @@ -389,6 +402,10 @@ def _route_specificity_key(route: Any) -> tuple[int, int, list[int], int]: self.router.routes.sort(key=_route_specificity_key) + def register_custom_routes(self) -> None: + """Hook for subclasses to register custom domain endpoints before initial route sorting.""" + pass + def _register_routes(self) -> None: """Dynamically generate and bind endpoints to the APIRouter.""" service_callable = self.service @@ -396,6 +413,22 @@ def _register_routes(self) -> None: target_pk_type = self._resolve_pk_type() pk_path = self._get_pk_path(target_pk_type) + if RouteKey.SCHEMA not in self.exclude: + + async def _schema_endpoint( + action: str | None = None, + ) -> ResponseWrapper: + return await self.schema_endpoint(action) + + self.router.add_api_route( + path=self.schema_path, + endpoint=_schema_endpoint, + methods=["GET"], + dependencies=self._get_route_dependencies(RouteKey.SCHEMA), + status_code=status.HTTP_200_OK, + response_model=ResponseWrapper, + ) + if RouteKey.POST not in self.exclude: c_schema = self.create_schema @@ -414,7 +447,7 @@ async def _create_endpoint( response_model=ResponseWrapper[self.schema_out] if self.schema_out else ResponseWrapper, - openapi_extra=self._get_openapi_extra(RouteKey.POST), + openapi_extra={"action": "create"}, ) if RouteKey.GET not in self.exclude: @@ -434,7 +467,7 @@ async def _get_endpoint( response_model=ResponseWrapper[self.schema_out] if self.schema_out else ResponseWrapper, - openapi_extra=self._get_openapi_extra(RouteKey.GET), + openapi_extra={"action": "view"}, ) if RouteKey.GET_ALL not in self.exclude: @@ -462,7 +495,7 @@ async def _get_all_endpoint( response_model=ResponseWrapper[list[self.schema_out]] if self.schema_out else ResponseWrapper, - openapi_extra=self._get_openapi_extra(RouteKey.GET_ALL), + openapi_extra={"action": "listview"}, ) if RouteKey.SEARCH not in self.exclude: @@ -482,7 +515,7 @@ async def _search_endpoint( response_model=ResponseWrapper[list[self.schema_out]] if self.schema_out else ResponseWrapper, - openapi_extra=self._get_openapi_extra(RouteKey.SEARCH), + openapi_extra={"action": "listview"}, ) if RouteKey.LOOKUP not in self.exclude and self.lookup_schema is not None: @@ -500,7 +533,7 @@ async def _lookup_endpoint( dependencies=self._get_route_dependencies(RouteKey.LOOKUP), status_code=status.HTTP_200_OK, response_model=ResponseWrapper[list[self.lookup_schema]], - openapi_extra=self._get_openapi_extra(RouteKey.LOOKUP), + openapi_extra={"action": "lookup"}, ) if RouteKey.UPDATE not in self.exclude: @@ -522,7 +555,7 @@ async def _update_endpoint( response_model=ResponseWrapper[self.schema_out] if self.schema_out else ResponseWrapper, - openapi_extra=self._get_openapi_extra(RouteKey.UPDATE), + openapi_extra={"action": "update"}, ) if RouteKey.PATCH not in self.exclude: @@ -544,7 +577,7 @@ async def _patch_endpoint( response_model=ResponseWrapper[self.schema_out] if self.schema_out else ResponseWrapper, - openapi_extra=self._get_openapi_extra(RouteKey.PATCH), + openapi_extra={"action": "update"}, ) if RouteKey.DELETE not in self.exclude: @@ -563,11 +596,68 @@ async def _delete_endpoint( dependencies=self._get_route_dependencies(RouteKey.DELETE), status_code=status.HTTP_200_OK, response_model=ResponseWrapper, - openapi_extra=self._get_openapi_extra(RouteKey.DELETE), + openapi_extra={"action": "delete"}, ) + self.register_custom_routes() self._sort_routes() + async def schema_endpoint(self, action: str | None = None) -> ResponseWrapper: + """Resolve dynamic model JSON schemas pruned according to active security context. + + Args: + action: Specific action schema identifier ('create', 'update', 'view', 'lookup'). + If omitted, returns all available schemas for this resource. + + Returns: + ResponseWrapper containing the resolved schema or schema mapping. + + Raises: + ValidationError: If an invalid action or unconfigured schema is requested. + """ + schemas: dict[str, type[BaseModel] | None] = { + "create": self.create_schema, + "update": self.update_schema, + "view": self.schema_out, + "lookup": self.lookup_schema, + } + + if action: + normalized_action = action.lower().strip() + if normalized_action not in schemas: + allowed = ", ".join(sorted(schemas.keys())) + raise ValidationError( + message=f"Invalid schema action '{action}'. Allowed actions are: {allowed}" + ) + + target_schema = schemas[normalized_action] + if target_schema is None: + raise ValidationError( + message=f"No schema configured for action '{action}' on {self.__class__.__name__}." + ) + + with ctx.scope(action=normalized_action): + schema_data = target_schema.model_json_schema() + + return ResponseWrapper( + data=schema_data, + meta={ + "action": normalized_action, + "restricted_fields": list(ctx.restricted_fields), + }, + ) + + resolved_schemas: dict[str, Any] = {} + for act_name, schema_cls in schemas.items(): + if schema_cls is not None: + with ctx.scope(action=act_name): + resolved_schemas[act_name] = schema_cls.model_json_schema() + + return ResponseWrapper( + data=resolved_schemas, + meta={"restricted_fields": list(ctx.restricted_fields)}, + ) + async def create_endpoint(self, data_in: Any, service: BaseService) -> ResponseWrapper: """Execute the POST creation transaction. @@ -637,7 +727,14 @@ async def search_endpoint( effective_size = min(search_in.size or default_size, max_size) if self.pagination_class.params_class == CursorParams: - pagination = CursorParams(cursor=search_in.cursor, size=effective_size) + sort_field = search_in.sort[0].field if search_in.sort else None + sort_order = search_in.sort[0].order if search_in.sort else None + pagination = CursorParams( + cursor=search_in.cursor, + size=effective_size, + cursor_field=sort_field, + order=sort_order, + ) else: pagination = PageNumberParams(page=search_in.page, size=effective_size) @@ -679,8 +776,16 @@ async def lookup_endpoint( if self.pagination_class: from zcore.db.pagination import CursorParams, PageNumberParams + sort_field = search_in.sort[0].field if search_in.sort else None + sort_order = search_in.sort[0].order if search_in.sort else None + if self.pagination_class.params_class == CursorParams: - pagination = CursorParams(cursor=search_in.cursor, size=search_in.size) + pagination = CursorParams( + cursor=search_in.cursor, + size=search_in.size, + cursor_field=sort_field, + order=sort_order, + ) else: pagination = PageNumberParams(page=search_in.page, size=search_in.size) diff --git a/tests/test_cache.py b/tests/test_cache.py index 78f113c..f9c5ff8 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -1,3 +1,4 @@ +import asyncio import gc import threading import time @@ -321,3 +322,42 @@ def test_dynamic_cache_configuration_bindings() -> None: base_cache = BaseCache[str](prefix="dynamic") assert base_cache._local_cache.maxsize == 42 assert base_cache.default_ttl == 123 + + +@pytest.mark.anyio +async def test_ensure_eviction_task_concurrent_thread_safety( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import concurrent.futures + + await close_cache() + monkeypatch.setattr(base_module, "_eviction_task", None) + + created_tasks = [] + original_create_task = asyncio.get_running_loop().create_task + + def tracked_create_task(coro: Any) -> Any: + task = original_create_task(coro) + created_tasks.append(task) + return task + + loop = asyncio.get_running_loop() + monkeypatch.setattr(loop, "create_task", tracked_create_task) + + num_threads = 20 + barrier = threading.Barrier(num_threads) + + def worker() -> None: + barrier.wait() + _ensure_eviction_task() + + with ( + patch("asyncio.get_running_loop", return_value=loop), + concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor, + ): + futures = [executor.submit(worker) for _ in range(num_threads)] + concurrent.futures.wait(futures) + + assert len(created_tasks) == 1 + assert base_module._eviction_task is created_tasks[0] + await close_cache() diff --git a/tests/test_cli.py b/tests/test_cli.py index a740b5a..8693b9c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -337,11 +337,11 @@ def test_cli_version_output_rc2() -> None: with pytest.raises(SystemExit) as exc_info: main() assert exc_info.value.code == 0 - assert "0.1.0-rc.2" in capture.get() + assert "0.1.0-rc.3" in capture.get() with console.capture() as capture: print_banner() - assert "0.1.0-rc.2" in capture.get() + assert "0.1.0-rc.3" in capture.get() def test_cli_scaffold_templates_use_database_settings_config(run_in_tmp_path: Path) -> None: diff --git a/tests/test_di.py b/tests/test_di.py index 619c863..f31234e 100644 --- a/tests/test_di.py +++ b/tests/test_di.py @@ -330,6 +330,73 @@ def test_constructor_annotated_parameters() -> None: assert isinstance(resolved.dep, ForwardB) +class ClientWithPrimitiveDefaults: + def __init__(self, timeout: int = 30, client: Any = None) -> None: + self.timeout = timeout + self.client = client + + +class ServiceWithDepAndDefaults: + def __init__(self, dep: ForwardB, timeout: int = 30, label: str = "default") -> None: + self.dep = dep + self.timeout = timeout + self.label = label + + +class ServiceWithRegisteredDefault: + def __init__(self, service: IService | None = None) -> None: + self.service = service + + +def test_auto_wire_skips_unregistered_default_parameters() -> None: + resolved = container.resolve(ClientWithPrimitiveDefaults) + assert isinstance(resolved, ClientWithPrimitiveDefaults) + assert resolved.timeout == 30 + assert resolved.client is None + assert container._dependency_signature_cache[ClientWithPrimitiveDefaults] == [] + + +def test_auto_wire_resolves_required_and_skips_unregistered_defaults() -> None: + resolved = container.resolve(ServiceWithDepAndDefaults) + assert isinstance(resolved, ServiceWithDepAndDefaults) + assert isinstance(resolved.dep, ForwardB) + assert resolved.timeout == 30 + assert resolved.label == "default" + assert container._dependency_signature_cache[ServiceWithDepAndDefaults] == [ForwardB] + + +def test_auto_wire_injects_registered_dependency_even_with_default() -> None: + class ServiceWithRegisteredClassDefault: + def __init__(self, service: IService = ServiceImpl()) -> None: + self.service = service + + registered = AnotherServiceImpl() + container.register_singleton(IService, registered) + resolved = container.resolve(ServiceWithRegisteredClassDefault) + assert resolved.service is registered + + +def test_auto_wire_injects_registered_dependency_with_optional_union() -> None: + registered = AnotherServiceImpl() + container.register_singleton(IService, registered) + resolved = container.resolve(ServiceWithRegisteredDefault) + assert resolved.service is registered + + +def test_auto_wire_skips_unregistered_optional_union() -> None: + resolved = container.resolve(ServiceWithRegisteredDefault) + assert resolved.service is None + + +def test_resolve_scoped_outside_scope_raises_di_exception() -> None: + container.register_scoped(IService, ServiceImpl) + assert _current_scope_id.get() is None + with pytest.raises(DIException, match="outside of an active scope"): + container.resolve(IService) + with pytest.raises(DIException, match="outside of an active scope"): + container.resolve(AsyncSession) + + def test_clear_scope_explicit() -> None: container.register_scoped(IService, ServiceImpl) scope_id = "test_clear" diff --git a/tests/test_repository.py b/tests/test_repository.py index 6cc6d6e..961882d 100644 --- a/tests/test_repository.py +++ b/tests/test_repository.py @@ -1,12 +1,13 @@ import uuid from collections.abc import AsyncGenerator +from datetime import date from typing import Any from unittest.mock import AsyncMock, patch import pytest import pytest_asyncio from pydantic import BaseModel -from sqlalchemy import Column, Integer, String +from sqlalchemy import Column, Date, Integer, String from sqlalchemy.orm import load_only from zcore.context.context import ctx @@ -40,11 +41,48 @@ class RepoTestUpdateSchema(BaseModel): description: str | None = None +class DateCursorModel(Base): + __tablename__ = f"date_cursor_model_{uuid.uuid4().hex[:6]}" + id = Column(Integer, primary_key=True, autoincrement=True) + operation_date = Column(Date, nullable=False) + + +class DateCursorCreateSchema(BaseModel): + operation_date: date + + +class DateCursorRepository(BaseRepository[DateCursorModel]): + def __init__(self, db: Any) -> None: + super().__init__(DateCursorModel, db) + + class RepoTestRepository(BaseRepository[RepoTestModel]): def __init__(self, db: Any) -> None: super().__init__(RepoTestModel, db) +class CompositePartitionModel(Base): + __tablename__ = f"composite_part_model_{uuid.uuid4().hex[:6]}" + id = Column(Integer, primary_key=True) + operation_date = Column(Date, primary_key=True) + message = Column(String, nullable=False) + + +class CompositeCreateSchema(BaseModel): + id: int + operation_date: date + message: str + + +class CompositeUpdateSchema(BaseModel): + message: str + + +class CompositeRepository(BaseRepository[CompositePartitionModel]): + def __init__(self, db: Any) -> None: + super().__init__(CompositePartitionModel, db) + + @pytest_asyncio.fixture(autouse=True) async def setup_test_tables(test_engine: Any) -> AsyncGenerator[None, None]: async with test_engine.begin() as conn: @@ -452,3 +490,149 @@ async def test_delete_accepts_model_instance_directly(db_session: Any) -> None: assert deleted is not None assert deleted.id == item.id assert await repo.get(id=item.id) is None + + +@pytest.mark.anyio +@pytest.mark.parametrize("order", ["desc", "asc"]) +async def test_repo_search_with_cursor_pagination_and_sort(db_session: Any, order: str) -> None: + repo = RepoTestRepository(db_session) + for i in range(6): + await repo.create( + RepoTestCreateSchema( + name=f"Alpha_{i:02d}", + description=f"Desc_{(5 - i):02d}", + ) + ) + + sort_field = "description" + search_req1 = SearchRequest( + filters=[], + sort=[{"field": sort_field, "order": order}], + ) + cursor_params1 = CursorParams(size=3) + + result_page1 = await repo.search(search_req1, pagination=cursor_params1) + assert len(result_page1.data) == 3 + assert result_page1.meta["has_more"] is True + next_cursor = result_page1.meta["next_cursor"] + assert next_cursor is not None + + search_req2 = SearchRequest( + filters=[], + sort=[{"field": sort_field, "order": order}], + ) + cursor_params2 = CursorParams(size=3, cursor=next_cursor) + + result_page2 = await repo.search(search_req2, pagination=cursor_params2) + assert len(result_page2.data) == 3 + assert result_page2.meta["has_more"] is False + + page1_ids = [item.id for item in result_page1.data] + page2_ids = [item.id for item in result_page2.data] + assert set(page1_ids).isdisjoint(set(page2_ids)) + + all_descriptions = [item.description for item in result_page1.data + result_page2.data] + expected_descriptions = sorted(all_descriptions, reverse=(order == "desc")) + assert all_descriptions == expected_descriptions + + +@pytest.mark.anyio +async def test_repo_cursor_pagination_with_date_column_coercion(db_session: Any) -> None: + repo = DateCursorRepository(db_session) + for i in range(5): + await repo.create(DateCursorCreateSchema(operation_date=date(2026, 10, 10 + i))) + + params1 = CursorParams(size=2, cursor_field="operation_date", order="desc") + res1 = await repo.get_list(pagination=params1) + assert len(res1.data) == 2 + assert res1.meta["has_more"] is True + next_cursor = res1.meta["next_cursor"] + assert next_cursor is not None + + params2 = CursorParams(size=2, cursor=next_cursor, cursor_field="operation_date", order="desc") + res2 = await repo.get_list(pagination=params2) + assert len(res2.data) == 2 + assert res2.meta["has_more"] is True + + first_page_dates = [item.operation_date for item in res1.data] + second_page_dates = [item.operation_date for item in res2.data] + assert all(isinstance(d, date) for d in first_page_dates + second_page_dates) + assert set(first_page_dates).isdisjoint(set(second_page_dates)) + assert first_page_dates[0] > second_page_dates[0] + + +@pytest.mark.anyio +async def test_repo_composite_guarded_properties(db_session: Any) -> None: + single_repo = RepoTestRepository(db_session) + assert single_repo.is_composite_pk is False + assert single_repo.pk is not None + assert single_repo.pk_name == "id" + + comp_repo = CompositeRepository(db_session) + assert comp_repo.is_composite_pk is True + assert comp_repo.pk_names == ("id", "operation_date") + + with pytest.raises(AttributeError) as exc_pk: + _ = comp_repo.pk + assert "composite primary key" in str(exc_pk.value) + + with pytest.raises(AttributeError) as exc_pk_name: + _ = comp_repo.pk_name + assert "composite primary key" in str(exc_pk_name.value) + + +@pytest.mark.anyio +async def test_repo_composite_primary_key_crud(db_session: Any) -> None: + repo = CompositeRepository(db_session) + assert repo.is_composite_pk is True + assert repo.pk_names == ("id", "operation_date") + + d1 = date(2026, 10, 18) + d2 = date(2026, 10, 19) + + item1 = await repo.create(CompositeCreateSchema(id=1, operation_date=d1, message="First")) + item2 = await repo.create(CompositeCreateSchema(id=2, operation_date=d2, message="Second")) + + assert await repo.exist(id=1, operation_date=d1) is True + assert await repo.exist(id=99, operation_date=d1) is False + + fetched_by_tuple = await repo.get(repo._build_pk_condition((1, d1))) + assert fetched_by_tuple is not None + assert fetched_by_tuple.message == "First" + + fetched_by_dict = await repo.get(repo._build_pk_condition({"id": 2, "operation_date": d2})) + assert fetched_by_dict is not None + assert fetched_by_dict.message == "Second" + + by_ids_tuples = await repo.get_by_ids([(1, d1), (2, d2)]) + assert len(by_ids_tuples) == 2 + + by_ids_dicts = await repo.get_by_ids( + [{"id": 1, "operation_date": d1}, {"id": 2, "operation_date": d2}] + ) + assert len(by_ids_dicts) == 2 + + by_ids_models = await repo.get_by_ids([item1, item2]) + assert len(by_ids_models) == 2 + + updated = await repo.update((1, d1), CompositeUpdateSchema(message="Updated First")) + assert updated is not None + assert updated.message == "Updated First" + + updated_items = await repo.update_multi( + { + (1, d1): CompositeUpdateSchema(message="Multi 1"), + (2, d2): CompositeUpdateSchema(message="Multi 2"), + } + ) + assert len(updated_items) == 2 + refreshed1 = await repo.get(repo._build_pk_condition((1, d1))) + assert refreshed1 is not None and refreshed1.message == "Multi 1" + + deleted = await repo.delete((1, d1)) + assert deleted is not None + assert await repo.get(repo._build_pk_condition((1, d1))) is None + + deleted_multi = await repo.delete_multi([(2, d2)]) + assert len(deleted_multi) == 1 + assert await repo.get(repo._build_pk_condition((2, d2))) is None diff --git a/tests/test_search.py b/tests/test_search.py index 5cec4fb..fbcbb5a 100644 --- a/tests/test_search.py +++ b/tests/test_search.py @@ -571,3 +571,28 @@ async def test_search_numeric_coercion_from_string(db_session: Any, seed_data: N req_invalid = SearchRequest(filters=[FilterItem(field="id", op="eq", value="not_a_number")]) engine.build_base_query(req_invalid) assert "failed to coerce" in str(exc_info.value).lower() + + +@pytest.mark.parametrize( + "page, cursor, should_raise", + [ + (1, None, False), + (1, "valid_cursor_token", False), + (2, None, False), + (2, "", False), + (2, " ", False), + (2, "valid_cursor_token", True), + (5, "valid_cursor_token", True), + ], +) +def test_search_request_pagination_mutual_exclusivity( + page: int, cursor: str | None, should_raise: bool +) -> None: + if should_raise: + with pytest.raises(Exception) as exc_info: + SearchRequest(page=page, cursor=cursor) + assert "Cannot combine 'cursor' with 'page' > 1" in str(exc_info.value) + else: + req = SearchRequest(page=page, cursor=cursor) + assert req.page == page + assert req.cursor == cursor diff --git a/tests/test_web.py b/tests/test_web.py index 4fd041b..2840799 100644 --- a/tests/test_web.py +++ b/tests/test_web.py @@ -15,7 +15,12 @@ from zcore.config import settings from zcore.context.context import ZContext, ctx -from zcore.db.pagination import PageNumberPagination, PageNumberParams, PaginatedResult +from zcore.db.pagination import ( + CursorPagination, + PageNumberPagination, + PageNumberParams, + PaginatedResult, +) from zcore.db.setup import Base, db_manager from zcore.exceptions.base import AppException, EntityNotFound from zcore.exceptions.handlers import app_exception_handler, register_exception_handlers @@ -85,6 +90,44 @@ class PrivateFieldModel(Zchema): internal_notes: str | None = None +class SearchCursorItem(Zchema): + __model__ = "cursor_dummy" + id: str + code: int + tag: str + + +class MockCursorSearchService: + def __init__(self) -> None: + self.items = [ + SearchCursorItem( + id=f"10000000-0000-0000-0000-00000000000{i}", code=i * 10, tag=f"tag_{i}" + ) + for i in range(5) + ] + + async def search( + self, + search_in: Any, + pagination: Any = None, + fields: Any = None, + options: Any = None, + ) -> Any: + order = search_in.sort[0].order if search_in.sort else "desc" + sorted_items = sorted(self.items, key=lambda x: x.code, reverse=(order == "desc")) + size = pagination.size if pagination else len(sorted_items) + data = sorted_items[:size] + has_more = len(sorted_items) > size + return PaginatedResult( + data=data, + meta={ + "next_cursor": "eyJ2YWx1ZSI6IDIwfQ" if has_more else None, + "has_more": has_more, + "size": size, + }, + ) + + class TargetService: def __init__(self, payload: dict[str, Any]) -> None: self.payload = payload @@ -247,7 +290,6 @@ class TargetRouter(BaseRouter[DummyCreate, DummyUpdate]): schema_out = DummyOut service = TargetService prefix = "/items" - expose_schemas = True def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: return [] @@ -264,7 +306,7 @@ def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: for header in expected_vary: assert header.lower() in vary_header.lower() if expected_vary: - schema_resp = await client.get("/items/?schema=true") + schema_resp = await client.get("/items/_schema?action=view") assert schema_resp.status_code == 200 schema_data = schema_resp.json() assert "password" not in schema_data["data"]["properties"] @@ -468,11 +510,12 @@ async def test_zchema_wildcard_pruning(monkeypatch: pytest.MonkeyPatch) -> None: @pytest.mark.anyio -async def test_method_schema_exposure(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_router_schema_introspection_endpoint(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( ZContext, "restricted_fields", property(lambda self: frozenset({"dummy.password"})) ) app = FastAPI() + register_exception_handlers(app) class ExposureRouter(BaseRouter[DummyCreate, DummyUpdate]): model = DummyModel @@ -481,7 +524,6 @@ class ExposureRouter(BaseRouter[DummyCreate, DummyUpdate]): schema_out = DummyOut service = MockFullService prefix = "/exposure" - expose_schemas = True def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: return [] @@ -490,12 +532,57 @@ def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: app.include_router(r.router) transport = ASGITransport(app=app, raise_app_exceptions=False) async with AsyncClient(transport=transport, base_url="http://test") as client: - res_post = await client.post("/exposure/?schema=true") - assert res_post.status_code == 200 - assert "name" in res_post.json()["data"]["properties"] - res_get = await client.get("/exposure/12345678-1234-5678-1234-567812345678?schema=true") - assert res_get.status_code == 200 - assert "password" not in res_get.json()["data"]["properties"] + # 1. Action-specific introspection: CREATE + res_create = await client.get("/exposure/_schema?action=create") + assert res_create.status_code == 200 + create_json = res_create.json() + assert create_json["success"] is True + assert "name" in create_json["data"]["properties"] + assert create_json["meta"]["action"] == "create" + + # 2. Action-specific introspection: VIEW (restricted password must be pruned) + res_view = await client.get("/exposure/_schema?action=view") + assert res_view.status_code == 200 + view_json = res_view.json() + assert view_json["success"] is True + assert "name" in view_json["data"]["properties"] + assert "password" not in view_json["data"]["properties"] + + # 3. Full resource introspection: No action query parameter + res_all = await client.get("/exposure/_schema") + assert res_all.status_code == 200 + all_json = res_all.json() + assert all_json["success"] is True + assert "create" in all_json["data"] + assert "view" in all_json["data"] + assert "update" in all_json["data"] + assert "password" not in all_json["data"]["view"]["properties"] + + # 4. Invalid action must yield 400 ValidationError + res_invalid = await client.get("/exposure/_schema?action=non_existing_action") + assert res_invalid.status_code == 400 + invalid_json = res_invalid.json() + assert invalid_json["success"] is False + assert "Invalid schema action" in invalid_json["message"] + + +@pytest.mark.anyio +async def test_router_schema_endpoint_exclusion() -> None: + class ExcludedSchemaRouter(BaseRouter[DummyCreate, DummyUpdate]): + model = DummyModel + create_schema = DummyCreate + update_schema = DummyUpdate + schema_out = DummyOut + service = MockFullService + prefix = "/excluded-schema" + exclude: ClassVar[set[RouteKey]] = {RouteKey.SCHEMA} + + def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: + return [] + + r = ExcludedSchemaRouter() + paths = [route.path for route in r.router.routes] + assert "/excluded-schema/_schema" not in paths @pytest.mark.anyio @@ -591,10 +678,13 @@ class SortRouter(BaseRouter[DummyCreate, DummyUpdate]): router_inst = SortRouter() paths = [r.path for r in router_inst.router.routes] assert "/test/search" in paths + assert "/test/_schema" in paths assert "/test/{id:uuid}" in paths idx_search = paths.index("/test/search") + idx_schema = paths.index("/test/_schema") idx_uuid = paths.index("/test/{id:uuid}") assert idx_search < idx_uuid + assert idx_schema < idx_uuid @pytest.mark.anyio @@ -702,13 +792,16 @@ class SpecificityRouter(BaseRouter[DummyCreate, DummyUpdate]): paths = [r.path for r in router_inst.router.routes] assert "/api/items/search" in paths assert "/api/items/lookup" in paths + assert "/api/items/_schema" in paths assert "/api/items/{id:uuid}" in paths idx_search = paths.index("/api/items/search") idx_lookup = paths.index("/api/items/lookup") + idx_schema = paths.index("/api/items/_schema") idx_dynamic = paths.index("/api/items/{id:uuid}") assert idx_search < idx_dynamic assert idx_lookup < idx_dynamic + assert idx_schema < idx_dynamic def test_zchema_timezone_serialization_recursive(monkeypatch: pytest.MonkeyPatch) -> None: @@ -921,3 +1014,177 @@ def test_zchema_private_fields_input_filtering_for_guest() -> None: assert "cost_price" not in filtered assert "internal_notes" not in filtered assert filtered["name"] == "ItemA" + + +@pytest.mark.anyio +async def test_router_search_with_cursor_pagination_and_sort( + monkeypatch: pytest.MonkeyPatch, +) -> None: + original_add_api_route = APIRouter.add_api_route + + def patched_add_api_route( + self: Any, path: Any, endpoint: Any, *args: Any, **kwargs: Any + ) -> Any: + return original_add_api_route( + self, path, clean_endpoint_signature(endpoint), *args, **kwargs + ) + + monkeypatch.setattr(APIRouter, "add_api_route", patched_add_api_route) + app = FastAPI() + + service_inst = MockCursorSearchService() + container.register_singleton(MockCursorSearchService, service_inst) + + class CursorSearchRouter(BaseRouter[DummyCreate, DummyUpdate]): + model = DummyModel + create_schema = DummyCreate + update_schema = DummyUpdate + schema_out = SearchCursorItem + service = MockCursorSearchService + prefix = "/cursor-search" + pagination_class = CursorPagination + + def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: + return [] + + r = CursorSearchRouter() + app.include_router(r.router) + + transport = ASGITransport(app=app, raise_app_exceptions=False) + async with AsyncClient(transport=transport, base_url="http://test") as client: + req_payload = { + "filters": [], + "sort": [{"field": "code", "order": "desc"}], + "size": 2, + } + res = await client.post("/cursor-search/search", json=req_payload) + assert res.status_code == 200 + body = res.json() + assert body["success"] is True + assert len(body["data"]) == 2 + assert body["meta"]["has_more"] is True + assert body["meta"]["next_cursor"] is not None + + next_cursor = body["meta"]["next_cursor"] + req_page2 = { + "filters": [], + "sort": [{"field": "code", "order": "desc"}], + "cursor": next_cursor, + "size": 2, + } + res_page2 = await client.post("/cursor-search/search", json=req_page2) + assert res_page2.status_code == 200 + body2 = res_page2.json() + assert body2["success"] is True + assert len(body2["data"]) == 2 + + +@pytest.mark.anyio +async def test_subclassed_router_custom_route_in_init_avoids_shadowing() -> None: + class DummyService: + pass + + class SubclassedInitRouter(BaseRouter[DummyCreate, DummyUpdate]): + model = DummyModel + create_schema = DummyCreate + update_schema = DummyUpdate + schema_out = DummyOut + service = DummyService + prefix = "/items-init" + + def __init__(self) -> None: + super().__init__() + self.router.add_api_route("/summary", self.summary_endpoint, methods=["GET"]) + + async def summary_endpoint(self) -> dict[str, bool]: + return {"ok": True} + + def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: + return [] + + r = SubclassedInitRouter() + paths = [route.path for route in r.router.routes] + idx_summary = paths.index("/items-init/summary") + idx_uuid = paths.index("/items-init/{id:uuid}") + assert idx_summary < idx_uuid + + app = FastAPI() + app.include_router(r.router) + + transport = ASGITransport(app=app, raise_app_exceptions=False) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.get("/items-init/summary") + assert res.status_code == 200 + assert res.json() == {"ok": True} + + +@pytest.mark.anyio +async def test_subclassed_router_hook_register_custom_routes() -> None: + class DummyService: + pass + + class SubclassedHookRouter(BaseRouter[DummyCreate, DummyUpdate]): + model = DummyModel + create_schema = DummyCreate + update_schema = DummyUpdate + schema_out = DummyOut + service = DummyService + prefix = "/items-hook" + + def register_custom_routes(self) -> None: + self.router.add_api_route("/reports", self.reports_endpoint, methods=["GET"]) + + async def reports_endpoint(self) -> dict[str, str]: + return {"report": "generated"} + + def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: + return [] + + r = SubclassedHookRouter() + paths = [route.path for route in r.router.routes] + idx_reports = paths.index("/items-hook/reports") + idx_uuid = paths.index("/items-hook/{id:uuid}") + assert idx_reports < idx_uuid + + app = FastAPI() + app.include_router(r.router) + + transport = ASGITransport(app=app, raise_app_exceptions=False) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.get("/items-hook/reports") + assert res.status_code == 200 + assert res.json() == {"report": "generated"} + + +@pytest.mark.anyio +async def test_search_endpoint_rejects_conflicting_pagination_params() -> None: + app = FastAPI() + register_exception_handlers(app) + + service_inst = MockFullService() + container.register_singleton(MockFullService, service_inst) + + class ConflictRouter(BaseRouter[DummyCreate, DummyUpdate]): + model = DummyModel + create_schema = DummyCreate + update_schema = DummyUpdate + schema_out = DummyOut + service = MockFullService + prefix = "/conflict-test" + + def get_route_dependencies(self, route_key: RouteKey, action: str) -> list[Any]: + return [] + + r = ConflictRouter() + app.include_router(r.router) + + transport = ASGITransport(app=app, raise_app_exceptions=False) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.post( + "/conflict-test/search", + json={"page": 3, "cursor": "eyJ2YWx1ZSI6IC...", "size": 10}, + ) + assert res.status_code == 422 + body = res.json() + assert body["success"] is False + assert "Cannot combine 'cursor' with 'page' > 1" in body["message"]