Browse the data

1import sqlite3
2import hashlib
3import secrets
4from fastapi import FastAPI, HTTPException, Header
5from pydantic import BaseModel
6from typing import Optional, List
7 
8app = FastAPI()
9 
10# In-memory stores
11users = {}
12tokens = {}
13recipes = {}
14recipe_id_counter = 1
15 
16# SQLite setup
17conn = sqlite3.connect(":memory:", check_same_thread=False)
18conn.execute("CREATE TABLE recipes (id INTEGER PRIMARY KEY, name TEXT, difficulty TEXT, dietary_tags TEXT, ingredients TEXT, prep_time INTEGER)")
19 
20def hash_password(password: str) -> str:
21    return hashlib.sha256(password.encode()).hexdigest()
22 
23def get_user_from_token(authorization: str = Header(None)):
24    if not authorization:
25        raise HTTPException(status_code=401, detail="Missing authorization header")
26    token = authorization.replace("Bearer ", "")
27    user_id = tokens.get(token)
28    if not user_id:
29        raise HTTPException(status_code=401, detail="Invalid token")
30    return user_id
31 
32class SignupRequest(BaseModel):
33    username: str
34    password: str
35 
36class LoginRequest(BaseModel):
37    username: str
38    password: str
39 
40class RecipeCreate(BaseModel):
41    name: str
42    difficulty: str
43    dietary_tags: str
44    ingredients: str
45    prep_time: int
46 
47@app.post("/signup")
48def signup(req: SignupRequest):
49    if req.username in users:
50        raise HTTPException(status_code=400, detail="User already exists")
51    users[req.username] = hash_password(req.password)
52    return {"message": "User created"}
53 
54@app.post("/login")
55def login(req: LoginRequest):
56    if req.username not in users or users[req.username] != hash_password(req.password):
57        raise HTTPException(status_code=401, detail="Invalid credentials")
58    token = secrets.token_hex(16)
59    tokens[token] = req.username
60    return {"token": token}
61 
62@app.get("/recipes/{recipe_id}")
63def get_recipe(recipe_id: int, authorization: str = Header(None)):
64    get_user_from_token(authorization)
65    cursor = conn.execute("SELECT id, name, difficulty, dietary_tags FROM recipes WHERE id = ?", (recipe_id,))
66    row = cursor.fetchone()
67    if not row:
68        raise HTTPException(status_code=404, detail="Recipe not found")
69    return {"id": row[0], "name": row[1], "difficulty": row[2], "dietary_tags": row[3]}
70 
71@app.post("/recipes")
72def create_recipe(req: RecipeCreate, authorization: str = Header(None)):
73    get_user_from_token(authorization)
74    global recipe_id_counter
75    conn.execute("INSERT INTO recipes (id, name, difficulty, dietary_tags, ingredients, prep_time) VALUES (?, ?, ?, ?, ?, ?)",
76                 (recipe_id_counter, req.name, req.difficulty, req.dietary_tags, req.ingredients, req.prep_time))
77    conn.commit()
78    recipe_id_counter += 1
79    return {"id": recipe_id_counter - 1}
80 
81@app.get("/recipes")
82def search_recipes(ingredients: str = "", max_prep_time: int = 1000000, authorization: str = Header(None)):
83    get_user_from_token(authorization)
84    ing_list = [i.strip() for i in ingredients.split(",") if i.strip()]
85    if not ing_list:
86        cursor = conn.execute("SELECT id, name, difficulty, dietary_tags FROM recipes WHERE prep_time <= ?", (max_prep_time,))
87    else:
88        like_clauses = " AND ".join(["ingredients LIKE ?" for _ in ing_list])
89        params = [f"%{ing}%" for ing in ing_list] + [max_prep_time]
90        cursor = conn.execute(f"SELECT id, name, difficulty, dietary_tags FROM recipes WHERE {like_clauses} AND prep_time <= ?", params)
91    rows = cursor.fetchall()
92    return [{"id": r[0], "name": r[1], "difficulty": r[2], "dietary_tags": r[3]} for r in rows]

1fastapi
2uvicorn