Introduction
Asynchronous cache manager designed for horizontally scaled web applications. NOTE: Currently has implementation only for FastAPI using Redis.
Requirements
Python 3.7+
Installation
$ pip install atomcache
---> 100%
Explanation schema
As UML
@startuml
!theme materia
participant Redis
participant Instance_A as A
participant Instance_B as B
participant Instance_N as C
B <-> Redis: GET: Cache=null & GET: Lock=null
B <-> Redis: SET: Lock = true
activate B #Red
A <--> Redis: GET: Cache=null & GET: Lock=true
activate A #Transparent
C <--> Redis: GET: Cache=null & GET: Lock=true
activate C #Transparent
B <--> B: Do the computation
B -> Redis: SET: Cache={...}
deactivate B
group Notify Cache SET
Redis -> C
Redis -> A
end
group GET Cache
Redis <-> C
deactivate C
Redis <-> A
deactivate A
end
@enduml
Examples:
Usage as FastAPI Dependency
- Create a file
events.pywith:
Callable: async def stop_app() -> None: await Cache.backend.close() return stop_app ">
from typing import Optional, Callable import aioredis from fastapi import FastAPI, Depends from atomcache import Cache def create_start_app_handler(app: FastAPI) -> Callable: async def start_app() -> None: redis: aioredis.Redis = await aioredis.from_url(url="redis://localhost", encoding="utf-8") await Cache.init(app, redis) return start_app def create_stop_app_handler(app: FastAPI) -> Callable: async def stop_app() -> None: await Cache.backend.close() return stop_app
- Create a file
main.pywith:
from typing import Optional
from fastapi import FastAPI, Depends
from atomcache import Cache
from .events import create_start_app_handler, create_stop_app_handler
app = FastAPI()
app.add_event_handler("startup", create_start_app_handler(app))
app.add_event_handler("shutdown", create_stop_app_handler(app))
@router.get("/resources", response_model=List[TheResponseModel], name="main:test-example")
async def resources(offset: int = 0, items: int = 10, cache: Cache = Depends(Cache(exp=600)):
cache_id = f"{offset}-{items}" # Build cache identifier
await cache.raise_try(cache_id) # Try to respond from cache
response = await db.find(TheResponseModel, skip=offset, limit=items)
await asyncio.sleep(10) # Do some heavy work for 10 sec, see `lock_timeout`
return cache.set(response, cache_id=cache_id)
Direct cache usage for avoiding repetitive calling on external resources:
List[dict]: cached_value = await cache.get(cache_id=ref) if cached_value is not None: return cached_value async with ClientSession() as session: async with session.get(f"https://external-api.io/{ref}") as response: if response.ok: cached_value = response.json() return cache.set(cached_value, cache_id=ref) return [] ">
from aiohttp import ClientSession from atomcache import Cache cache = Cache(exp=1200, namespace="my-namespace:") async def requesting_helper(ref: str) -> List[dict]: cached_value = await cache.get(cache_id=ref) if cached_value is not None: return cached_value async with ClientSession() as session: async with session.get(f"https://external-api.io/{ref}") as response: if response.ok: cached_value = response.json() return cache.set(cached_value, cache_id=ref) return []
20.2k Jan 9, 2023
9 Dec 21, 2022
3 Dec 13, 2022
2 Nov 4, 2021
4 Aug 16, 2021
106 Dec 16, 2022
1 Nov 7, 2021
1 Dec 19, 2021
11 Nov 30, 2022
1 Jul 17, 2022
1 Feb 5, 2022
1 Feb 13, 2022
1 Jan 16, 2022
1 Jan 3, 2022
1 Feb 10, 2022
5.7k Jan 8, 2023
32 Mar 5, 2022
20.9k Jan 1, 2023
1.4k Jan 1, 2023
1.7k Jan 5, 2023
15 Nov 27, 2022
1k Dec 15, 2022
304 Nov 7, 2022
5 Jul 9, 2022
92 Dec 13, 2022
2 Aug 14, 2022
77 Oct 29, 2022
6 Oct 8, 2022
2k Jan 2, 2023