Desarrollo de una API REST para gestión de ofertas flash con FastAPI y Redis

redis_client.set('key', 'value')
value = redis_client.get('key')
print(value)  # Output: value

redis_client.set('counter', 1)
redis_client.incr('counter')
print(redis_client.get('counter'))  # Output: 2
redis_client.decr('counter')
print(redis_client.get('counter'))  # Output: 1

redis_client.lpush('list', 'value1')
redis_client.rpush('list', 'value2')
list_items = redis_client.lrange('list', 0, -1)
print(list_items)  # Output: ['value1', 'value2']

redis_client.hset('hash', mapping={'field1': 'value1', 'field2': 'value2'})
field1 = redis_client.hget('hash', 'field1')
field2 = redis_client.hget('hash', 'field2')
print(field1)  # Output: value1
print(field2)  # Output: value2

{
    "product_id": "123",
    "discount": "30",
    "stock": 100,
    "expires_in": 3600
}

{
    "user_id": "456"
}

import time
from contextlib import asynccontextmanager
from pathlib import Path

import redis.exceptions
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse

from app.models import OfferCreate, Reservation
from app.redis_client import get_redis_client

redis_client = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_client
    redis_client = get_redis_client() (1)
    yield
    redis_client.close()

app = FastAPI(title="API Ofertas Flash", version="1.0.0", lifespan=lifespan)

@app.get("/", response_class=HTMLResponse)
def root():
    html_path = Path(__file__).parent.parent / "public" / "index.html"
    return html_path.read_text(encoding="utf-8") (2)

@app.get("/api/test")
def test():
    return {"status": 200, "message": "API is working"}

HSET offer:123 product_id 123 discount 30 stock 100 expires_at 1717609200

@app.post("/api/offers", status_code=200)
def create_offer(offer: OfferCreate):
    offer_key = f"offer:{offer.product_id}"
    expires_at = int(time.time()) + offer.expires_in

    redis_client.hset(offer_key, mapping={ (1)
        "product_id": offer.product_id,
        "discount":   offer.discount,
        "stock":      offer.stock,
        "expires_at": expires_at,
    })
    redis_client.expire(offer_key, offer.expires_in)

    return {
        "status": 200,
        "message": "Offer created successfully",
        "offer": {
            "product_id": offer.product_id,
            "discount":   offer.discount,
            "stock":      offer.stock,
            "expires_at": expires_at,
        }
    }

HGETALL offer:123

@app.get("/api/offers/{product_id}")
def get_offer(product_id: str):
    offer_key = f"offer:{product_id}"

    if not redis_client.exists(offer_key):
        raise HTTPException(status_code=404, detail="Offer not found") (1)

    offer          = redis_client.hgetall(offer_key)
    time_remaining = redis_client.ttl(offer_key)

    return {
        "status": 200,
        "offer":  offer,
        "time_remaining": time_remaining,
    }

WATCH offer:123
stock = HGET offer:123 stock
if stock > 0
    MULTI
    HINCRBY offer:123 stock -1
    LPUSH offer:123:reservations user_456
    EXEC

@app.post("/api/offers/{product_id}/reserve")
def reserve_offer(product_id: str, reservation: Reservation):
    offer_key = f"offer:{product_id}"

    if not redis_client.exists(offer_key):
        raise HTTPException(status_code=404, detail="Offer not found")

    with redis_client.pipeline() as pipe: (1)
        try:
            pipe.watch(offer_key)
            stock = int(redis_client.hget(offer_key, "stock") or 0)

            if stock <= 0:
                raise HTTPException(status_code=400, detail="Out of stock")

            pipe.multi() (2)
            pipe.hincrby(offer_key, "stock", -1)
            pipe.lpush(f"{offer_key}:reservations", f"user_{reservation.user_id}")
            pipe.execute() (3)

            return {
                "status":          200,
                "message":         "Product reserved successfully",
                "remaining_stock": stock - 1,
            }

        except redis.exceptions.WatchError: (4)
            raise HTTPException(
                status_code=500,
                detail="Failed to reserve product, please try again"
            )

LRANGE offer:123:reservations 0 -1

@app.get("/api/offers/{product_id}/reservations")
def get_reservations(product_id: str):
    reservations_key = f"offer:{product_id}:reservations"

    if not redis_client.exists(reservations_key):
        raise HTTPException(
            status_code=404,
            detail="No reservations found for this offer"
        )

    reservations = redis_client.lrange(reservations_key, 0, -1)

    return {"status": 200, "reservations": reservations}

DEL offer:123
DEL offer:123:reservations

@app.delete("/api/offers/{product_id}")
def delete_offer(product_id: str):
    offer_key = f"offer:{product_id}"

    if not redis_client.exists(offer_key):
        raise HTTPException(status_code=404, detail="Offer not found")

    redis_client.delete(offer_key)
    redis_client.delete(f"{offer_key}:reservations")

    return {"status": 200, "message": "Offer deleted successfully"}