FastAPI Caching and Background Tasks¶

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from redis import asyncio as aioredis

@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost:6379", encoding="utf8")
    FastAPICache.init(RedisBackend(redis), prefix="cache")

@router.get("/hotels")
@cache(expire=30)  # cache for 30 seconds
async def get_hotels():
    hotels = await HotelDAO.find_all()
    # Convert to Pydantic BEFORE cache decorator processes response
    return parse_obj_as(list[SHotel], hotels)

from celery import Celery

celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

@celery_app.task
def process_image(path: str):
    img = Image.open(path)
    img.save(f"processed_{path}")
    return f"Processed: {path}"

@celery_app.task
def send_email(to: str, subject: str, body: str):
    # Email sending logic
    return f"Sent to {to}"

@app.post("/upload")
async def upload(file: UploadFile):
    path = save_file(file)
    task = process_image.delay(path)  # async to Celery
    return {"task_id": task.id}

@app.get("/task/{task_id}")
async def get_status(task_id: str):
    from celery.result import AsyncResult
    result = AsyncResult(task_id)
    return {
        "status": result.status,  # PENDING, STARTED, SUCCESS, FAILURE
        "result": result.result if result.ready() else None,
    }

# Retry on failure
@celery_app.task(bind=True, max_retries=3)
def fetch_data(self, url):
    try:
        return requests.get(url).json()
    except Exception as exc:
        self.retry(exc=exc, countdown=60)

# Task chains and groups
from celery import chain, group
chain(task1.s(arg), task2.s())()        # sequential pipeline
group(task.s(i) for i in range(10))()   # parallel execution

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

@app.post("/items/")
async def create_item(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "Item created")
    return {"status": "created"}

from queue import Queue, LifoQueue, PriorityQueue

q = Queue()           # FIFO
q.put("first"); q.get()  # "first"

stack = LifoQueue()   # LIFO (stack)
stack.put("a"); stack.put("b"); stack.get()  # "b"

pq = PriorityQueue()  # lowest value first
pq.put((3, "low")); pq.put((1, "high")); pq.get()  # (1, "high")