Spaces:
Running
Running
from library.librerias import * | |
from models.productos import Productos | |
from fastapi import APIRouter, HTTPException, Query | |
import urllib.parse | |
from typing import List | |
router = APIRouter( | |
prefix="/productos", | |
tags=["Productos"], | |
responses={404: {"description": "No encontrado"}}, | |
) | |
"""CREATE TABLE productos ( | |
ID_Producto INTEGER PRIMARY KEY AUTOINCREMENT, | |
Nombre TEXT, | |
Precio REAL | |
);""" | |
def get_productos(): | |
try: | |
with DatabaseConnection().get_connection() as conn: | |
cursor = conn.cursor() | |
cursor.execute("SELECT * FROM productos") | |
productos = cursor.fetchall() | |
return productos | |
except Exception as e: | |
print(e) | |
return [] | |
# @router.post("/") | |
def post_producto(producto: Productos): | |
try: | |
with DatabaseConnection().get_connection() as conn: | |
cursor = conn.cursor() | |
cursor.execute( | |
"INSERT INTO productos (Nombre, Precio) VALUES (?, ?)", | |
(producto.Nombre, producto.Precio), | |
) | |
conn.commit() | |
return {"message": "Producto creado"} | |
except Exception as e: | |
print(e) | |
return [] | |
# search by Nombre order by Precio | |
def search_productos(nombre: str = Query(..., alias="Nombre")): | |
try: | |
with DatabaseConnection().get_connection() as conn: | |
cursor = conn.cursor() | |
nombre_decodificado = urllib.parse.unquote(nombre) | |
like_pattern = f"%{nombre_decodificado}%" | |
cursor.execute( | |
"SELECT ID_Producto, Nombre, Precio FROM productos WHERE Nombre LIKE ? ORDER BY Precio", | |
(like_pattern,), | |
) | |
productos = cursor.fetchall() | |
return [ | |
Productos(ID_Producto=row[0], Nombre=row[1], Precio=row[2]) | |
for row in productos | |
] | |
except Exception as e: | |
print(e) | |
raise HTTPException(status_code=500, detail="Error al buscar los productos") | |