iframe-proxy | Sunbelt Computer Software

126 lines (100 loc) · 4.5 KB
import redis
from fastapi import FastAPI, HTTPException, Request
from models import ItemPayload
app = FastAPI()
redis_client = redis.StrictRedis(host="0.0.0.0", port=6379, db=0, decode_responses=True)
@app.get("/")
def home(request: Request) -> dict[str, str]:
    url: str = (
        f"https://{os.getenv('CODESPACE_NAME')}-8000.{os.getenv('GITHUB_CODESPACES_PORT_FORWARDING_DOMAIN')}/"
        if os.getenv("CODESPACE_NAME")
        else str(request.base_url)
    return {
        "message": f"Navigate to the following URL to access the Swagger UI: {url}docs"
# Route to add an item
@app.post("/items/{item_name}/{quantity}")
def add_item(item_name: str, quantity: int) -> dict[str, ItemPayload]:
    if quantity <= 0:
        raise HTTPException(status_code=400, detail="Quantity must be greater than 0.")
    # Check if item already exists
    item_id_str: str | None = redis_client.hget("item_name_to_id", item_name)
    if item_id_str is not None:
        item_id = int(item_id_str)
        quantity = redis_client.hincrby(f"item_id:{item_id}", "quantity", quantity)
        # Generate an id for the item
        item_id: int = redis_client.incr("item_ids")
        redis_client.hset(
            f"item_id:{item_id}",
            mapping={
                "item_id": item_id,
                "item_name": item_name,
                "quantity": quantity,
        # Create a set so we can search by name too
        redis_client.hset("item_name_to_id", item_name, item_id)
    return {
        "item": ItemPayload(item_id=item_id, item_name=item_name, quantity=quantity)
# Route to list a specific item by id but using Redis
@app.get("/items/{item_id}")
def list_item(item_id: int) -> dict[str, dict[str, str]]:
    if not redis_client.hexists(f"item_id:{item_id}", "item_id"):
        raise HTTPException(status_code=404, detail="Item not found.")
        return {"item": redis_client.hgetall(f"item_id:{item_id}")}
@app.get("/items")
def list_items() -> dict[str, list[ItemPayload]]:
    items: list[ItemPayload] = []
    stored_items: dict[str, str] = redis_client.hgetall("item_name_to_id")
    for name, id_str in stored_items.items():
        item_id: int = int(id_str)
        item_name_str: str | None = redis_client.hget(f"item_id:{item_id}", "item_name")
        if item_name_str is not None:
            item_name: str = item_name_str
        else:
            continue  # skip this item if it has no name
        item_quantity_str: str | None = redis_client.hget(
            f"item_id:{item_id}", "quantity"
        if item_quantity_str is not None:
            item_quantity: int = int(item_quantity_str)
        else:
            item_quantity = 0
        items.append(
            ItemPayload(item_id=item_id, item_name=item_name, quantity=item_quantity)
    return {"items": items}
# Route to delete a specific item by id but using Redis
@app.delete("/items/{item_id}")
def delete_item(item_id: int) -> dict[str, str]:
    if not redis_client.hexists(f"item_id:{item_id}", "item_id"):
        raise HTTPException(status_code=404, detail="Item not found.")
        item_name: str | None = redis_client.hget(f"item_id:{item_id}", "item_name")
        redis_client.hdel("item_name_to_id", f"{item_name}")
        redis_client.delete(f"item_id:{item_id}")
        return {"result": "Item deleted."}
# Route to remove some quantity of a specific item by id but using Redis
@app.delete("/items/{item_id}/{quantity}")
def remove_quantity(item_id: int, quantity: int) -> dict[str, str]:
    if not redis_client.hexists(f"item_id:{item_id}", "item_id"):
        raise HTTPException(status_code=404, detail="Item not found.")
    item_quantity: str | None = redis_client.hget(f"item_id:{item_id}", "quantity")
    # if quantity to be removed is higher or equal to item's quantity, delete the item
    if item_quantity is None:
        existing_quantity: int = 0
        existing_quantity: int = int(item_quantity)
    if existing_quantity <= quantity:
        item_name: str | None = redis_client.hget(f"item_id:{item_id}", "item_name")
        redis_client.hdel("item_name_to_id", f"{item_name}")
        redis_client.delete(f"item_id:{item_id}")
        return {"result": "Item deleted."}
        redis_client.hincrby(f"item_id:{item_id}", "quantity", -quantity)
        return {"result": f"{quantity} items removed."}
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

main.py

Latest commit

History

main.py

File metadata and controls
