Spaces:
Sleeping
Sleeping
File size: 3,029 Bytes
96f792c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 |
from fastapi import APIRouter, Depends, HTTPException, status, File, UploadFile, Form
import os
from sqlalchemy.orm import Session
from typing import List
from app.database import get_db
from app import models, schemas
from app.auth import get_current_admin_user
router = APIRouter(prefix="/admin", tags=["admin"])
@router.post("/users/{user_id}/promote", response_model=schemas.UserResponse)
def promote_user_to_admin(
user_id: int,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_admin_user)
):
user = db.query(models.User).filter(models.User.id == user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
user.role = models.UserRole.ADMIN
db.commit()
db.refresh(user)
return user
@router.post("/products", response_model=schemas.ProductResponse, status_code=status.HTTP_201_CREATED)
def create_product(
name: str = Form(...),
brand: str = Form(None),
calories: float = Form(...),
protein: float = Form(...),
fat: float = Form(...),
carbohydrates: float = Form(...),
sodium: float = Form(...),
sugar: float = Form(...),
fiber: float = Form(None),
cholesterol: float = Form(None),
serving_size: str = Form(None),
image: UploadFile = File(None),
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_admin_user)
):
image_path = None
if image:
# Create .tmp directory if it doesn't exist
os.makedirs(".tmp", exist_ok=True)
file_location = f".tmp/{image.filename}"
with open(file_location, "wb+") as file_object:
file_object.write(image.file.read())
image_path = file_location
product_data = {
"name": name,
"brand": brand,
"calories": calories,
"protein": protein,
"fat": fat,
"carbohydrates": carbohydrates,
"sodium": sodium,
"sugar": sugar,
"fiber": fiber,
"cholesterol": cholesterol,
"serving_size": serving_size,
"image_path": image_path
}
db_product = models.Product(**product_data)
db.add(db_product)
db.commit()
db.refresh(db_product)
return db_product
@router.get("/products", response_model=List[schemas.ProductResponse])
def list_products(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_admin_user)
):
products = db.query(models.Product).offset(skip).limit(limit).all()
return products
@router.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(
product_id: int,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_admin_user)
):
product = db.query(models.Product).filter(models.Product.id == product_id).first()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
db.delete(product)
db.commit()
return None
|