Created
October 18, 2023 00:53
-
-
Save Filimoa/a639a0e77ba0cea842021f32113b1739 to your computer and use it in GitHub Desktop.
Fastapi Redis Dependency
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| Much of this was borrowed from: | |
| https://github.com/Tert0/fastapi-framework/blob/develop/fastapi_framework/redis.py | |
| Assumes redis = "^4.5.1" | |
| Meant to be used like this: | |
| @app.on_event("startup") | |
| async def on_startup(): | |
| await redis_dependency.init() | |
| @app.get("/set/{key}/{value}") | |
| async def test(key: str, value: str, redis: Redis = Depends(redis_dependency)): | |
| await redis.set(key, value) | |
| return "Done" | |
| """ | |
| from typing import Any, Optional, Set | |
| from redis import asyncio as aioredis | |
| from src.lib.core.config import settings | |
| class RedisBackend: | |
| redis_connection: aioredis.Redis | |
| @staticmethod | |
| async def init(url: str) -> "RedisBackend": | |
| redis = RedisBackend() | |
| redis.redis_connection = await aioredis.Redis.from_url(url) | |
| return redis | |
| async def get(self, key: str): | |
| """Get Value from Key""" | |
| return await self.redis_connection.get(key) | |
| async def set( | |
| self, | |
| key: str, | |
| value: str, | |
| expire: Optional[int] = None, | |
| pexpire: Optional[int] = None, | |
| ): | |
| """Set Key to Value""" | |
| return await self.redis_connection.set( | |
| name=key, | |
| value=value, | |
| ex=expire, | |
| px=pexpire, | |
| ) | |
| async def pttl(self, key: str) -> int: | |
| """Get PTTL from a Key""" | |
| return int(await self.redis_connection.pttl(key)) | |
| async def ttl(self, key: str) -> int: | |
| """Get TTL from a Key""" | |
| return int(await self.redis_connection.ttl(key)) | |
| async def pexpire(self, key: str, pexpire: int) -> bool: | |
| """Sets and PTTL for a Key""" | |
| return bool(await self.redis_connection.pexpire(key, pexpire)) | |
| async def expire(self, key: str, expire: int) -> bool: | |
| """Sets and TTL for a Key""" | |
| return bool(await self.redis_connection.expire(key, expire)) | |
| async def incr(self, key: str) -> int: | |
| """Increases an Int Key""" | |
| return int(await self.redis_connection.incr(key)) | |
| async def decr(self, key: str) -> int: | |
| """Decreases an Int Key""" | |
| return int(await self.redis_connection.decr(key)) | |
| async def delete(self, key: str): | |
| """Delete value of a Key""" | |
| return await self.redis_connection.delete(key) | |
| async def smembers(self, key: str) -> Set: | |
| """Gets Set Members""" | |
| return set(await self.redis_connection.smembers(key)) | |
| async def sadd(self, key: str, value: Any) -> bool: | |
| """Adds a Member to a Dict""" | |
| return bool(await self.redis_connection.sadd(key, value)) | |
| async def srem(self, key: str, member: Any) -> bool: | |
| """Removes a Member from a Set""" | |
| return bool(await self.redis_connection.srem(key, member)) | |
| async def exists(self, key: str) -> bool: | |
| """Checks if a Key exists""" | |
| return bool(await self.redis_connection.exists(key)) | |
| class RedisDependency: | |
| """FastAPI Dependency for Redis Connections""" | |
| redis: Optional[RedisBackend] = None | |
| async def __call__(self): | |
| if self.redis is None: | |
| await self.init() | |
| return self.redis | |
| async def init(self, url: Optional[str] = None): | |
| """Initialises the Redis Dependency""" | |
| self.redis = await RedisBackend.init(url or settings.CELERY_BROKER_URL) | |
| redis_dependency: RedisDependency = RedisDependency() | |
| async def get_redis() -> RedisBackend: | |
| """Returns a NEW Redis connection""" | |
| return await RedisBackend.init(settings.CELERY_BROKER_URL) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment