147 lines
No EOL
5.1 KiB
Python
147 lines
No EOL
5.1 KiB
Python
import sqlite3
|
|
import products, recipes, db, meals, persons
|
|
import datetime
|
|
|
|
from pydantic import BaseModel
|
|
from typing import List, Annotated
|
|
from fastapi import FastAPI, Depends, Query
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
app = FastAPI()
|
|
|
|
# Add CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"]
|
|
)
|
|
|
|
# Dependency to create SQLite connection
|
|
async def get_db():
|
|
sql_db = await db.connect()
|
|
try:
|
|
yield sql_db
|
|
finally:
|
|
await sql_db.close()
|
|
|
|
@app.get("/recipes/parse")
|
|
async def parse_recipe_handler(url: str, conn: sqlite3.Connection = Depends(get_db)) -> recipes.Recipe:
|
|
parsed = await recipes.parse_recipe(conn, url)
|
|
if not parsed:
|
|
return JSONResponse(status_code=400, content={'message': 'Recipe not found'})
|
|
return parsed
|
|
|
|
@app.get("/recipes/ingredients/parse")
|
|
async def parse_ingredients(lines: Annotated[
|
|
List[str],
|
|
Query(alias="ingredients",
|
|
title="Array of ingredients to parse")],
|
|
conn: sqlite3.Connection = Depends(get_db)) -> List[recipes.Ingredient]:
|
|
ingredients = recipes.parse_ingredient_from_nlp(lines)
|
|
await recipes.match_existing_products(conn, ingredients)
|
|
return ingredients
|
|
|
|
class ProductUrl(BaseModel):
|
|
url: str
|
|
tags: List[str] = []
|
|
|
|
@app.post("/products/")
|
|
async def create_product(url: ProductUrl, conn: sqlite3.Connection = Depends(get_db)) -> products.Product:
|
|
return await products.get_or_create(conn, url.url, url.tags)
|
|
|
|
async def load_full_recipe(conn: sqlite3.Connection, id: int) -> recipes.Recipe:
|
|
r = await recipes.find_recipe_by_id(conn, id)
|
|
if not r:
|
|
return None
|
|
|
|
r.ingredients = []
|
|
async for ingredient in recipes.find_ingredients_by_recipe_id(conn, id):
|
|
ingredient.product = await products.find_product_by_id(conn, ingredient.product_id)
|
|
r.ingredients.append(ingredient)
|
|
|
|
return r
|
|
|
|
@app.get("/recipes/")
|
|
async def get_recipes(q: str | None = None, conn: sqlite3.Connection = Depends(get_db)) -> List[recipes.Recipe]:
|
|
result = []
|
|
if q:
|
|
async for recipe in recipes.find_recipes_by_name(conn, q):
|
|
result.append(recipe)
|
|
else:
|
|
async for recipe in recipes.get_all(conn):
|
|
result.append(recipe)
|
|
|
|
for recipe in result:
|
|
recipe.ingredients = []
|
|
async for ingredient in recipes.find_ingredients_by_recipe_id(conn, recipe.id):
|
|
ingredient.product = await products.find_product_by_id(conn, ingredient.product_id)
|
|
recipe.ingredients.append(ingredient)
|
|
|
|
return result
|
|
|
|
@app.get("/recipes/{recipe_id}")
|
|
async def get_recipe(recipe_id: int, conn: sqlite3.Connection = Depends(get_db)) -> recipes.Recipe:
|
|
r = await load_full_recipe(conn, recipe_id)
|
|
if not r:
|
|
return JSONResponse(status_code=404, content={'message': 'Recipe not found'})
|
|
|
|
return r
|
|
|
|
@app.post('/recipes/')
|
|
async def create_recipe(item: recipes.Recipe, conn: sqlite3.Connection = Depends(get_db)) -> recipes.Recipe:
|
|
if not item.ingredients:
|
|
return JSONResponse(status_code=400, content={'message': 'Recipe must have at least one ingredient'})
|
|
|
|
for ingredient in item.ingredients:
|
|
if not ingredient.product:
|
|
return JSONResponse(status_code=400, content={'message': 'Ingredient must have a product'})
|
|
|
|
await recipes.insert_recipe(conn, item)
|
|
for ingredient in item.ingredients:
|
|
ingredient.recipe_id = item.id
|
|
ingredient.product_id = ingredient.product.id
|
|
await recipes.insert_ingredient(conn, ingredient)
|
|
await conn.commit()
|
|
|
|
return item
|
|
|
|
@app.get("/meals/")
|
|
async def get_meals(start: datetime.datetime, end: datetime.datetime, conn: sqlite3.Connection = Depends(get_db)) -> List[meals.Meal]:
|
|
result = []
|
|
async for meal in meals.find_meals_by_date_range(conn, start, end):
|
|
meal.recipe = await load_full_recipe(conn, meal.recipe_id)
|
|
result.append(meal)
|
|
|
|
return result
|
|
|
|
@app.post("/meals/")
|
|
async def create_meal(meal: meals.Meal, conn: sqlite3.Connection = Depends(get_db)) -> meals.Meal:
|
|
if not meal.chef:
|
|
return JSONResponse(status_code=400, content={'message': 'Meal must have at least one chef'})
|
|
|
|
if not meal.cleanup:
|
|
return JSONResponse(status_code=400, content={'message': 'Meal must have at least one cleanup person'})
|
|
|
|
if not meal.consumers:
|
|
return JSONResponse(status_code=400, content={'message': 'Meal must have at least one consumer'})
|
|
|
|
await meals.insert_meal(conn, meal)
|
|
await conn.commit()
|
|
return meal
|
|
|
|
@app.get("/persons/")
|
|
async def get_persons(conn: sqlite3.Connection = Depends(get_db)) -> List[meals.Person]:
|
|
result = []
|
|
async for person in persons.get_all(conn):
|
|
result.append(person)
|
|
|
|
return result
|
|
|
|
@app.post("/persons/")
|
|
async def create_person(person: persons.Person, conn: sqlite3.Connection = Depends(get_db)) -> persons.Person:
|
|
await persons.insert(conn, person)
|
|
await conn.commit()
|
|
return person |