feat: helper class for Redis

fastapi_extras/databases/redis.py

@@ -0,0 +1,17 @@
+from typing import AsyncGenerator, Union
+
+import redis.asyncio as redis
+from pydantic import RedisDsn
+
+
+class RedisManager:
+    def __init__(self, url: Union[RedisDsn, str]):
+        self.pool = redis.ConnectionPool.from_url(str(url))
+
+    async def __call__(self) -> AsyncGenerator[redis.Redis, None]:
+        cli = redis.Redis(connection_pool=self.pool)
+
+        try:
+            yield cli
+        finally:
+            await cli.aclose()

requirements-dev.txt

@@ -3,6 +3,7 @@ build==1.2.1
 pytest==8.2.0
 pytest-cov==5.0.0
 pytest-xdist==3.6.1
+redis==5.0.4
 ruff==0.4.4
 setuptools==69.5.1
 twine==5.0.0

tests/databases/test_redis.py

@@ -0,0 +1,79 @@
+from typing import Union
+
+import pytest
+from fastapi import Depends, FastAPI, HTTPException, status
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+from pytest import MonkeyPatch
+from redis.asyncio import Redis
+from typing_extensions import Annotated
+
+from fastapi_extras.databases.redis import RedisManager
+
+
+class FakeRedis:
+    db = {}
+    closed = []
+
+    async def get(self, key: str) -> Union[bytes, None]:
+        return FakeRedis.db.get(key)
+
+    async def set(self, key: str, val: str):
+        FakeRedis.db[key] = val.encode()
+
+    async def aclose(self):
+        FakeRedis.closed.append(id(self))
+
+
+class Item(BaseModel):
+    key: str
+    val: str
+
+
+app = FastAPI()
+redis_manager = RedisManager("redis://localhost:6379/0")
+
+
+@app.get("/items/{key}", response_model=Item)
+async def read(key: str, redis: Annotated[Redis, Depends(redis_manager)]):
+    item = await redis.get(key)
+
+    if not item:
+        HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
+
+    return Item(key=key, val=item)
+
+
+@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
+async def write(item: Item, redis: Annotated[Redis, Depends(redis_manager)]):
+    await redis.set(item.key, item.val)
+    return item
+
+
+client = TestClient(app)
+
+
+@pytest.fixture(autouse=True)
+def redis_mock(monkeypatch: MonkeyPatch):
+    monkeypatch.setattr("redis.asyncio.Redis.get", FakeRedis.get)
+    monkeypatch.setattr("redis.asyncio.Redis.set", FakeRedis.set)
+    monkeypatch.setattr("redis.asyncio.Redis.aclose", FakeRedis.aclose)
+
+
+def test_redis_manager():
+    data = [
+        {"key": "foo", "val": "bar"},
+        {"key": "bar", "val": "baz"},
+        {"key": "baz", "val": "foo"},
+    ]
+
+    for item in data:
+        response = client.post("/items/", json=item)
+        assert response.status_code == 201
+        assert response.json() == item
+
+        response = client.get(f"/items/{item['key']}")
+        assert response.status_code == 200
+        assert response.json() == item
+
+    assert len(set(FakeRedis.closed)) == len(data) * 2