Spaces:
Sleeping
Sleeping
File size: 4,516 Bytes
9e07f58 |
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 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 |
from fastapi import FastAPI, Depends, HTTPException, status, File, UploadFile, Form
from fastapi.responses import JSONResponse
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from database import async_session, engine
import models
import schemas
import os
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
from fastapi.staticfiles import StaticFiles
# Monta o diretório "uploads" para servir arquivos estáticos
app.mount("/uploads", StaticFiles(directory="uploads"), name="uploads")
UPLOAD_DIRECTORY = "uploads"
if not os.path.exists(UPLOAD_DIRECTORY):
os.makedirs(UPLOAD_DIRECTORY)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
async def get_db():
async with async_session() as session:
yield session
@app.on_event("startup")
async def on_startup():
async with engine.begin() as conn:
await conn.run_sync(models.Base.metadata.create_all)
from sqlalchemy import or_
# Nova rota para buscar promoções por palavra-chave
@app.get("/promocoes/search/", response_model=list[schemas.Promocao])
async def buscar_promocoes(q: str, db: AsyncSession = Depends(get_db)):
# Realiza a busca no título ou descrição
result = await db.execute(
select(models.Promocao)
.where(or_(models.Promocao.titulo.contains(q), models.Promocao.descricao.contains(q)))
)
promocoes = result.scalars().all()
return promocoes
# 1. Criar Promoção
@app.post("/empresas/{empresa_id}/promocoes/", response_model=schemas.Promocao)
async def criar_promocao(
empresa_id: int,
titulo: str = Form(...),
descricao: str = Form(...),
imagem: UploadFile = File(...),
db: AsyncSession = Depends(get_db)
):
# Validações de campos obrigatórios
if not titulo or not descricao:
raise HTTPException(status_code=400, detail="Campos de título e descrição são obrigatórios")
if not imagem:
raise HTTPException(status_code=400, detail="O arquivo de imagem é obrigatório")
# Salvar a imagem no diretório de uploads
image_location = os.path.join(UPLOAD_DIRECTORY, imagem.filename)
with open(image_location, "wb") as file:
content = await imagem.read()
file.write(content)
image_url = f"https://fabiosantos-buscaofertaapi.hf.space/uploads/{imagem.filename}"
# Criação da promoção
nova_promocao = models.Promocao(
titulo=titulo,
descricao=descricao,
imagem=image_url,
empresa_id=empresa_id
)
db.add(nova_promocao)
await db.commit()
await db.refresh(nova_promocao)
return nova_promocao
# 2. Listar Promoções de uma Empresa
@app.get("/empresas/{empresa_id}/promocoes/", response_model=list[schemas.Promocao])
async def listar_promocoes_empresa(empresa_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(models.Promocao).where(models.Promocao.empresa_id == empresa_id))
promocoes = result.scalars().all()
return promocoes
# 3. Atualizar Promoção
@app.put("/empresas/{empresa_id}/promocoes/{promocao_id}", response_model=schemas.Promocao)
async def atualizar_promocao(
empresa_id: int,
promocao_id: int,
promocao: schemas.PromocaoCreate,
db: AsyncSession = Depends(get_db)
):
result = await db.execute(select(models.Promocao).where(models.Promocao.id == promocao_id, models.Promocao.empresa_id == empresa_id))
promocao_db = result.scalars().first()
if not promocao_db:
raise HTTPException(status_code=404, detail="Promoção não encontrada.")
promocao_db.titulo = promocao.titulo
promocao_db.descricao = promocao.descricao
await db.commit()
await db.refresh(promocao_db)
return promocao_db
# 4. Deletar Promoção
@app.delete("/empresas/{empresa_id}/promocoes/{promocao_id}", status_code=status.HTTP_204_NO_CONTENT)
async def deletar_promocao(empresa_id: int, promocao_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(models.Promocao).where(models.Promocao.id == promocao_id, models.Promocao.empresa_id == empresa_id))
promocao_db = result.scalars().first()
if not promocao_db:
raise HTTPException(status_code=404, detail="Promoção não encontrada.")
await db.delete(promocao_db)
await db.commit()
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT, content={"message": "Promoção deletada com sucesso"})
|