Metadata-Version: 2.1
Name: PGCacheWatch
Version: 0.2.1
Summary: A Python library for real-time PostgreSQL event-driven cache invalidation.
Author: janbjorge
License: Apache 2.0
Project-URL: Documentation, https://github.com/janbjorge/pgcachewatch/
Project-URL: Homepage, https://github.com/janbjorge/pgcachewatch/
Project-URL: Issues, https://github.com/janbjorge/pgcachewatch/issues
Project-URL: Repository, https://github.com/janbjorge/pgcachewatch/
Classifier: Development Status :: 4 - Beta
Classifier: Environment :: Other Environment
Classifier: Environment :: Web Environment
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Natural Language :: English
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Utilities
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: asyncpg>=0.27.0
Requires-Dist: pydantic>=2.0.0
Provides-Extra: dev
Requires-Dist: asyncpg-stubs; extra == "dev"
Requires-Dist: fastapi; extra == "dev"
Requires-Dist: httpx; extra == "dev"
Requires-Dist: mypy-extensions; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: ruff; extra == "dev"

# PGCacheWatch
[![CI](https://github.com/janbjorge/PGCacheWatch/actions/workflows/ci.yml/badge.svg)](https://github.com/janbjorge/PGCacheWatch/actions/workflows/ci.yml?query=branch%3Amain)
[![pypi](https://img.shields.io/pypi/v/PGCacheWatch.svg)](https://pypi.python.org/pypi/PGCacheWatch)
[![downloads](https://static.pepy.tech/badge/PGCacheWatch/month)](https://pepy.tech/project/PGCacheWatch)
[![versions](https://img.shields.io/pypi/pyversions/PGCacheWatch.svg)](https://github.com/janbjorge/PGCacheWatch)

PGCacheWatch is a Python library crafted to empower applications with real-time PostgreSQL event notifications for efficient cache invalidation, directly leveraging existing PostgreSQL infrastructure. This approach eliminates the need for additional technologies for cache management, simplifying your stack while enhancing performance and real-time data consistency.

## Key Advantages
- **Leverage Existing Infrastructure**: Utilizes PostgreSQL's native NOTIFY/LISTEN capabilities for event-driven cache invalidation, avoiding the overhead of integrating external caching systems.
- **Asynchronous and Efficient**: Built on top of `asyncpg` for asynchronous database communication, ensuring non-blocking I/O operations and optimal performance.
- **Flexible Cache Invalidation Strategies**: Offers a variety of strategies (e.g., Greedy, Windowed, Timed) for nuanced cache invalidation control, tailored to different application needs.
- **Simple Yet Powerful API**: Designed with simplicity in mind, offering a straightforward setup process and an intuitive API for managing cache invalidation logic.

## Installation
To install PGCacheWatch, run the following command in your terminal:
```bash
pip install pgcachewatch
```

## Using PGCacheWatch
### Setting Up
Initialize PostgreSQL triggers to emit NOTIFY events on data changes. PGCacheWatch provides utility scripts for easy trigger setup
```bash
pgcachewatch install <tables-to-cache>
```

### FastAPI Example
Example showing how to use PGCacheWatch for cache invalidation in a FastAPI app

```python
import contextlib
import typing

import asyncpg
from fastapi import FastAPI
from pgcachewatch import decorators, listeners, models, strategies

listener = listeners.PGEventQueue()


@contextlib.asynccontextmanager
async def app_setup_teardown(_: FastAPI) -> typing.AsyncGenerator[None, None]:
    conn = await asyncpg.connect()
    await listener.connect(conn, models.PGChannel("ch_pgcachewatch_table_change"))
    yield
    await conn.close()


APP = FastAPI(lifespan=app_setup_teardown)


# Only allow for cache refresh after an update
@decorators.cache(
    strategy=strategies.Gready(
        listener=listener,
        predicate=lambda x: x.operation == "update",
    )
)
async def cached_query() -> dict[str, str]:
    # Simulate a database query
    return {"data": "query result"}


@APP.get("/data")
async def get_data() -> dict:
    return await cached_query()
```
