oplnk-python-utils 1.0.0

Reusable utilities for caching, data manipulation, parsing, and MongoDB query building

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

Installation

Install from PyPI:

pip install oplnk-python-utils

Or install in development mode:

pip install -e /path/to/oplnk-python-utils

Add to your requirements.txt:

oplnk-python-utils>=1.0.0

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.

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"]}

# 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
• merge_dicts(*dicts): Intelligently merge multiple dictionaries with nested structure support

Parsing Utilities (core.parsing)

Functions for parsing query parameters and type conversion.

from oplnk_python_utils.core.parsing import (
    parse_params_to_dict,
    infer_type,
    parse_key_string_to_list,
    parse_list_to_dict
)

# 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(["1", "2"])      # Returns: [1, 2] (list with converted types)

# 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
• infer_type(value): Smart type inference from string values (supports int, float, bool, datetime, lists)
• 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

Serialization Utilities (core.serialization)

Functions for converting data types to serializable formats.

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)
# All datetime objects converted to ISO strings recursively
# Non-datetime values remain unchanged

Serialization Functions

• datetime_serializer(data): Recursively convert datetime objects to ISO format strings in any data structure

Cache Module

Advanced caching utilities with Redis integration and intelligent cache key management.

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 MD5 hash

# Get from cache or execute function and cache result
async def expensive_operation(query):
    # Some expensive database operation
    return await db.collection.find(query).to_list()

result = await get_or_set_cache(
    cache_key="user_query_abc123",
    fetch_function=expensive_operation,
    query={"status": "active"},
    expiration=3600  # 1 hour
)

# Cache with automatic serialization of datetime objects
from datetime import datetime
async def get_user_data():
    return {
        "user": "john",
        "last_login": datetime.now(),
        "preferences": {"theme": "dark"}
    }

cached_data = await get_or_set_cache(
    cache_key="user_john_data",
    fetch_function=get_user_data,
    expiration=1800
)

# Invalidate multiple cache keys by pattern
invalidated_count = await invalidate_cache_keys("user_*")
print(f"Invalidated {invalidated_count} cache entries")

Cache Functions

• generate_cache_id(data, prefix=""): Generate MD5 hash from normalized data for consistent cache keys
• **get_or_set_cache(cache_key, fetch_function, expiration=3600, **kwargs)**: Get cached result or execute function and cache with automatic datetime serialization
• invalidate_cache_keys(pattern): Delete cache keys matching pattern (supports wildcards)

Query Module

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

Usage

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=True
)

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

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, ...): Complete query method with parameter parsing

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 section for detailed documentation of these utilities.

Supported Filter Operators

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
• $between (requires array with 2 elements)

Logical

• $and, $or

Existence

• $exists, $notExists
• $null, $notNull

Arrays

• $listContains, $listNotContains

Query Structure

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

Integration with Existing CRUD

Before (Original CRUD + Endpoints)

# 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)

# 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, use_cache=False):
        # Just 6 lines! All complexity abstracted away
        return super().read_query(
            db_collection=self.db, params=params, query=query,
            pipeline=pipeline, pipeline_count=pipeline_count,
            use_cache=use_cache, cache_client=redis,
            cache_key_prefix=f"app:{self.collection}"
        )

# Endpoint - Even simpler with explicit imports!
from oplnk_python_utils.core.parsing import parse_params_to_dict

@router.get("/query")
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, use_cache=True)

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

Advanced Features

Custom Relation Definitions

relations = [
    # Basic relation: [Type, field_name, prefix, 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"],

    # Special Users relation (built-in handling)
    ["Users", "person", "", ""]
]

Custom Pipelines

# 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
)

Smart Text Search

The query builder automatically handles mixed-type searching:

# Search for "10" will match:
# - String fields containing "10"
# - Numeric fields equal to 10
# - Numeric fields in range 10.0-10.999
# - Numbers starting with 10 (100, 1000, etc.)
filters = {
    "amount": {"$containsi": "10"}
}

Dependencies

• pymongo: MongoDB driver
• redis: Redis client
• fastapi: For HTTP exceptions
• bson: For ObjectId handling

License

This package is licensed under the MIT License. See the LICENSE file for details.

Contributing

This package is developed and maintained by OpenLink SpA. For questions or support, please contact [e-mail].

Repository

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