Metadata-Version: 2.4
Name: oplnk-python-utils
Version: 1.6.0
Summary: Reusable utilities for caching, data manipulation, parsing, and MongoDB query building
Author-email: OpenLink SpA <contacto@openlink.cl>
License-Expression: MIT
Project-URL: Homepage, https://github.com/openlinkspa/oplnk-python-utils
Project-URL: Repository, https://github.com/openlinkspa/oplnk-python-utils
Project-URL: Changelog, https://github.com/openlinkspa/oplnk-python-utils/blob/main/CHANGELOG.md
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: fastapi>=0.104
Requires-Dist: pymongo>=4.6
Requires-Dist: redis>=5.0
Requires-Dist: starlette>=0.27
Provides-Extra: sql
Requires-Dist: sqlalchemy[asyncio]>=2.0; extra == "sql"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-cov>=5.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.24; extra == "dev"
Requires-Dist: fakeredis[lua]>=2.20; extra == "dev"
Requires-Dist: testcontainers[mongodb]>=4.15; extra == "dev"
Requires-Dist: httpx>=0.24; extra == "dev"
Requires-Dist: black>=21.0; extra == "dev"
Requires-Dist: flake8>=3.8; extra == "dev"
Requires-Dist: libcst>=1.5; extra == "dev"
Requires-Dist: sqlalchemy[asyncio]>=2.0; extra == "dev"
Requires-Dist: aiosqlite>=0.20; extra == "dev"
Dynamic: license-file

# OpenLink Python Utils

A comprehensive Python package providing reusable utilities for caching, data manipulation, parsing, and MongoDB query building. Developed by OpenLink SpA for use across multiple projects.

## Overview

This package abstracts common functionality for:

-  **Caching**: Redis-based caching utilities with serialization support
-  **Query Building**: MongoDB aggregation pipeline builders with advanced filtering, sorting, pagination, and population features

## Requirements

-  Python 3.10 or later (CI tests 3.10 to 3.13).
-  fastapi 0.104 or later, pymongo 4.6 or later and redis-py 5.0 or later; pip installs them with the package.
-  For the cache, a Redis server 4.0 or later that allows `EVAL`. For text filters on numeric fields in `QueryBuilder`, MongoDB 4.2 or later.

## Installation

Install from PyPI:

```bash
pip install oplnk-python-utils
```

Pin the exact version in `requirements.txt`:

```
oplnk-python-utils==1.6.0
```

For the optional SQLAlchemy compiler and adapter (Q2; SQL Server first):

```bash
pip install "oplnk-python-utils[sql]"
```

For development, from a clone (pip 21.3 or later):

```bash
pip install -e ".[dev]"
```

The package ships type hints (`py.typed`), so mypy and pyright check calls into it.

## Core Module

The core module contains general-purpose utilities organized by functionality. These are the foundational utilities used by other modules and can be used independently.

### Data Utilities (`core.data`)

Functions for data structure manipulation and normalization.

```python
from oplnk_python_utils.core.data import normalize, merge_dicts

# Normalize data for consistent processing (useful for caching)
data = {"z": 1, "a": {"c": 3, "b": 2}}
normalized = normalize(data)
# Result: {"a": {"b": 2, "c": 3}, "z": 1} - keys sorted recursively

# Merge multiple dictionaries with nested structure support
result = merge_dicts(
    {"filters": {"name": "John"}},
    {"filters": {"age": 25}},
    {"sort": ["name:asc"]}
)
# Result: {"filters": {"name": "John", "age": 25}, "sort": ["name:asc"]}
# The input dictionaries are not modified.

# Advanced merging with arrays and conflicts
dict1 = {"tags": ["python"], "meta": {"version": 1}}
dict2 = {"tags": ["fastapi"], "meta": {"author": "team"}}
merged = merge_dicts(dict1, dict2)
# Result: {"tags": ["python", "fastapi"], "meta": {"version": 1, "author": "team"}}
```

#### Data Functions

-  **`normalize(value)`**: Recursively normalize data structures by sorting dictionary keys (mixed key types allowed); sets become sorted lists
-  **`merge_dicts(*dicts)`**: Intelligently merge multiple dictionaries with nested structure support, without modifying the inputs

### Parsing Utilities (`core.parsing`)

Functions for parsing query parameters, type conversion, and safe field-path handling for MongoDB aggregation.

```python
from oplnk_python_utils.core.parsing import (
    parse_params_to_dict,
    infer_type,
    parse_key_string_to_list,
    parse_list_to_dict
)
from starlette.datastructures import QueryParams

# Parse nested query parameters (main function)
params = {
    "filters[name][$containsi]": "john",
    "filters[age][$gte]": "18",
    "pagination[page]": "1",
    "sort[0]": "name:asc",
    "populate[0]": "user"
}
parsed = parse_params_to_dict(params)
# Result: {
#   "filters": {"name": {"$containsi": "john"}, "age": {"$gte": 18}},
#   "pagination": {"page": 1},
#   "sort": ["name:asc"],
#   "populate": ["user"]
# }

# Smart type inference with various formats
value = infer_type("123")           # Returns: 123 (int)
value = infer_type("12.5")          # Returns: 12.5 (float)
value = infer_type("true")          # Returns: True (bool)
value = infer_type("false")         # Returns: False (bool)
value = infer_type("hello")         # Returns: "hello" (str)
value = infer_type("007")           # Returns: "007" (str, not a canonical number)
value = infer_type("2024-01-01T10:00:00.000Z")  # Returns: aware datetime in UTC
value = infer_type("2024-01-01T10:00:00")       # Returns: "2024-01-01T10:00:00" (str, no offset)
value = infer_type(["1", "2"])      # Returns: [1, 2] (list with converted types)

# Lists are built by index, so keys that share an index form one item
parse_params_to_dict(QueryParams("filters[$or][0][a]=x&filters[$or][0][b]=y&filters[$or][1][c]=z"))
# Result: {"filters": {"$or": [{"a": "x", "b": "y"}, {"c": "z"}]}}

# `key[]` and repeated keys keep every value (pass QueryParams, not a dict)
parse_params_to_dict(QueryParams("filters[status][$in][]=a&filters[status][$in][]=b"))
# Result: {"filters": {"status": {"$in": ["a", "b"]}}}

# Lower-level parsing functions
keys = parse_key_string_to_list("filters[name][$eq]")
# Result: ["filters", "name", "$eq"]

nested_dict = parse_list_to_dict(["filters", "name", "$eq"], "john")
# Result: {"filters": {"name": {"$eq": "john"}}}
```

#### Parsing Functions

-  **`parse_params_to_dict(params)`**: Parse FastAPI QueryParams or dict into structured nested dictionary. Lists are grouped by index, and `key[]` and repeated keys give lists. Raises `ValueError` for non-mapping input, a top-level index key (`?0=x`), a key nested more than `MAX_NESTING_DEPTH` (32) levels of brackets, or one key used with two shapes
-  **`infer_type(value)`**: Smart type inference from string values: canonical numbers, bool, ISO-8601 datetimes with an offset (`Z` or `±HH:MM`), and lists. Anything else stays a string
-  **`parse_key_string_to_list(s)`**: Parse bracketed notation strings like `"filters[name][$eq]"`
-  **`parse_list_to_dict(lst, value)`**: Convert key list and value into nested dictionary structure

#### Field Path Utilities for Aggregation (NEW)

-  **`normalize_field_path(field)`**: Normalize paths with array notation for MongoDB aggregation.
   -  Example: `"user.addresses[1].street" -> "user.addresses.1.street"`
-  **`validate_field_path(field)`**: Validate field paths before using them in pipelines: whitespace or `$` anywhere, including at the start, makes a path invalid.
-  **`build_aggregation_field_expression(field)`**: Produce a proper aggregation expression (`"items[0].price"` → `"$items.0.price"`); a path that starts with `$` raises `ValueError`.
   -  Example: `"user.addresses[1].street" -> "$user.addresses.1.street"`

### Serialization Utilities (`core.serialization`)

Functions for converting data types to serializable formats.

```python
from oplnk_python_utils.core.serialization import datetime_serializer
from datetime import datetime

# Serialize datetime objects for JSON/cache storage
data = {
    "created_at": datetime.now(),
    "users": [
        {"last_login": datetime.now(), "name": "John"},
        {"last_login": datetime.now(), "name": "Jane"}
    ],
    "metadata": {
        "processed_at": datetime.now()
    }
}

serialized = datetime_serializer(data)
# Returns a new structure; `data` is not modified
# datetime, date and time become ISO strings, ObjectId becomes str,
# Decimal becomes int or float, tuples become lists, NaN/infinity become None
# Other values remain unchanged; cyclic data raises ValueError
```

#### Serialization Functions

-  **`datetime_serializer(data)`**: Recursively convert a structure into JSON-safe values (datetime, date, time, ObjectId, Decimal, tuples, NaN/infinity) without modifying the input

## Cache Module

Advanced caching utilities with Redis integration and intelligent cache key management. The module provides both low-level caching utilities and high-level endpoint decorators for comprehensive caching strategies.

### Low-Level Cache Utilities

```python
from oplnk_python_utils.cache import generate_cache_id, get_or_set_cache, invalidate_cache_keys

# Generate consistent cache keys from complex data
cache_key = generate_cache_id({
    "filters": {"name": "John", "status": "active"},
    "sort": ["created_at:desc"]
})
# Automatically normalizes data internally and creates SHA256 hash

# Get from cache or execute a function and cache result (sync fetch_func)
def fetch_users():
    # Some expensive database operation (sync example)
    return list(db.collection.find({"status": "active"}))

result = get_or_set_cache(
    redis_client=redis,
    cache_key="user_query_abc123",
    fetch_func=fetch_users,  # zero-arg callable
    ttl=3600,  # 1 hour
    serializer=datetime_serializer
)

# Invalidate multiple cache keys by pattern (returns -1 if a Redis error stopped it)
invalidated_count = invalidate_cache_keys(redis, "user_*")
print(f"Invalidated {invalidated_count} cache entries")
```

### Endpoint-Level Caching (NEW)

High-level decorators for caching entire endpoint responses with automatic cache key generation and tag-based invalidation.

```python
from oplnk_python_utils.cache import cache_response, cache_factory, build_cache_key, invalidate_tag, ainvalidate_tag
from fastapi import FastAPI, Request
import redis

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0)

# Method 1: Direct decorator usage
@app.get("/users")
@cache_response(
    redis_client=redis_client,
    api_name="myapi",
    collection="users",
    endpoint_type="read",
    tags=["users", "all_data"],
    ttl=3600,  # 1 hour
    use_cache=True
)
async def get_users(request: Request):
    # Your endpoint logic here
    return {"users": [...]}

# Method 2: Factory pattern for reusable decorators
users_cache = cache_factory(redis_client, "myapi", "users")

@app.get("/users")
@users_cache(
    endpoint_type="read",
    tags=["users", "public"],
    ttl=1800  # 30 minutes
)
async def get_users(request: Request):
    return {"users": [...]}

@app.post("/users")
async def create_user(request: Request):
    # ... write the user ...
    # Writes are not cached (only GET and HEAD are, by default). Drop the cached
    # reads of the collection, so the next GET /users runs again.
    await ainvalidate_tag(redis_client, "users")
    return {"created": True}

# Manual cache key building (build_cache_key is async)
@app.get("/custom")
async def custom_endpoint(request: Request):
    cache_key = await build_cache_key(request, "myapi", "custom", "read")
    # Use cache_key for custom caching logic
    return {"data": "custom"}

# Tag-based cache invalidation
@app.delete("/users/{user_id}")
async def delete_user(user_id: str):
    # Delete user logic here
    # ...

    # Invalidate all cached responses tagged with 'users'
    await ainvalidate_tag(redis_client, "users")
    return {"deleted": True}
```

Only GET and HEAD requests are cached. Any other method runs the endpoint every time and never contacts Redis, `?clear_cache` included. A read-only POST (a summary that takes its filters in the body) opts in with `methods=["POST"]`, and is keyed by its JSON body. Never opt a write in: call `ainvalidate_tag` (or `invalidate_tag` in a sync handler) in the write handler instead.

### Cache Key Structure

The caching system generates structured cache keys based on:

-  **API Name**: Namespace for your application
-  **Collection**: The data collection being cached
-  **Endpoint Type**: Type of operation (read, create, update, delete)
-  **Query Hash**: SHA256 of the normalized method, path, path parameters, query parameters (every value of a repeated one), the body for methods other than GET and HEAD, and any `vary_on` headers

Example cache key: `myapi:users:read:a1b2c3d4e5f6...`

**Cached responses are shared by every caller.** The key does not include who is calling. On a hit, FastAPI still runs the endpoint's dependencies (`Depends`, `Security`) but not the endpoint body, so an endpoint whose output depends on the caller (rows filtered by the current user, a tenant header) would serve one user's response to another. Such an endpoint must opt in to a key per caller:

```python
@users_cache(endpoint_type="mine", tags=["users"], ttl=300, vary_on=["Authorization"])
async def my_items(request: Request, user=Depends(get_current_user)): ...

@reports_cache(endpoint_type="summary", tags=["reports"], ttl=300,
               key_builder=lambda request: request.headers.get("x-tenant", ""))
async def report(request: Request): ...
```

`vary_on` adds the named headers' values to the key. `key_builder(request) -> str` (a sync function) replaces the key material: requests that return equal strings share an entry. Both are hashed, so header values never appear in Redis keys. Pass one or the other, not both. If you cannot say what makes a response differ, do not cache the endpoint.

### Tag-Based Cache Management

```python
# Cache responses with multiple tags
@cache_response(
    redis_client=redis_client,
    api_name="ecommerce",
    collection="products",
    endpoint_type="read",
    tags=["products", "catalog", "public_data"],
    ttl=7200  # 2 hours
)
async def get_products(request: Request):
    return {"products": [...]}

# Invalidate specific tag groups
invalidate_tag(redis_client, "products")      # Invalidates all product-related cache
invalidate_tag(redis_client, "catalog")       # Invalidates all catalog cache
invalidate_tag(redis_client, "public_data")   # Invalidates all public data cache
```

Tags are static: an entry is invalidated only through the tags it was stored with. A response that populates a related collection (for example fuel supplies with their `Establishment`) is not invalidated when that collection changes, and keeps the old related data until its TTL. List the related collection among the endpoint's tags, or invalidate the endpoint's tag in the related collection's write handler:

```python
# 1. Tag the endpoint with the related collection too
@route_cache(endpoint_type="all", tags=["fuel_supply", "establishment"], ttl=300)
async def list_fuel_supply(request: Request): ...

# 2. Or invalidate both tags in the related collection's write handler
@router.put("/establishment/{id}")
async def update_establishment(id: str, data: EstablishmentIn):
    ...  # write
    await ainvalidate_tag(redis_client, "establishment")
    await ainvalidate_tag(redis_client, "fuel_supply")
```

### Cache Functions Reference

#### Low-Level Functions

-  **`generate_cache_id(params)`**: Generate SHA256 hash from normalized data for consistent cache keys
-  **`get_or_set_cache(redis_client, cache_key, fetch_func, clear_cache=False, ttl=None, serializer=None, max_retries=3, retry_delay=0.1, tags=None)`**: Get cached result or execute function and cache with automatic serialization; `tags` registers a newly stored key under `tag:<tag>` so that `invalidate_tag` deletes it
-  **`invalidate_cache_keys(redis_client, pattern)`**: Delete cache keys matching a SCAN glob pattern (escape `*`, `?`, `[`, `]` and `\` with a backslash to match them literally; a pattern without wildcards deletes that one key without a SCAN); returns the number of deleted keys, or -1 when a Redis error stopped it; raises `TypeError` for an async client (use `ainvalidate_cache_keys`)
-  **`ainvalidate_cache_keys(redis_client, pattern)`**: Async `invalidate_cache_keys`, with the same pattern rules and return value; takes an async or a sync client

#### Endpoint Caching Functions

-  **`build_cache_key(request, api_name, collection, endpoint_type, vary_on=None)`**: Build structured cache key from request parameters
-  **`cache_response(redis_client, api_name, collection, endpoint_type, tags, ttl=None, use_cache=True, max_retries=3, retry_delay=0.1, *, methods=None, vary_on=None, key_builder=None, can_clear=None, clear_cache_interval=0, single_flight=True)`**: Decorator for caching endpoint responses; an async `key_builder` raises `TypeError` when decorating; `ttl` is a positive int (seconds), a positive `datetime.timedelta` or `None` (no expiry); any other value raises `ValueError` when decorating
-  **`cache_factory(redis_client, api_name, collection)`**: Factory function to create reusable cache decorators; the returned decorator takes `endpoint_type, tags, ttl=None, use_cache=True, max_retries=3, retry_delay=0.1` and keyword-only `methods`, `vary_on`, `key_builder`, `can_clear`, `clear_cache_interval` and `single_flight`
-  **`invalidate_tag(redis_client, tag)`**: Invalidate all cache keys associated with a specific tag; returns the number of tagged keys deleted, or -1 when a Redis error stopped it; raises `TypeError` for an async client (use `ainvalidate_tag`)
-  **`aget_or_set_cache(redis_client, cache_key, fetch_func, clear_cache=False, ttl=None, serializer=None, max_retries=3, retry_delay=0.1, tags=None)`**: Async `get_or_set_cache`; takes an async or a sync client, and a `fetch_func` that may return an awaitable
-  **`ainvalidate_tag(redis_client, tag)`**: Async `invalidate_tag`; takes an async or a sync client

#### Redis errors, retries and logs

A Redis error never fails a request or a query: the cache logs one WARNING for the call and returns the endpoint's, the fetch function's or the query's result. After that first failure, no cache call of the same request contacts Redis again: the decorator, `get_or_set_cache`, `aget_or_set_cache` and the `QueryBuilder` count. Outside a decorated request, each call stands alone. A failed `invalidate_tag` or `invalidate_cache_keys` also marks the request, but they still run when called after a failure.

Connection and timeout errors are retried only when the Redis client has no retry policy of its own (`retry=Retry(...)` with retries, or `retry_on_error`). redis-py 8's `redis.Redis()` retries by default; redis-py 5's does not. The library then makes `max_retries` attempts per operation (default 3), waiting `retry_delay` seconds (default 0.1), then twice as long each time, plus up to 50% jitter, at most 2 seconds per wait and 5 seconds in total per operation.

With a sync client, the decorator runs every Redis command in the cache's own thread pool, so Redis never blocks the event loop. The library sets no timeout, so give your client socket timeouts; without them a Redis that stops answering holds each cached request, and decorating an endpoint logs a WARNING once per client. For example: `redis.Redis(host=..., port=..., socket_timeout=1, socket_connect_timeout=1)`.

Async clients (`redis.asyncio.Redis`) work with `cache_response` and `cache_factory`; use `aget_or_set_cache`, `ainvalidate_tag` and `ainvalidate_cache_keys` with them, since the sync helpers raise `TypeError`. The `QueryBuilder` count cache needs a sync client: with an async one it counts without the cache and logs a WARNING once per client.

Tagged entries are stored and invalidated by Lua scripts, so the Redis server must be 4.0 or later and allow `EVAL` and `EVALSHA` (ACL category `@scripting`). An entry and its tags are stored together or not at all. Each tag has a generation counter (`tag-gen:<tag>`, never expires): `invalidate_tag` deletes the tagged entries in batches, deletes `tag:<tag>` and bumps the counter, and a cached call whose read came before an invalidation does not store what it fetched. A tag set lives as long as its longest-lived entry. If the server rejects scripts, the cache logs one WARNING per client and uses separate commands, which can leave a stale or partly tagged entry when an invalidation or a failure falls between them. Redis Cluster clients are not supported: their errors are logged and the calls run uncached. Under an `allkeys-*` eviction policy Redis can evict a `tag-gen:<tag>` counter, which weakens the fence for that tag; `volatile-*` policies never evict it. An endpoint that invalidates one of its own tags in its body never caches, since its store always comes after an invalidation.

These log texts from the `oplnk_python_utils.cache.cache_utils` logger are stable for alerting:

-  WARNING `Redis error during <operation>: <error>`, where `<operation>` is `cache read`, `cache store`, `cache invalidation`, `tag invalidation` or `purge rate limit`. `extra`: `cache_operation`, `cache_attempts`, `cache_error_type`, and `cache_key`, `cache_pattern` or `cache_tag`.
-  WARNING `Failed to decode cached data for key <key>: <error>` (text unchanged).
-  WARNING `Redis rejected the cache's scripts (<error>); this client now stores and invalidates tagged entries with separate commands, which are not atomic`, once per client (`extra`: `cache_operation="configuration"`, `cache_error_type`).
-  DEBUG `Cache hit for key <key>` and `Cache miss for key <key>`. `extra`: `cache_key`, `cache_hit`.
-  WARNING `Redis client for <host>:<port> has no socket timeout; …` (`extra`: `cache_operation="configuration"`).
-  From `oplnk_python_utils.query.query_builder`: WARNING `Query count cache skipped: <client class> is an async Redis client and the count cache needs a sync one`, once per client.

### Automatic Cache Invalidation

A cached GET or HEAD request with `?clear_cache=true` purges its collection's cache before the read: it invalidates the tag named after `collection` (the entries tagged with it and that collection's cached query counts), then runs the endpoint and caches the fresh response. `true`, `1`, `yes` and `on` turn it on, in any case; `false`, `0`, `no`, `off`, anything else or no value leave it off, and a repeated parameter uses its last value. `QueryBuilder` reads its own `clear_cache` parameter the same way.

Any caller may purge by default. `can_clear(request)`, sync or async, restricts it; a refused request is served as if it had no `clear_cache`, and the `QueryBuilder` count it runs is a normal cached read too. A sync `can_clear` runs on the event loop, so it must not block; if it raises, the request fails (a 500), as with a failing dependency:

```python
def is_admin(request: Request) -> bool:
    return "admin" in getattr(request.state, "roles", ())

@users_cache(endpoint_type="all", tags=["users"], ttl=300, can_clear=is_admin)
async def list_users(request: Request): ...
```

By default every `?clear_cache=true` purges, as in 1.3.0, so a "Refresh" button always shows fresh data. `clear_cache_interval` opts in to a limit: a collection is then purged at most once per that many seconds, across every process that shares the Redis (the first purge sets the key `tag-purge:<collection>`, holding a random value of its own, with that TTL). The interval is at most 9223090561878065 seconds, the largest TTL Redis accepts; a larger value raises `ValueError` when decorating. A `?clear_cache=true` that arrives within the interval purges nothing, but still skips its own cached entry, runs the endpoint and stores its response, and the `QueryBuilder` count it runs is computed fresh and replaces only its own cached count. Other entries of the collection, such as the grid a "Refresh" button refetches without `clear_cache`, may then be up to the interval old; set an interval only where that trade-off is acceptable. If Redis fails while claiming the slot, the request runs uncached, as after any Redis error. If Redis answers the purge's invalidation with an error (such as WRONGTYPE, OOM or READONLY), the request runs uncached and deletes the key it set, with a script that deletes it only while it still holds that request's value, so the next `?clear_cache=true` purges; the invalidation may have run in part, and purging again is harmless. After a connection or timeout error, or on a server that refuses the cache's scripts, the key is left to expire with the interval. `read_query` called outside a decorated request purges its prefix's counts on `clear_cache` as before.

Concurrent misses on the same key in one process run the endpoint once: the first request runs it and stores the response, and the requests that arrive meanwhile wait and get what it stored. If it stores nothing (the endpoint raised or was cancelled, it returned a `Response`, the `response_model` rejected the value, or a Redis error or an invalidation stopped the store), each waiting request runs the endpoint itself. A refresh refused by the purge interval always runs its own. Separate processes, workers or pods still run the endpoint once each. Pass `single_flight=False` to turn it off.

## Query Module

The query module provides a MongoDB aggregation pipeline builder with advanced features.

### Usage

```python
from oplnk_python_utils.query import QueryBuilder
from pymongo import MongoClient

# Initialize MongoDB
db = MongoClient()["mydb"]
collection = db["users"]

# Create query builder
query_builder = QueryBuilder(
    collection_prefix="myapp",
    relations=[
        ["User", "user_id", "myapp", "users"],
        ["Role", "role_ids", "myapp", "roles"]
    ]
)

# Build query
query = {
    "filters": {
        "name": {"$containsi": "john"},
        "age": {"$gte": 18},
        "$or": [
            {"status": {"$eq": "active"}},
            {"last_login": {"$gte": "2023-01-01"}}
        ]
    },
    "sort": ["name:asc", "age:desc"],
    "pagination": {"page": 1, "pageSize": 10},
    "populate": ["role"],
    "fields": ["name", "email", "age"]
}

# Method 1: Using parsed query dictionary
results = query_builder.get_query_response(
    db_collection=collection,
    query=query,
    cache_client=redis_client,
    cache_key_prefix="users",
    use_cache=True
)

# Method 2: Using raw query parameters (like from FastAPI request)
raw_params = {
    "filters[name][$containsi]": "john",
    "filters[age][$gte]": "18",
    "sort[0]": "name:asc",
    "pagination[page]": "1",
    "pagination[pageSize]": "10"
}

# Option A: Parse manually then use get_query_response
from oplnk_python_utils.core.parsing import parse_params_to_dict
parsed_query = parse_params_to_dict(raw_params)
results = query_builder.get_query_response(
    db_collection=collection,
    query=parsed_query,
    cache_client=redis_client,
    cache_key_prefix="users",
    use_cache=True
)

# Option B: Use read_query with automatic parsing (recommended)
results = query_builder.read_query(
    db_collection=collection,
    params=raw_params,  # Will be parsed automatically using core.parsing
    cache_client=redis_client,
    cache_key_prefix="users",
    use_cache=False  # Note: Endpoint-level caching is now preferred over query-level caching
)

print(results)
# {
#     "data": [...],
#     "meta": {"totalCount": 150}
# }
```

### SQL: the same contract over SQLAlchemy models

`SQLQueryBuilder` (`oplnk_python_utils.query.sql`, optional `sql` extra) derives
the query surface from the model and answers with the same `{data, meta}` shape,
so an endpoint is a subclass and one call:

```python
from oplnk_python_utils.query.sql import SQLQueryBuilder

from .models import Category, Item


class ItemQuery(SQLQueryBuilder):
    def __init__(self):
        super().__init__(
            Item,
            relations={"category": Category},
            protected_user_fields=("hashed_password",),
            max_page_size=100,
        )


result = await ItemQuery().read_query(
    session,                        # the caller's AsyncSession
    params=request.query_params,    # parsed like the Mongo path
    use_cache=True,
    cache_client=redis_client,
    cache_key_prefix="items",
)
# {
#     "data": [{"id": 1, "name": "...", "category": {"id": 2, "name": "..."}}],
#     "meta": {"totalCount": 150}
# }
```

-  **Inference**: every mapped column is queryable with the operator family of its
  type; a boolean `disable` column is the soft delete, the primary key is the
  default order, and `mapped_column(info={"query": {"exclude": True}})` hides a
  column from filters, sort, `fields` and the output.
-  **Relations**: a foreign key gives a many-to-one relation named after the
  column without `_id` (`category_id` → `category`); `relationship()` gives
  one-to-many and many-to-many; a `uselist=False` relationship is one object. A
  master with no declared relationship resolves its child by model name
  (case-insensitively) and rejects an ambiguous one naming the candidates, which
  `relations={"entries": LedgerEntry}` resolves.
-  **Populate**: many-to-one/one-to-one nest an object or `null`; one-to-many and
  many-to-many nest an array (possibly `[]`) read with one extra query on the same
  session, ordered by the child's default order, honoring its soft delete and
  never touching the parent's pagination or `totalCount`. Populate null-fills the
  parent row; a filter over the same relation excludes it.
-  **Count cache**: with `use_cache=True` and a `cache_client`, `totalCount` is
  cached under `<cache_key_prefix>:query:count:<hash>` (source + query without
  pagination) for `count_cache_ttl` seconds; `clear_cache=True` invalidates the
  prefix before counting, and a Redis error degrades to a fresh count.
-  **Errors**: only `QueryValidationError` (the 400 contract); the builder never
  raises web exceptions and never commits or closes the session. Multi-level
  populate, JSON-path filtering over payload columns and `distinct`/full-text
  search are rejected with a message pointing at the low-level
  `SQLQueryAdapter`. See `docs/query-sql-adapter.md`.

### QueryBuilder Methods

#### Pipeline Builders

-  **`build_filter_pipeline(filters)`**: Build MongoDB filter pipeline
-  **`build_sort_pipeline(sort_params)`**: Build sort pipeline
-  **`build_fields_pipeline(fields_params)`**: Build field projection pipeline
-  **`build_distinct_pipeline(distinct_params)`**: Build distinct pipeline
-  **`build_pagination_pipeline(pagination_params)`**: Build pagination pipeline
-  **`build_populate_pipeline(populate_params)`**: Build population pipeline

#### Aggregation Builders

-  **`make_aggregation_query(query, base_pipeline=None)`**: Build complete aggregation pipeline
-  **`make_count_aggregation_query(query, base_pipeline=None)`**: Build count aggregation pipeline

#### Query Execution

-  **`get_query_response(db_collection, query, ...)`**: Execute query with optional caching
-  **`read_query(db_collection, params=None, query=None, ..., count_cache_ttl=1800)`**: Complete query method with parameter parsing

With `use_cache=True` and a `cache_client`, `totalCount` is cached under `<cache_key_prefix>:query:count:<hash>`, where the hash covers the collection, the query without pagination, `pipeline_count`, and whether a `pipeline` comes without `pipeline_count`; the count never runs the data `pipeline`, so its content is not in the key, and a compiled regex counts with its flags. The entry lives `count_cache_ttl` seconds (30 minutes by default; `None` keeps it until it is invalidated; any other value than a positive int raises `ValueError`). It is registered under the tag made of the collection name with `<collection_prefix>_` removed, so `invalidate_tag(redis, "fuel_supply")` after a write to `est_fuel_supply` clears it. `clear_cache=true` deletes every count under the prefix.

#### Core Module Integration

The QueryBuilder class leverages utilities from the core module for enhanced functionality:

-  **Parameter parsing**: Uses `core.parsing.parse_params_to_dict()` for automatic query parameter parsing
-  **Data normalization**: Uses `core.data.normalize()` for consistent cache key generation
-  **Datetime serialization**: Uses `core.serialization.datetime_serializer()` for cache-friendly data

See the [Core Module](#core-module) section for detailed documentation of these utilities.

### Supported Filter Operators

Filters accept only the operators below, plus `$nin` as an alias of `$notIn`. Any other operator, such as `$regex`, `$not`, `$expr` or `$where`, anywhere in `filters` (inside `$or`/`$and` or inside another operator's value too), answers HTTP 400 `Operator '$regex' is not allowed in filters`, whether or not protected fields are configured. A field path segment cannot start with `$`. An operator's value is a value: an operator inside it, at any depth (`filters[stock][$eq][$gt]=3`), answers HTTP 400. A query parameter may nest at most 32 levels of brackets, and `filters` at most 32 levels (`MAX_NESTING_DEPTH`); the deepest realistic query, `filters[$or][0][$and][0][spec][size][$eq]`, nests 7.

Several operators on one field all apply: `filters[stock][$gte]=3&filters[stock][$lte]=8` keeps both bounds.

A repeated key or `key[]` is an array, like indexed keys: `filters[status][$in]=a&filters[status][$in]=b` is `$in: ["a", "b"]`, and a repeated `$eq` or `$ne` compares with the array. `$lt`, `$lte`, `$gt`, `$gte`, the text operators and pagination take one value and use the last one.

#### Text Search

-  `$contains`, `$containsi` (case-insensitive)
-  `$notContains`, `$notContainsi`
-  `$startsWith`, `$startsWithi`
-  `$endsWith`, `$endsWithi`
-  `$eqi` (case-insensitive equality), `$nei`

#### Comparison

-  `$eq`, `$ne`, `$lt`, `$lte`, `$gt`, `$gte`
-  `$in`, `$notIn` (`$nin` is an alias of `$notIn`)
-  `$between` (requires array with 2 elements)

#### Logical

-  `$and`, `$or`

#### Existence

-  `$exists`, `$notExists`
-  `$null`, `$notNull`

#### Arrays

-  `$listContains`, `$listNotContains`

### Query Structure

```python
query = {
    "filters": {
        # Filter conditions
        "field": {"$operator": "value"},
        "$or": [
            {"field1": {"$eq": "value1"}},
            {"field2": {"$gt": 10}}
        ]
    },
    "sort": ["field1:asc", "field2:desc"],  # asc or desc, in any case
    "pagination": {
        "page": 1,
        "pageSize": 20
    },
    "populate": ["relation1", "relation2"],  # or {"relation1": True}
    "fields": ["field1", "field2", "field3"],
    "distinct": ["field1", "field2"]
}
```

### Errors and status codes

`read_query` and `get_query_response` raise FastAPI's `HTTPException`:

| Status | When | Detail |
| --- | --- | --- |
| 400 | The client sent a query the builder cannot run: malformed parameters (`filters[n]=4&filters[n][$eq]=5`), an invalid field path in `filters`, `sort`, `fields` or `distinct` (`filters[na me][$eq]=x`, or a segment that starts with `$`, `sort[0]=$name:asc`), a filter field with a plain value instead of operators (`filters[status]=active`), an operator that Supported Filter Operators does not list, anywhere in `filters` (`filters[name][$regex]=x`), `$or`/`$and` as a field's operator (`filters[name][$or]=x`), `$between` without exactly 2 values, `$or`/`$and` that is not a list of conditions, a sort item that is not `field:order` (`sort[0]=name`), a sort order other than `asc`/`desc` (`sort[0]=name:1`), overlapping distinct fields, overlapping `fields` or populate fields (`fields[0]=spec&fields[1]=spec.color`), an invalid path in `populate[...][fields]` (`populate[users][fields][0]=$where`), a populate name that is a list or an object (`populate[0][0]=x`), a `populate` that is neither a list nor an object together with `fields`, an operator inside another operator's value, a parameter or `filters` nested more than 32 levels, or invalid pagination | The reason, for example `Invalid field path 'na me' in filters` or `Operator '$regex' is not allowed in filters` |
| 400 | A protected user field in filters, sort or distinct (see Protected User Fields) | `Field 'hashed_password' is protected …` |
| 500 | Anything else: MongoDB or Redis unreachable, a relation that cannot be looked up and was set after the `QueryBuilder` was created, a bug | `Internal server error`; the error and its traceback are logged on `oplnk_python_utils.query.query_builder` |

The builder methods raise `QueryValidationError` for the 400 cases. It is a `ValueError` exported from `oplnk_python_utils.query`; raise it in your overrides to answer 400 with your own message. `pagination[page]` (default 1) and `pagination[pageSize]` (default 10) must be integers of at least 1, and strings such as `"5"` are read as integers. There is no maximum page size by default; set one per subclass:

```python
class CRUD(QueryBuilder):
    max_page_size = 500  # pageSize=501 answers 400; the default page size is capped at 500
```

### Logging

The query module logs on `oplnk_python_utils.query.query_builder` (a child of `oplnk_python_utils.query`) and never prints:

-  ERROR: `Query failed: <error>`, with the traceback, for every 500.
-  WARNING: something the query still skips, for example `Skipping populate name 10 in the fields projection: it is not a string` or `Query count cache skipped: …`.
-  DEBUG: diagnostics, such as `Query rejected: <reason>` for each `QueryValidationError` and `Distinct field …` records for distinct queries. With `read_query(..., debug=True)` it also logs the input, the parsed query, the data and count pipelines, the count cache key, the count, the first document as the client gets it (after `sanitize_result`) and the timings.

`debug=True` only produces DEBUG records. To see them, enable DEBUG for the module, for example `logging.getLogger("oplnk_python_utils.query").setLevel(logging.DEBUG)` with a handler that prints DEBUG (`logging.basicConfig(level=logging.DEBUG)` in development).

## Integration with Existing CRUD

### Before (Original CRUD + Endpoints)

```python
# CRUD Class - 30+ lines of complex query logic
class CRUD:
    def __init__(self, collection, relations=[]):
        # ... original implementation

    def read_query(self, query, pipeline=None, pipeline_count=None, use_cache=False):
        # ... 30+ lines of complex caching and query logic

# Endpoint - Manual parameter parsing
from src.utils import parse_params_to_dict

@router.get("/query")
async def get_data(request: Request):
    params = request.query_params
    query = parse_params_to_dict(params)  # Manual parsing
    return crud.read_query(query, pipeline=base_pipeline, use_cache=True)
```

### After (Using Utils Package)

```python
# CRUD Class - Now super clean!
from oplnk_python_utils.query import QueryBuilder

class CRUD(QueryBuilder):
    def __init__(self, collection, relations=[]):
        self.collection = collection
        self.db = db[self.collection]
        super().__init__(collection_prefix=PREFIX, relations=relations)

    def read_query(self, params=None, query=None, pipeline=None, pipeline_count=None):
        # Just 4 lines! All complexity abstracted away
        return super().read_query(
            db_collection=self.db, params=params, query=query,
            pipeline=pipeline, pipeline_count=pipeline_count
        )

# Endpoint - With new endpoint-level caching!
from oplnk_python_utils.cache import ainvalidate_tag, cache_factory
from oplnk_python_utils.core.parsing import parse_params_to_dict

# Create reusable cache decorator for this collection
users_cache = cache_factory(redis_client, "myapp", "users")

@router.get("/query")
@users_cache(
    endpoint_type="read",
    tags=["users", "queries"],
    ttl=3600  # Cache for 1 hour
)
async def get_data(request: Request):
    # Option 1: Manual parsing (explicit control)
    query = parse_params_to_dict(request.query_params)
    return crud.read_query(query=query, pipeline=base_pipeline)

    # Option 2: Automatic parsing (convenience)
    return crud.read_query(
        params=request.query_params,  # Parsed automatically
        pipeline=base_pipeline
    )

# Write operations are not cached; they invalidate the cached reads themselves
@router.post("/users")
async def create_user(request: Request, user_data: dict):
    result = crud.create(user_data)
    await ainvalidate_tag(redis_client, "users")  # the next GET reads the users again
    return result
```

## Advanced Features

### Custom Relation Definitions

```python
relations = [
    # Basic relation: [Type, field_name, prefix, collection_name]
    # populate[0]=user writes the related document under "users", the collection name
    ["User", "user_id", "myapp", "users"],

    # Array relation
    ["Role", "role_ids", "myapp", "roles"],

    # Custom alias: [Type, field_name, prefix, collection_name, alias]
    ["Department", "dept_id", "myapp", "departments", "department"]
]
```

The field must end with `_id` (one id) or `_ids` (a list of ids); only that suffix is removed to get the populate name (`user_id` → `user`). The populated document is written under the alias when given, else under the collection name when given, else under the populate name.

Relations are checked when the `QueryBuilder` is created. A field with no `_id` in its name and no collection raises `ValueError`, for example `["Users", "person", "", ""]`: `Relation 'Users' cannot be looked up: its field 'person' does not end with '_id' or '_ids' and it declares no collection (fourth element)`. With a collection, such a field is read as one id of that collection, its populate name is the field itself, and a WARNING is logged when the relation is declared. The id field is kept when the document is written under another key; when it is written under the field itself (`("Users", "person", None, "users", "person")`), it replaces the id, and an id that matches no document drops the key. In that case populate replaces the id before filters run, so `populate[0]=person` with `filters[person][$eq]=<id>` matches nothing; filter on the populated document's fields instead (`filters[person][full_name][$eq]=<name>`). To populate the user who created or modified a document, use the reserved audit fields below.

A `fields` path under a populated key (`fields[1]=establishment.name`, `fields[1]=creation_user_info.full_name`) keeps only that path of the populated object.

### Reserved Audit Fields (v1.2.1)

`RESERVED_AUDIT_FIELDS`, exported from `oplnk_python_utils.query`, holds each audit field's collection, alias and default projection.

The QueryBuilder automatically handles common audit fields without requiring manual configuration:

```python
# No need to define these in relations anymore:
query_builder = QueryBuilder(
    collection_prefix="myapp",
    relations=[
        # creation_user and modification_user are automatically handled!
        ["User", "user_id", "myapp", "users"],
    ]
)

# Usage - just populate them:
query = {
    "populate": ["creation_user", "modification_user"]
}

# Result will include (original ID removed, replaced by populated object):
# {
#     "data": [{
#         "_id": "123",
#         "creation_user_info": {  // Populated object
#             "_id": "user_id_string",
#             "name": "John Doe",
#             "person": "person_id"
#         }
#     }]
# }
```

**Reserved fields:**

-  `creation_user` → automatically populated as `creation_user_info`, original ID is excluded
-  `modification_user` → automatically populated as `modification_user_info`, original ID is excluded

These fields are only populated when explicitly requested via the `populate` parameter, following the Strapi philosophy of on-demand loading.

### Protected User Fields

`PROTECTED_USER_FIELDS` (`hashed_password` and `active_directory`) never leave the QueryBuilder:

-  They are removed from every lookup the QueryBuilder builds (reserved audit fields, relations, the count pipeline) and from the top level of the main query's documents, even when a client lists them in `fields` or `populate[...][fields]`.
-  Filtering, sorting or requesting distinct values on a protected field returns HTTP 400, for example `filters[users][hashed_password][$startsWith]=...` or `distinct[hashed_password]=1`.
-  Filters accept only the documented operators (see Supported Filter Operators), so expression operators such as `$expr`, `$where`, `$function` and `$accumulator`, which can read any field, return HTTP 400, also when the protection is turned off.

Other user fields, such as `email`, `role` and names, can still be requested with `populate[...][fields]`. The default projection of the reserved audit fields is unchanged.

Override the list in your subclass to add or remove fields (a single string is one field; an empty tuple or `None` turns the protection off, while filters still accept only the documented operators):

```python
from oplnk_python_utils.query import PROTECTED_USER_FIELDS, QueryBuilder

class CRUD(QueryBuilder):
    protected_user_fields = (*PROTECTED_USER_FIELDS, "otp_secret")
```

The check works on field names, in every collection, not only in `users`: any filter, sort or distinct path with a protected name in it is rejected, and top-level fields with a protected name are removed. Two things stay your responsibility: nested copies in other collections (for example `owner.hashed_password`) are not removed from results, and lookups you add yourself through the `pipeline` argument need their own projection.

### Populate Field Sanitization (v1.2.1)

The behavior of original ID fields depends on whether they are being populated:

#### Without populate - ID field shown as-is (from database):

```python
# No populate - field shown normally from database
query = {
    "fields": ["name", "establishment_id"]
}

# Response:
# {
#     "data": [{
#         "_id": "123",
#         "name": "Product",
#         "establishment_id": "est_123"  // Shown normally
#     }]
# }
```

#### With populate - ID field removed by default:

```python
# With populate but no explicit fields request
query = {
    "populate": ["establishment"]
}

# Response - establishment_id automatically excluded:
# {
#     "data": [{
#         "_id": "123",
#         "name": "Product",
#         "establishment": {  // Populated object
#             "_id": "est_123",
#             "name": "Main Office"
#         }
#     }]
# }
```

#### Related document not found - null, ID kept:

```python
query = {"populate": ["establishment"]}
# A document whose establishment_id points to nothing:
# { "_id": "124", "name": "Product", "establishment": null, "establishment_id": "est_missing" }
```

Audit fields behave the same (`creation_user_info: null` keeps `creation_user`), including values that are not an ObjectId such as `"system"`. A list of ids (`*_ids`) is removed whenever it is populated, unless `fields` lists it. `populate[x]=false` leaves the document as stored.

#### With populate AND explicit fields request - show both:

```python
# Request both populate and the original ID field explicitly
query = {
    "populate": ["establishment"],
    "fields": ["name", "establishment_id"]  # Explicitly request the ID
}

# Response - both are present:
# {
#     "data": [{
#         "_id": "123",
#         "name": "Product",
#         "establishment_id": "est_123",  // Shown because explicitly requested
#         "establishment": {
#             "_id": "est_123",
#             "name": "Main Office"
#         }
#     }]
# }
```

#### Audit fields behavior:

```python
# Without populate - original field shown
query = {
    "fields": ["name", "creation_user"]
}
# Response:
# { "creation_user": "user_123" }

# With populate - original field removed, *_info shown instead
query = {
    "populate": ["creation_user"]
}
# Response:
# { "creation_user_info": { "_id": "user_123", ... } }
# Note: creation_user field is NOT shown

# With populate AND explicit request for original - show both
query = {
    "populate": ["creation_user"],
    "fields": ["creation_user"]  # Explicitly request the ID
}
# Response:
# { "creation_user": "user_123", "creation_user_info": { "_id": "user_123", ... } }
```

### Custom Pipelines

```python
# Custom base pipeline
base_pipeline = [
    {"$match": {"deleted": False}},
    {"$lookup": {...}}
]

results = query_builder.get_query_response(
    db_collection=collection,
    query=query,
    pipeline=base_pipeline,  # Custom data pipeline
    pipeline_count=custom_count_pipeline  # Custom count pipeline
)
```

`meta.totalCount` counts the documents `data` returns across all pages, or its groups for `distinct`. A filter on a related document (`filters[establishment][name][$eq]=...`) needs that relation in `populate`, in both data and count. When `pipeline` adds lookups of its own that filters read and no `pipeline_count` is given, the count looks those relations up by name, as in 1.3.0; passing a `pipeline_count` with the same lookups is the exact form.

### Smart Text Search

Text operators match string fields with an escaped regex. When the value looks like a number (`10`, `-5`, `10.5`) they also match numeric fields through the number's decimal text:

-  `$containsi=10` matches strings containing "10" and the numbers 10, 100, 210, 1050 and 10.5.
-  That text is the one MongoDB `$toString` gives (no leading zeros, no trailing fraction zeros, no `.0`), so `$containsi=10.0` does not match 10 and `$startsWith=05` does not match 5. A `Decimal128` keeps its trailing zeros: `10.0` stored as `Decimal128` matches `$containsi=10.0`.
-  `$startsWith` and `$endsWith` compare the start or end of that text.
-  `$eqi` and `$nei` compare numbers by value.
-  `$notContains`, `$notContainsi` and `$nei` exclude both the matching strings and the matching numbers.
-  Words such as `inf` are text.

Matching numbers through their text uses `$expr` (MongoDB 4.2 or later) and cannot use an index.

```python
filters = {
    "amount": {"$containsi": "10"}
}
```

## Migrating from 1.3.0

`tools/migrate` is a codemod that moves a backend from oplnk-python-utils 1.3 or earlier to 1.5.0. Every breaking change a 1.3.0 service faces is 1.4.0's; 1.4.0 → 1.5.0 adds no breaking change, only the optional `oplnk-python-utils[sql]` extra, so the same procedure lands on 1.5.0. It lives in this repository and is not part of the published package. Run it from a checkout of this repository; it needs `libcst`, which the dev extras install. The step-by-step procedure, including what the codemod cannot see, is [`docs/migration-1.4.md`](docs/migration-1.4.md):

```bash
# list what would change; nothing is written
uv run --with libcst python -m tools.migrate ../oplnk-spv2-core
# exit 1 when changes are pending (for CI); nothing is written
uv run --with libcst python -m tools.migrate ../oplnk-spv2-core --check
# apply the changes and write both reports
uv run --with libcst python -m tools.migrate ../oplnk-spv2-core --write \
    --report migration.md --json migration.json
```

-  `--write` refuses a repository whose git working tree has uncommitted or untracked files, so the result is one reviewable `git diff`; `--allow-dirty` overrides this. Listing and `--check` never write, and work outside git.
-  `--write` replaces each file atomically (a temporary file in the same directory, then a rename), keeping its permissions. A file that cannot be written (read-only, deleted, full disk) is printed as `ERROR  <path>: <reason>` and keeps its old content; its changes are listed under it as not applied and are not counted, in the console and in both reports. The other files are still written.
-  The codemod is idempotent: a second `--write` changes nothing.
-  Python files keep their formatting and comments (libcst rewrites only the nodes it changes). `requirements*.txt` files keep their encoding (UTF-8, or UTF-8 or UTF-16 with a BOM) and line endings. Files it cannot read, parse or decode are skipped and printed as `SKIP <path>: <reason>`, and a rewrite that would not parse leaves its file unchanged and skipped. A symlink is never read or written through, even when it points inside the repository: it is skipped as `symlink (not followed)`, so every change stays inside the repository and in its `git diff`.
-  Hidden directories, `node_modules`, `site-packages` and virtualenvs are not read.
-  Every change and every flag is printed as `path:line` with its rule id. `--report` writes the same list as Markdown and `--json` as JSON, with the skipped files and the rules. Reports are replaced atomically like the other files; a report path that is a symlink is refused (exit 2).
-  Exit codes: 0 done, 1 `--check` found pending changes, 2 refused before anything was written (missing directory, dirty tree, not a git repository, a `--report` or `--json` path whose directory does not exist or is not writable), 3 a file or report could not be written. Flags and skipped files do not fail `--check`, since template files such as `init/*_place_holder.py` never parse; read the `SKIP` lines in its output. An edit that leaves a file's bytes as they were is not a change.

Rules are listed in `tools/migrate/registry.py`, which runs them in this order. An `auto` rule changes files and may also flag what it cannot decide; a `flag` rule only reports:

| Rule | Kind | What it does |
| --- | --- | --- |
| `D6-methods` | auto | A cache decorator on a POST, PUT, PATCH or DELETE route gets `methods=["<METHOD>"]`, since 1.4.0 caches only GET and HEAD by default. It also flags the route: confirm it is read-only, or remove the decorator. Only routers created with `APIRouter(...)` or `FastAPI(...)` in the same module count; a decorator such as an imported router's `post` or `mock.patch`, and methods it cannot read, are flagged instead. |
| `D4-ttl-zero` | auto | `ttl=0` on a cache decorator becomes `use_cache=False`. Other literal TTLs that 1.4.0 rejects (negative, float, bool, string) are flagged. |
| `D1-ainvalidate` | auto | An `invalidate_tag(...)` or `invalidate_cache_keys(...)` statement inside an `async def` becomes `await ainvalidate_tag(...)` or `await ainvalidate_cache_keys(...)`, and the async name is added to its `from oplnk_python_utils.cache import ...`. The sync name is removed from that import once nothing uses it; it stays when `__all__` lists it or another module of the repository imports it from this one. Calls in sync functions stay; other uses in an `async def` are flagged. |
| `REQ-bump` | auto | Pins `oplnk-python-utils` to `==1.5.0` in every `requirements*.txt`, keeping the file's encoding (UTF-8, or UTF-8/UTF-16 with a BOM) and line endings. A pin it cannot compare (a pre-release such as `==1.5.1rc1`) or above 1.5.0 is flagged and left as it is. Flags `redis` below 5.0, `fastapi` below 0.104 and `pymongo` below 4.6, Python below 3.10 (`Dockerfile`, `.python-version`, `runtime.txt`), the Redis server requirement (4.0 or later, with `EVAL`/`EVALSHA` allowed), and, for a pin before 1.3.0, the populate changes since then. |
| `DEV-fakeredis-lua` | auto | `fakeredis` becomes `fakeredis[lua]`, so tests run the cache's Lua scripts. |
| `D10-operators` | flag | A literal filter passed to `read_query` or `get_query_response` (`{"filters": {...}}` or `"filters[x][$op]"` keys) that uses an operator outside the documented list plus `$nin`, such as `$regex`, `$not` or `$expr`: 1.4.0 answers HTTP 400. Filters built at run time are not seen. |
| `D12-relations` | flag | A relation declared in `relations=[...]` or `relations = [...]`, in a module that imports the library or a `QueryBuilder` subclass of the repository, whose field has no `_id`/`_ids`. Without a collection, 1.4.0 raises `ValueError` when the `QueryBuilder` is created; with one, it logs a WARNING and `populate` returns the related document: review the fields it exposes. |
| `C07-reports` | flag | `import oplnk_python_utils.reports` or `from oplnk_python_utils import reports`: the module no longer exists. |
| `C01-lookup-override` | flag | A class method `get_lookup_relation` or `get_count_lookup_relation`: the library's adds `_protected_fields_stages()`, which an override skips. |
| `B01-redis-timeout` | flag | A sync `redis.Redis(...)`, `StrictRedis(...)` or `from_url(...)` without `socket_timeout`: the library sets none, and 1.4.0 logs a WARNING once per such client used by a cached endpoint. |
| `C11-parse-params` | flag | A direct call of `parse_params_to_dict`, whose lists follow their index, repeated keys give lists, and malformed keys raise `ValueError`. |
| `A04-builders` | flag | A direct call of a builder method (`build_filter_pipeline`, `make_aggregation_query`, ...), which raises `QueryValidationError` instead of skipping invalid items. |
| `A08-debug` | flag | `debug=True` on `read_query` or `get_query_response`: diagnostics are DEBUG log records instead of prints. |

Review the diff and every flag before committing.

## Dependencies

All four are required, even if you only use `core`, because importing the package also imports `cache` and `query` (C-08).

-  `fastapi>=0.104`: `QueryBuilder` raises its `HTTPException`; the cache decorator uses its `Request`, `jsonable_encoder` and `response_model` serialization.
-  `pymongo>=4.6`: only the `bson` package it installs (`ObjectId`); the library never imports `pymongo` itself.
-  `starlette>=0.27`: the cache's `Request`/`Response` types. fastapi 0.104 already requires starlette 0.27.x, so this adds no constraint.
-  `redis>=5.0`: the cache's sync and `redis.asyncio` clients and errors. The retry detection needs redis-py 4.4.0 or later; CI tests 5.0.4 and 5.2.1, the versions the services pin, and the latest release.

The SQL compiler/adapter is optional: `oplnk-python-utils[sql]` adds `sqlalchemy[asyncio]>=2.0`. The base package, the Mongo path and `import oplnk_python_utils` do not import SQLAlchemy; database drivers (`aioodbc`, `asyncpg`, `aiomysql`, `aiosqlite`) belong to the application that chooses the engine. See `docs/query-sql-adapter.md`.

## License

This package is licensed under the MIT License. See the [LICENSE](LICENSE) file for details.

## Running the tests

The suite runs MongoDB and Redis as real servers in Docker through testcontainers, so Docker must be running. Tests that only need in-memory Redis use fakeredis and do not start any container; the dev extras install `fakeredis[lua]` (lupa) so the cache's Lua scripts run there too.

```bash
uv venv --python 3.11 .venv
uv pip install --python .venv/bin/python -e ".[dev]"
.venv/bin/pytest --cov
```

CI (`.github/workflows/tests.yml`) runs the same suite on Python 3.10 to 3.13 with the latest dependencies, plus one job with the versions consumer services pin (fastapi 0.110.0, pymongo 4.6.2, redis 5.2.1). A job fails when line and branch coverage drops below 90%.

Tests marked `xfail(strict=True)` describe the correct behavior for a known defect from `docs/audit/2026-10-audit.md`; the marker's reason names the finding. Fixing the defect makes the test pass, and strict mode then fails the run until the marker is removed.

## Releasing

`.github/workflows/release.yml` publishes the package. Pushing a `v*` tag publishes to PyPI; running the workflow manually publishes to TestPyPI first (`pypi` on the input for the real one). It checks that the tag matches `__version__`, builds the wheel and sdist, runs `twine check --strict`, and only then uploads that exact artifact.

Authentication is an API token from the repository secrets: `PYPI_API_TOKEN` (pypi.org) and `TESTPYPI_API_TOKEN` (test.pypi.org). A version can only be published once, so re-running a published tag fails by design. When the account can be configured for Trusted Publishing, the tokens can be dropped for OIDC without changing the flow.

To publish by hand instead, build locally and upload with the credentials in `~/.pypirc`:

```bash
python -m build
python -m twine check --strict dist/*
python -m twine upload dist/*
```

## Contributing

This package is developed and maintained by OpenLink SpA. For questions or support, please contact [contacto@openlink.cl](mailto:contacto@openlink.cl).

## Repository

Source code is available at: [https://github.com/openlinkspa/oplnk-python-utils](https://github.com/openlinkspa/oplnk-python-utils)
