mirror of
https://github.com/mealie-recipes/mealie.git
synced 2026-02-16 04:43:13 -05:00
bdaf758712
* fix ts types * feat(code-generation): ♻️ update code-generation formats * new scope * add step button * fix linter error * update code-generation tags * feat(backend): ✨ start multi-tenant support * feat(backend): ✨ group invitation token generation and signup * refactor(backend): ♻️ move group admin actions to admin router * set url base to include `/admin` * feat(frontend): ✨ generate user sign-up links * test(backend): ✅ refactor test-suite to further decouple tests (WIP) * feat(backend): 🐛 assign owner on backup import for recipes * fix(backend): 🐛 assign recipe owner on migration from other service Co-authored-by: hay-kot <hay-kot@pm.me>
123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
import json
|
|
import shutil
|
|
from zipfile import ZipFile
|
|
|
|
from fastapi import Depends, File
|
|
from fastapi.datastructures import UploadFile
|
|
from scrape_schema_recipe import scrape_url
|
|
from sqlalchemy.orm.session import Session
|
|
from starlette.responses import FileResponse
|
|
|
|
from mealie.core.dependencies import temporary_zip_path
|
|
from mealie.core.root_logger import get_logger
|
|
from mealie.db.database import db
|
|
from mealie.db.db_setup import generate_session
|
|
from mealie.routes.routers import UserAPIRouter
|
|
from mealie.schema.recipe import CreateRecipeByURL, Recipe, RecipeImageTypes
|
|
from mealie.schema.recipe.recipe import CreateRecipe, RecipeSummary
|
|
from mealie.services.image.image import write_image
|
|
from mealie.services.recipe.recipe_service import RecipeService
|
|
from mealie.services.scraper.scraper import create_from_url
|
|
|
|
user_router = UserAPIRouter()
|
|
logger = get_logger()
|
|
|
|
|
|
@user_router.get("", response_model=list[RecipeSummary])
|
|
async def get_all(start=0, limit=None, service: RecipeService = Depends(RecipeService.private)):
|
|
return service.get_all(start, limit)
|
|
|
|
|
|
@user_router.post("", status_code=201, response_model=str)
|
|
def create_from_name(data: CreateRecipe, recipe_service: RecipeService = Depends(RecipeService.private)) -> str:
|
|
""" Takes in a JSON string and loads data into the database as a new entry"""
|
|
return recipe_service.create_one(data).slug
|
|
|
|
|
|
@user_router.post("/create-url", status_code=201, response_model=str)
|
|
def parse_recipe_url(url: CreateRecipeByURL, recipe_service: RecipeService = Depends(RecipeService.private)):
|
|
""" Takes in a URL and attempts to scrape data and load it into the database """
|
|
|
|
recipe = create_from_url(url.url)
|
|
return recipe_service.create_one(recipe).slug
|
|
|
|
|
|
@user_router.post("/test-scrape-url")
|
|
def test_parse_recipe_url(url: CreateRecipeByURL):
|
|
# TODO: Replace with more current implementation of testing schema
|
|
return scrape_url(url.url)
|
|
|
|
|
|
@user_router.post("/create-from-zip")
|
|
async def create_recipe_from_zip(
|
|
session: Session = Depends(generate_session),
|
|
temp_path=Depends(temporary_zip_path),
|
|
archive: UploadFile = File(...),
|
|
):
|
|
""" Create recipe from archive """
|
|
|
|
with temp_path.open("wb") as buffer:
|
|
shutil.copyfileobj(archive.file, buffer)
|
|
|
|
recipe_dict = None
|
|
recipe_image = None
|
|
|
|
with ZipFile(temp_path) as myzip:
|
|
for file in myzip.namelist():
|
|
if file.endswith(".json"):
|
|
with myzip.open(file) as myfile:
|
|
recipe_dict = json.loads(myfile.read())
|
|
elif file.endswith(".webp"):
|
|
with myzip.open(file) as myfile:
|
|
recipe_image = myfile.read()
|
|
|
|
recipe: Recipe = db.recipes.create(session, Recipe(**recipe_dict))
|
|
|
|
write_image(recipe.slug, recipe_image, "webp")
|
|
|
|
return recipe
|
|
|
|
|
|
@user_router.get("/{slug}", response_model=Recipe)
|
|
def get_recipe(recipe_service: RecipeService = Depends(RecipeService.read_existing)):
|
|
""" Takes in a recipe slug, returns all data for a recipe """
|
|
return recipe_service.item
|
|
|
|
|
|
@user_router.get("/{slug}/zip")
|
|
async def get_recipe_as_zip(
|
|
slug: str, session: Session = Depends(generate_session), temp_path=Depends(temporary_zip_path)
|
|
):
|
|
""" Get a Recipe and It's Original Image as a Zip File """
|
|
recipe: Recipe = db.recipes.get(session, slug)
|
|
|
|
image_asset = recipe.image_dir.joinpath(RecipeImageTypes.original.value)
|
|
|
|
with ZipFile(temp_path, "w") as myzip:
|
|
myzip.writestr(f"{slug}.json", recipe.json())
|
|
|
|
if image_asset.is_file():
|
|
myzip.write(image_asset, arcname=image_asset.name)
|
|
|
|
return FileResponse(temp_path, filename=f"{slug}.zip")
|
|
|
|
|
|
@user_router.put("/{slug}")
|
|
def update_recipe(data: Recipe, recipe_service: RecipeService = Depends(RecipeService.write_existing)):
|
|
""" Updates a recipe by existing slug and data. """
|
|
|
|
return recipe_service.update_one(data)
|
|
|
|
|
|
@user_router.patch("/{slug}")
|
|
def patch_recipe(data: Recipe, recipe_service: RecipeService = Depends(RecipeService.write_existing)):
|
|
""" Updates a recipe by existing slug and data. """
|
|
|
|
return recipe_service.patch_one(data)
|
|
|
|
|
|
@user_router.delete("/{slug}")
|
|
def delete_recipe(recipe_service: RecipeService = Depends(RecipeService.write_existing)):
|
|
""" Deletes a recipe by slug """
|
|
return recipe_service.delete_one()
|