Metadata-Version: 2.4
Name: pydantic-typed-cache
Version: 0.1.0
Summary: Async cache library for Pydantic models without FastAPI dependencies
Author-email: Wonderweeks Company <dev@mom-mom.net>
License: MIT
Project-URL: Homepage, https://github.com/mom-mom/pydantic-cache
Project-URL: Documentation, https://github.com/mom-mom/pydantic-cache#readme
Project-URL: Repository, https://github.com/mom-mom/pydantic-cache.git
Project-URL: Issues, https://github.com/mom-mom/pydantic-cache/issues
Project-URL: Inspiration, https://github.com/long2ice/fastapi-cache
Keywords: cache,pydantic,async,redis,asyncio,decorator
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Framework :: Pydantic
Classifier: Framework :: AsyncIO
Requires-Python: <4.0,>=3.13
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic>=2.11.7
Requires-Dist: redis>=5.0.0
Requires-Dist: pendulum>=3.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: ruff>=0.12.8; extra == "dev"
Requires-Dist: pre-commit>=3.0.0; extra == "dev"
Dynamic: license-file

# Pydantic Typed Cache

An async cache library for Pydantic models without FastAPI dependencies. This library provides a simple decorator-based caching mechanism for async functions that return Pydantic models or other Python objects.

> **Note**: This project was inspired by [fastapi-cache](https://github.com/long2ice/fastapi-cache) but designed to work independently of FastAPI/Starlette, making it suitable for any async Python application.

## Features

- 🚀 Simple decorator-based caching for async functions
- 🔄 Support for both async and sync functions (sync functions run in thread pool)
- 📦 Multiple backends: Redis, In-Memory
- 🎯 Type-safe with Pydantic model support
- 🔑 Customizable cache key generation
- 📝 Multiple serialization options (JSON, Pickle)
- ⚡ Zero FastAPI/Starlette dependencies
- 🔧 Flexible type conversion with `model` parameter
- ✅ Support for nullable/optional types with proper None handling
- 🎭 Support for Union types and complex type hints

## Installation

```bash
pip install pydantic-typed-cache
```

## Quick Start

```python
import asyncio
from pydantic import BaseModel
from pydantic_cache import PydanticCache, cache
from pydantic_cache.backends.inmemory import InMemoryBackend

# Define a Pydantic model
class User(BaseModel):
    id: int
    name: str
    email: str

# Initialize cache
backend = InMemoryBackend()
PydanticCache.init(backend, prefix="myapp", expire=60)

# Cache a function
@cache(expire=120, namespace="users")
async def get_user(user_id: int) -> User:
    # Expensive operation (e.g., database query)
    return User(id=user_id, name="John", email="john@example.com")

# Use the cached function
async def main():
    user = await get_user(1)  # First call - cache miss
    user = await get_user(1)  # Second call - cache hit
    
asyncio.run(main())
```

## Advanced Features

### Type Conversion with `model` Parameter

The `model` parameter allows you to force type conversion of cached values. This works with any Python type, not just Pydantic models:

```python
# Convert string to int
@cache(model=int)
async def get_count() -> str:
    return "42"  # Will be converted to int 42

# Convert dict to Pydantic model
class UserResponse(BaseModel):
    id: int
    name: str

@cache(model=UserResponse)
async def get_user_data(user_id: int) -> dict:
    # Returns dict that will be converted to UserResponse
    return {"id": user_id, "name": "Alice"}

# Work with collections
@cache(model=list[int])
async def get_scores() -> list[str]:
    return ["95", "87", "92"]  # Converted to [95, 87, 92]

# Support Union types
@cache(model=int | str)
async def get_flexible_value(use_number: bool) -> any:
    return 123 if use_number else "hello"

# Support Optional types
@cache(model=int | None)
async def get_optional_value(value: str | None) -> str | None:
    return value  # "123" becomes 123, None stays None
```

### Nullable/Optional Type Support

The library properly handles None values in optional types, distinguishing between cached None values and cache misses:

```python
# Function returning Optional types
@cache(namespace="users")
async def get_user(user_id: int) -> User | None:
    if user_id < 0:
        return None  # This None will be cached
    return User(id=user_id, name="John", email="john@example.com")

# First call with -1
user = await get_user(-1)  # Returns None (cache miss, stores None)
user = await get_user(-1)  # Returns None (cache hit, retrieves None)

# Pydantic models with optional fields
class Profile(BaseModel):
    id: int
    bio: str | None = None
    age: int | None = None

@cache(namespace="profiles")
async def get_profile(user_id: int) -> Profile:
    return Profile(id=user_id, bio=None, age=None)  # None fields are properly cached
```

### Sync Function Support

Sync functions are automatically wrapped and run in a thread pool:

```python
@cache(namespace="compute")
def expensive_computation(x: int, y: int) -> int:
    # This sync function will be run in a thread pool
    import time
    time.sleep(1)
    return x * y

# Can be called as async
result = await expensive_computation(10, 20)
```

## Backends

### In-Memory Backend

Perfect for development and testing:

```python
from pydantic_cache.backends.inmemory import InMemoryBackend

backend = InMemoryBackend()
PydanticCache.init(backend)
```

### Redis Backend

For production use with persistence:

```python
from redis.asyncio import Redis
from pydantic_cache.backends.redis import RedisBackend

redis = Redis(host="localhost", port=6379)
backend = RedisBackend(redis)
PydanticCache.init(backend)
```

## Configuration

### Global Configuration

```python
PydanticCache.init(
    backend=backend,
    prefix="myapp",        # Prefix for all cache keys
    expire=300,            # Default expiration in seconds
    coder=JsonCoder,       # Serialization method (JsonCoder or PickleCoder)
    key_builder=my_key_builder,  # Custom key builder function
    enable=True            # Enable/disable caching globally
)
```

### Per-Decorator Configuration

```python
@cache(
    expire=120,           # Override default expiration
    namespace="users",    # Namespace for this function
    coder=PickleCoder,   # Override default coder
    key_builder=custom_key_builder,  # Custom key builder
    model=UserModel      # Force type conversion
)
async def cached_function():
    pass
```

## Serialization

### JsonCoder (Default)

- Human-readable cache values
- Good for debugging
- Supports most Python types and Pydantic models
- Smaller cache size

### PickleCoder

- Supports all Python objects
- Faster serialization
- Binary format (not human-readable)
- Better for complex nested structures

```python
from pydantic_cache.coder import JsonCoder, PickleCoder

# Set globally
PydanticCache.init(backend, coder=PickleCoder)

# Or per decorator
@cache(coder=JsonCoder)
async def my_function():
    pass
```

## Cache Management

```python
# Clear specific key
await PydanticCache.clear(key="specific_key")

# Clear entire namespace
await PydanticCache.clear(namespace="users")

# Clear all cache
await PydanticCache.clear()

# Disable caching temporarily
PydanticCache.set_enable(False)

# Re-enable caching
PydanticCache.set_enable(True)

# Check if caching is enabled
is_enabled = PydanticCache.get_enable()
```

## Custom Key Builder

Create custom cache key generation logic:

```python
from pydantic_cache.types import KeyBuilder

def custom_key_builder(
    func,
    namespace: str,
    args: tuple,
    kwargs: dict
) -> str:
    # Custom logic to generate cache key
    func_name = func.__name__
    args_str = "_".join(str(arg) for arg in args)
    return f"{namespace}:{func_name}:{args_str}"

# Use globally
PydanticCache.init(backend, key_builder=custom_key_builder)

# Or per decorator
@cache(key_builder=custom_key_builder)
async def my_function():
    pass
```

## Examples

### Complex Type Conversions

```python
# Nested structures
@cache(model=list[list[int]])
async def get_matrix() -> list[list[str]]:
    return [["1", "2"], ["3", "4"]]  # Converted to [[1, 2], [3, 4]]

# List of Pydantic models
class Item(BaseModel):
    id: int
    name: str
    price: float

@cache(model=list[Item])
async def get_items() -> list[dict]:
    return [
        {"id": "1", "name": "Item 1", "price": "9.99"},
        {"id": "2", "name": "Item 2", "price": "19.99"}
    ]

# Complex Union types
@cache(model=User | dict | None)
async def get_flexible_data(data_type: str) -> any:
    if data_type == "user":
        return {"id": 1, "name": "Alice", "email": "alice@example.com"}
    elif data_type == "dict":
        return {"key": "value"}
    else:
        return None
```

### Model-to-Model Conversion

```python
class DetailedUser(BaseModel):
    id: int
    name: str
    email: str
    age: int
    bio: str

class SimpleUser(BaseModel):
    id: int
    name: str

# Convert DetailedUser to SimpleUser
@cache(model=SimpleUser)
async def get_simple_user(user_id: int) -> DetailedUser:
    return DetailedUser(
        id=user_id,
        name="Alice",
        email="alice@example.com",
        age=30,
        bio="Developer"
    )
    # Result will be SimpleUser with only id and name
```

### Working with External APIs

```python
class APIResponse(BaseModel):
    status: str
    data: dict
    timestamp: str | None = None

# Force API responses to be validated as Pydantic models
@cache(model=APIResponse, expire=300, namespace="api")
async def fetch_from_api(endpoint: str) -> dict:
    # Make actual API call here
    return {
        "status": "success",
        "data": {"result": "some data"},
        "timestamp": "2024-01-01T12:00:00Z"
    }

# Result is always validated as APIResponse model
response = await fetch_from_api("/users")
print(response.status)  # Type-safe access
```

## Testing

```python
import pytest
from pydantic_cache import PydanticCache
from pydantic_cache.backends.inmemory import InMemoryBackend

@pytest.fixture
async def cache_setup():
    backend = InMemoryBackend()
    PydanticCache.init(backend, prefix="test", expire=60)
    yield
    await backend.clear()

async def test_caching(cache_setup):
    call_count = 0
    
    @cache(namespace="test")
    async def get_value():
        nonlocal call_count
        call_count += 1
        return "result"
    
    result1 = await get_value()
    result2 = await get_value()
    
    assert result1 == result2
    assert call_count == 1  # Called only once due to caching
```

## Acknowledgments

This project was inspired by [fastapi-cache](https://github.com/long2ice/fastapi-cache) by @long2ice. While fastapi-cache provides excellent caching capabilities for FastAPI applications, pydantic-typed-cache was created to offer similar functionality for general async Python applications without the FastAPI/Starlette dependency.

## License

MIT
