Spaces:
Sleeping
Sleeping
File size: 2,948 Bytes
5043970 b4a1d6c d12a716 28a462a d12a716 5043970 28a462a 5043970 |
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 |
# uvicorn app:app --host localhost --port 7860 --reload
from library.librerias import *
from routers.routers import *
from pydantic import BaseModel
import sqlite3
app = FastAPI(
title="API de la heladería Anhelados",
description="Esta es la API de la microempresa Anhelados, la cual se dedica a la venta de helados y postres. Aquí se pueden realizar operaciones CRUD para los recursos de la empresa.",
version="2.8.7",
openapi_url="/anhelados-openapi.json",
docs_url="/anhelados-docs",
redoc_url="/anhelados-redoc",
contact={
"name": "Anhelados",
"url": "https://www.anhelados.com",
"email": "contacto@anhelados.com",
},
license_info={"name": "MIT", "url": "https://opensource.org/licenses/MIT"},
)
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# saludo
@app.get("/", summary="Saludo", description="Saludo de la API")
def read_root():
return {"message": "Bienvenido a la API de la heladeria"}
# Modelo para recibir consultas SQL
class SQLQuery(BaseModel):
query: str
# Conexión a la base de datos SQLite
def get_db_connection():
conn = sqlite3.connect('database/anhelados.db')
conn.row_factory = sqlite3.Row
return conn
@app.post("/execute", summary="Ejecutar consulta SQL", description="Ejecuta una consulta SQL en la base de datos SQLite")
def execute_query(sql_query: SQLQuery):
conn = get_db_connection()
try:
# Limpiar la consulta para eliminar '\nObservation' y reemplazar comillas escapadas
cleaned_query = sql_query.query.replace("\nObservation", "").replace("\\'", "'")
print("Solicitud POST completa:", sql_query.json())
print("Consulta recibida en el servidor:", cleaned_query)
cursor = conn.cursor()
cursor.execute(cleaned_query)
conn.commit()
if cursor.description: # Verifica si la consulta tiene resultados
results = cursor.fetchall()
return {"results": [dict(row) for row in results]}
return {"message": "Consulta ejecutada exitosamente"}
except sqlite3.Error as e:
raise HTTPException(status_code=400, detail=str(e))
finally:
conn.close()
# Routers
app.router.include_router(almacen.router)
app.router.include_router(clientes.router)
app.router.include_router(detalle_pedidos.router)
app.router.include_router(detalle_ventas.router)
app.router.include_router(empleados.router)
app.router.include_router(gastos_imprevistos.router)
app.router.include_router(gastos.router)
app.router.include_router(insumos.router)
app.router.include_router(maquinas.router)
app.router.include_router(pedidos_proveedor.router)
app.router.include_router(produccion.router)
app.router.include_router(productos.router)
app.router.include_router(registro_energetico.router)
app.router.include_router(ventas.router)
|