OpticaAPI_ADS / app.py
JairoDanielMT's picture
modificacion de tablas
b90e2c0
# uvicorn app:app --host localhost --port 7860 --reload
from fastapi import Request
from urllib.parse import quote
from fastapi.staticfiles import StaticFiles
from reportlab.lib.units import cm
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from io import BytesIO
from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
from typing import List
import os
import sqlite3
import threading
from queue import Queue
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class DatabaseConnection:
"""
Clase para gestionar las conexiones a la base de datos con un pool.
"""
_instance = None
_lock = threading.Lock()
_connection_pool = Queue(maxsize=5)
def __new__(cls):
"""
Crea una nueva instancia de la clase si no existe.
"""
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance.conn = cls._instance._create_connection()
return cls._instance
def _create_connection(self):
"""
Crea una conexi贸n a la base de datos.
"""
if not self._connection_pool.empty():
return self._connection_pool.get()
else:
connection = sqlite3.connect("optica.db")
connection.row_factory = sqlite3.Row
return connection
def get_connection(self):
"""
Obtener el objeto de conexi贸n de la base de datos.
"""
return self._instance._create_connection()
def release_connection(self):
"""
Liberar la conexi贸n de nuevo al pool.
"""
if self._instance is not None:
self._connection_pool.put(self._instance.conn)
self._instance.conn = None # Marcar la instancia como sin conexi贸n
# saludo
@app.get("/")
def read_root():
return {"mensaje": "Bienvenido a la API de Optica"}
"""
CREATE TABLE Roles (
id_rol INTEGER PRIMARY KEY AUTOINCREMENT,
rol VARCHAR
);
"""
# metodo get de roles con manejo de errores
@app.get("/roles")
def get_roles():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Roles")
roles = cursor.fetchall()
return roles
except Exception as e:
print(e)
return []
# get de roles por id
@app.get("/roles/{id_rol}")
def get_rol(id_rol: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Roles WHERE id_rol = ?", [id_rol])
rol = cursor.fetchone()
if rol:
return rol
else:
raise HTTPException(status_code=404, detail="Rol no encontrado")
except Exception as e:
print(e)
return []
# busqueda por nombre de rol con like con manejo de errores y con metodo get
@app.get("/roles/busqueda/{rol}")
def get_rol_busqueda(rol: str):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Roles WHERE rol LIKE ?", ["%" + rol + "%"])
rol = cursor.fetchall()
if rol:
return rol
else:
raise HTTPException(status_code=404, detail="Rol no encontrado")
except Exception as e:
print(e)
return []
# clase y metodo post para buscar rol
class RolBusqueda(BaseModel):
rol: str
@app.post("/roles/busqueda")
def get_rol_busqueda(rol: RolBusqueda):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Roles WHERE rol LIKE ?", ["%" + rol.rol + "%"])
rol = cursor.fetchall()
if rol:
return rol
else:
raise HTTPException(status_code=404, detail="Rol no encontrado")
except Exception as e:
print(e)
return []
# clase rol para el metodo post
class Rol(BaseModel):
rol: str
# metodo post de roles con manejo de errores
@app.post("/roles")
def create_rol(rol: Rol):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("INSERT INTO Roles (rol) VALUES (?)", [rol.rol])
conn.commit()
return {"mensaje": "Rol creado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear el rol"}
# clase rol para el metodo put
class RolPut(BaseModel):
id_rol: int
rol: str
# metodo put de roles con manejo de errores
@app.put("/roles")
def update_rol(rol: RolPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Roles SET rol = ? WHERE id_rol = ?", [rol.rol, rol.id_rol]
)
conn.commit()
return {"mensaje": "Rol actualizado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar el rol"}
# metodo delete de roles con manejo de errores
@app.delete("/roles/{id_rol}")
def delete_rol(id_rol: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM Roles WHERE id_rol = ?", [id_rol])
conn.commit()
return {"mensaje": "Rol eliminado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar el rol"}
"""
CREATE TABLE Usuario (
id_usuario INTEGER PRIMARY KEY AUTOINCREMENT,
nombre_usuario VARCHAR,
dni VARCHAR,
clave VARCHAR,
id_rol INTEGER REFERENCES Roles(id_rol),
estado VARCHAR
);
"""
# metodo get de usuarios
@app.get("/usuarios")
def get_usuarios():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Usuario")
usuarios = cursor.fetchall()
return usuarios
except Exception as e:
print(e)
return []
# get de usuarios por id
@app.get("/usuarios/{id_usuario}")
def get_usuario(id_usuario: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Usuario WHERE id_usuario = ?", [id_usuario])
usuario = cursor.fetchone()
if usuario:
return usuario
else:
raise HTTPException(status_code=404, detail="Usuario no encontrado")
except Exception as e:
print(e)
return []
# busqueda por nombre de usuario con like con manejo de errores y con metodo get
@app.get("/usuarios/busqueda/{nombre_usuario}")
def get_usuario_busqueda(nombre_usuario: str):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Usuario WHERE nombre_usuario LIKE ?",
["%" + nombre_usuario + "%"],
)
usuario = cursor.fetchall()
if usuario:
return usuario
else:
raise HTTPException(status_code=404, detail="Usuario no encontrado")
except Exception as e:
print(e)
return []
# busqueda por nombre de usuario con like con manejo de errores y con metodo post y clase
class UsuarioBusqueda(BaseModel):
nombre_usuario: str
# metodo post
@app.post("/usuarios/busqueda")
def get_usuario_busqueda(usuario: UsuarioBusqueda):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Usuario WHERE nombre_usuario LIKE ?",
["%" + usuario.nombre_usuario + "%"],
)
usuario = cursor.fetchall()
if usuario:
return usuario
else:
raise HTTPException(status_code=404, detail="Usuario no encontrado")
except Exception as e:
print(e)
return []
# clase usuario para el metodo post
class Usuario(BaseModel):
nombre_usuario: str
dni: str
clave: str
id_rol: int
estado: str
# metodo post de usuarios con manejo de errores
@app.post("/usuarios")
def create_usuario(usuario: Usuario):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO Usuario (nombre_usuario, dni, clave, id_rol, estado) VALUES (?, ?, ?, ?, ?)",
[
usuario.nombre_usuario,
usuario.dni,
usuario.clave,
usuario.id_rol,
usuario.estado,
],
)
conn.commit()
return {"mensaje": "Usuario creado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear el usuario"}
# clase usuario para el metodo put
class UsuarioPut(BaseModel):
id_usuario: int
nombre_usuario: str
dni: str
clave: str
id_rol: int
estado: str
# metodo put de usuarios con manejo de errores
@app.put("/usuarios")
def update_usuario(usuario: UsuarioPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Usuario SET nombre_usuario = ?, dni = ?, clave = ?, id_rol = ?, estado = ? WHERE id_usuario = ?",
[
usuario.nombre_usuario,
usuario.dni,
usuario.clave,
usuario.id_rol,
usuario.estado,
usuario.id_usuario,
],
)
conn.commit()
return {"mensaje": "Usuario actualizado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar el usuario"}
# metodo delete de usuarios con manejo de errores
@app.delete("/usuarios/{id_usuario}")
def delete_usuario(id_usuario: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM Usuario WHERE id_usuario = ?", [id_usuario])
conn.commit()
return {"mensaje": "Usuario eliminado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar el usuario"}
"""
CREATE TABLE Cliente (
id_cliente INTEGER PRIMARY KEY AUTOINCREMENT,
nombres_y_apellidos VARCHAR,
edad INTEGER,
telefono INTEGER,
direccion VARCHAR
);
"""
# metodo get de clientes
@app.get("/clientes")
def get_clientes():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Cliente")
clientes = cursor.fetchall()
return clientes
except Exception as e:
print(e)
return []
# get de clientes por id
@app.get("/clientes/{id_cliente}")
def get_cliente(id_cliente: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Cliente WHERE id_cliente = ?", [id_cliente])
cliente = cursor.fetchone()
if cliente:
return cliente
else:
raise HTTPException(status_code=404, detail="Cliente no encontrado")
except Exception as e:
print(e)
return []
# busqueda por nombre y apellido de cliente con like con manejo de errores y con metodo get
@app.get("/clientes/busqueda/{nombres_y_apellidos}")
def get_cliente_busqueda(nombres_y_apellidos: str):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Cliente WHERE nombres_y_apellidos LIKE ?",
["%" + nombres_y_apellidos + "%"],
)
cliente = cursor.fetchall()
if cliente:
return cliente
else:
raise HTTPException(status_code=404, detail="Cliente no encontrado")
except Exception as e:
print(e)
return []
# busqueda por nombre y apellido de cliente con like con manejo de errores y con metodo post y clase
class ClienteBusqueda(BaseModel):
nombres_y_apellidos: str
# metodo post
@app.post("/clientes/busqueda")
def get_cliente_busqueda(cliente: ClienteBusqueda):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Cliente WHERE nombres_y_apellidos LIKE ?",
["%" + cliente.nombres_y_apellidos + "%"],
)
cliente = cursor.fetchall()
if cliente:
return cliente
else:
raise HTTPException(status_code=404, detail="Cliente no encontrado")
except Exception as e:
print(e)
return []
# clase cliente para el metodo post
class Cliente(BaseModel):
nombres_y_apellidos: str
edad: int
telefono: int
direccion: str
# metodo post de clientes con manejo de errores
@app.post("/clientes")
def create_cliente(cliente: Cliente):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO Cliente (nombres_y_apellidos, edad, telefono, direccion) VALUES (?, ?, ?, ?)",
[
cliente.nombres_y_apellidos,
cliente.edad,
cliente.telefono,
cliente.direccion,
],
)
conn.commit()
return {"mensaje": "Cliente creado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear el cliente"}
# clase cliente para el metodo put
class ClientePut(BaseModel):
id_cliente: int
nombres_y_apellidos: str
edad: int
telefono: int
direccion: str
# metodo put de clientes con manejo de errores
@app.put("/clientes")
def update_cliente(cliente: ClientePut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Cliente SET nombres_y_apellidos = ?, edad = ?, telefono = ?, direccion = ? WHERE id_cliente = ?",
[
cliente.nombres_y_apellidos,
cliente.edad,
cliente.telefono,
cliente.direccion,
cliente.id_cliente,
],
)
conn.commit()
return {"mensaje": "Cliente actualizado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar el cliente"}
# metodo delete de clientes con manejo de errores
@app.delete("/clientes/{id_cliente}")
def delete_cliente(id_cliente: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM Cliente WHERE id_cliente = ?", [id_cliente])
conn.commit()
return {"mensaje": "Cliente eliminado correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar el cliente"}
"""
CREATE TABLE Medidas (
id_medidas INTEGER PRIMARY KEY AUTOINCREMENT,
Esfera_OD_lejos FLOAT,
Cilindro_OD_lejos FLOAT,
Eje_OD_lejos FLOAT,
Agudeza_visual_OD_lejos FLOAT,
Esfera_OI_lejos FLOAT,
Cilindro_OI_lejos FLOAT,
Eje_OI_lejos FLOAT,
Agudeza_visual_OI_lejos FLOAT,
Esfera_OD_cerca FLOAT,
Cilindro_OD_cerca FLOAT,
Eje_OD_cerca FLOAT,
Agudeza_visual_OD_cerca FLOAT,
Esfera_OI_cerca FLOAT,
Cilindro_OI_cerca FLOAT,
Eje_OI_cerca FLOAT,
Agudeza_visual_OI_cerca FLOAT,
id_cliente INTEGER REFERENCES Cliente(id_cliente)
);
"""
# metodo get de medidas
@app.get("/medidas")
def get_medidas():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Medidas")
medidas = cursor.fetchall()
return medidas
except Exception as e:
print(e)
return []
# get de medidas por id
@app.get("/medidas/{id_medidas}")
def get_medida(id_medidas: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Medidas WHERE id_medidas = ?", [id_medidas])
medida = cursor.fetchone()
if medida:
return medida
else:
raise HTTPException(status_code=404, detail="Medida no encontrada")
except Exception as e:
print(e)
return []
# metodo de busqueda que sea por id_cliente y retorne las medidas asociadas a ese cliente con clase y metodo post
class MedidaBusqueda(BaseModel):
id_cliente: int
@app.post("/medidas/busqueda")
def post_medida_busqueda(medida: MedidaBusqueda):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Medidas WHERE id_cliente = ?", [medida.id_cliente]
)
medida = cursor.fetchall()
if medida:
return medida
else:
raise HTTPException(status_code=404, detail="Medida no encontrada")
except Exception as e:
print(e)
return []
# clase medida para el metodo post
class Medida(BaseModel):
Esfera_OD_lejos: float
Cilindro_OD_lejos: float
Eje_OD_lejos: float
Agudeza_visual_OD_lejos: float
Esfera_OI_lejos: float
Cilindro_OI_lejos: float
Eje_OI_lejos: float
Agudeza_visual_OI_lejos: float
Esfera_OD_cerca: float
Cilindro_OD_cerca: float
Eje_OD_cerca: float
Agudeza_visual_OD_cerca: float
Esfera_OI_cerca: float
Cilindro_OI_cerca: float
Eje_OI_cerca: float
Agudeza_visual_OI_cerca: float
id_cliente: int
# metodo post de medidas con manejo de errores
@app.post("/medidas")
def create_medida(medida: Medida):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO Medidas (Esfera_OD_lejos, Cilindro_OD_lejos, Eje_OD_lejos, Agudeza_visual_OD_lejos, Esfera_OI_lejos, Cilindro_OI_lejos, Eje_OI_lejos, Agudeza_visual_OI_lejos, Esfera_OD_cerca, Cilindro_OD_cerca, Eje_OD_cerca, Agudeza_visual_OD_cerca, Esfera_OI_cerca, Cilindro_OI_cerca, Eje_OI_cerca, Agudeza_visual_OI_cerca, id_cliente) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
[
medida.Esfera_OD_lejos,
medida.Cilindro_OD_lejos,
medida.Eje_OD_lejos,
medida.Agudeza_visual_OD_lejos,
medida.Esfera_OI_lejos,
medida.Cilindro_OI_lejos,
medida.Eje_OI_lejos,
medida.Agudeza_visual_OI_lejos,
medida.Esfera_OD_cerca,
medida.Cilindro_OD_cerca,
medida.Eje_OD_cerca,
medida.Agudeza_visual_OD_cerca,
medida.Esfera_OI_cerca,
medida.Cilindro_OI_cerca,
medida.Eje_OI_cerca,
medida.Agudeza_visual_OI_cerca,
medida.id_cliente,
],
)
conn.commit()
return {"mensaje": "Medida creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la medida"}
# clase medida para el metodo put
class MedidaPut(BaseModel):
id_medidas: int
Esfera_OD_lejos: float
Cilindro_OD_lejos: float
Eje_OD_lejos: float
Agudeza_visual_OD_lejos: float
Esfera_OI_lejos: float
Cilindro_OI_lejos: float
Eje_OI_lejos: float
Agudeza_visual_OI_lejos: float
Esfera_OD_cerca: float
Cilindro_OD_cerca: float
Eje_OD_cerca: float
Agudeza_visual_OD_cerca: float
Esfera_OI_cerca: float
Cilindro_OI_cerca: float
Eje_OI_cerca: float
Agudeza_visual_OI_cerca: float
id_cliente: int
# metodo put de medidas con manejo de errores
@app.put("/medidas")
def update_medida(medida: MedidaPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Medidas SET Esfera_OD_lejos = ?, Cilindro_OD_lejos = ?, Eje_OD_lejos = ?, Agudeza_visual_OD_lejos = ?, Esfera_OI_lejos = ?, Cilindro_OI_lejos = ?, Eje_OI_lejos = ?, Agudeza_visual_OI_lejos = ?, Esfera_OD_cerca = ?, Cilindro_OD_cerca = ?, Eje_OD_cerca = ?, Agudeza_visual_OD_cerca = ?, Esfera_OI_cerca = ?, Cilindro_OI_cerca = ?, Eje_OI_cerca = ?, Agudeza_visual_OI_cerca = ?, id_cliente = ? WHERE id_medidas = ?",
[
medida.Esfera_OD_lejos,
medida.Cilindro_OD_lejos,
medida.Eje_OD_lejos,
medida.Agudeza_visual_OD_lejos,
medida.Esfera_OI_lejos,
medida.Cilindro_OI_lejos,
medida.Eje_OI_lejos,
medida.Agudeza_visual_OI_lejos,
medida.Esfera_OD_cerca,
medida.Cilindro_OD_cerca,
medida.Eje_OD_cerca,
medida.Agudeza_visual_OD_cerca,
medida.Esfera_OI_cerca,
medida.Cilindro_OI_cerca,
medida.Eje_OI_cerca,
medida.Agudeza_visual_OI_cerca,
medida.id_cliente,
medida.id_medidas,
],
)
conn.commit()
return {"mensaje": "Medida actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la medida"}
# metodo delete de medidas con manejo de errores
@app.delete("/medidas/{id_medidas}")
def delete_medida(id_medidas: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM Medidas WHERE id_medidas = ?", [id_medidas])
conn.commit()
return {"mensaje": "Medida eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la medida"}
"""
CREATE TABLE Prescripcion (
id_prescripcion INTEGER PRIMARY KEY AUTOINCREMENT,
id_medidas INTEGER REFERENCES Medidas(id_medidas),
detalle_lunas VARCHAR,
fecha VARCHAR
);
"""
# metodo get de prescripciones
@app.get("/prescripciones")
def get_prescripciones():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Prescripcion")
prescripciones = cursor.fetchall()
return prescripciones
except Exception as e:
print(e)
return []
# get de prescripciones por id
@app.get("/prescripciones/{id_prescripcion}")
def get_prescripcion(id_prescripcion: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Prescripcion WHERE id_prescripcion = ?", [id_prescripcion]
)
prescripcion = cursor.fetchone()
if prescripcion:
return prescripcion
else:
raise HTTPException(status_code=404, detail="Prescripcion no encontrada")
except Exception as e:
print(e)
return []
"""
CREATE TABLE Prescripcion (
id_prescripcion INTEGER PRIMARY KEY AUTOINCREMENT,
id_medidas INTEGER REFERENCES Medidas(id_medidas),
detalle_lunas VARCHAR,
fecha VARCHAR
);
"""
# clase prescripcion para el metodo post
class Prescripcion(BaseModel):
id_medidas: int
detalle_lunas: str
fecha: str
# metodo post de prescripciones con manejo de errores
@app.post("/prescripciones")
def create_prescripcion(prescripcion: Prescripcion):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO Prescripcion (id_medidas, detalle_lunas, fecha) VALUES (?, ?, ?, ?)",
[
prescripcion.id_medidas,
prescripcion.detalle_lunas,
prescripcion.fecha,
],
)
conn.commit()
return {"mensaje": "Prescripcion creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la prescripcion"}
# clase prescripcion para el metodo put
class PrescripcionPut(BaseModel):
id_prescripcion: int
id_medidas: int
detalle_lunas: str
fecha: str
# metodo put de prescripciones con manejo de errores
@app.put("/prescripciones")
def update_prescripcion(prescripcion: PrescripcionPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Prescripcion SET id_medidas = ?, detalle_lunas = ?, fecha = ? WHERE id_prescripcion = ?",
[
prescripcion.id_medidas,
prescripcion.detalle_lunas,
prescripcion.fecha,
prescripcion.id_prescripcion,
],
)
conn.commit()
return {"mensaje": "Prescripcion actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la prescripcion"}
# metodo delete de prescripciones con manejo de errores
@app.delete("/prescripciones/{id_prescripcion}")
def delete_prescripcion(id_prescripcion: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"DELETE FROM Prescripcion WHERE id_prescripcion = ?", [id_prescripcion]
)
conn.commit()
return {"mensaje": "Prescripcion eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la prescripcion"}
"""
CREATE TABLE Gama (
id_gama INTEGER PRIMARY KEY AUTOINCREMENT,
gama VARCHAR
);
"""
# metodo get de gamas
@app.get("/gamas")
def get_gamas():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Gama")
gamas = cursor.fetchall()
return gamas
except Exception as e:
print(e)
return []
# get de gamas por id
@app.get("/gamas/{id_gama}")
def get_gama(id_gama: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Gama WHERE id_gama = ?", [id_gama])
gama = cursor.fetchone()
if gama:
return gama
else:
raise HTTPException(status_code=404, detail="Gama no encontrada")
except Exception as e:
print(e)
return []
# busqueda por nombre de gama con like con manejo de errores y con metodo get
@app.get("/gamas/busqueda/{gama}")
def get_gama_busqueda(gama: str):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Gama WHERE gama LIKE ?", ["%" + gama + "%"])
gama = cursor.fetchall()
if gama:
return gama
else:
raise HTTPException(status_code=404, detail="Gama no encontrada")
except Exception as e:
print(e)
return []
# busqueda por nombre de gama con like con manejo de errores y con metodo post y clase
class GamaBusqueda(BaseModel):
gama: str
# metodo post
@app.post("/gamas/busqueda")
def get_gama_busqueda(gama: GamaBusqueda):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Gama WHERE gama LIKE ?", ["%" + gama.gama + "%"])
gama = cursor.fetchall()
if gama:
return gama
else:
raise HTTPException(status_code=404, detail="Gama no encontrada")
except Exception as e:
print(e)
return []
# clase gama para el metodo post
class Gama(BaseModel):
gama: str
# metodo post de gamas con manejo de errores
@app.post("/gamas")
def create_gama(gama: Gama):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO Gama (gama) VALUES (?)",
[
gama.gama,
],
)
conn.commit()
return {"mensaje": "Gama creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la gama"}
# clase gama para el metodo put
class GamaPut(BaseModel):
id_gama: int
gama: str
# metodo put de gamas con manejo de errores
@app.put("/gamas")
def update_gama(gama: GamaPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Gama SET gama = ? WHERE id_gama = ?",
[
gama.gama,
gama.id_gama,
],
)
conn.commit()
return {"mensaje": "Gama actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la gama"}
# metodo delete de gamas con manejo de errores
@app.delete("/gamas/{id_gama}")
def delete_gama(id_gama: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM Gama WHERE id_gama = ?", [id_gama])
conn.commit()
return {"mensaje": "Gama eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la gama"}
"""
CREATE TABLE Montura (
id_montura INTEGER PRIMARY KEY AUTOINCREMENT,
nombre_montura VARCHAR,
id_gama INTEGER REFERENCES Gama(id_gama),
imagen VARCHAR
);
"""
# metodo get de monturas
@app.get("/monturas")
def get_monturas():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Montura")
monturas = cursor.fetchall()
return monturas
except Exception as e:
print(e)
return []
# get de monturas por id
@app.get("/monturas/{id_montura}")
def get_montura(id_montura: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Montura WHERE id_montura = ?", [id_montura])
montura = cursor.fetchone()
if montura:
return montura
else:
raise HTTPException(status_code=404, detail="Montura no encontrada")
except Exception as e:
print(e)
return []
# busqueda por nombre de montura con like con manejo de errores y con metodo get
@app.get("/monturas/busqueda/{nombre_montura}")
def get_montura_busqueda(nombre_montura: str):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Montura WHERE nombre_montura LIKE ?",
["%" + nombre_montura + "%"],
)
montura = cursor.fetchall()
if montura:
return montura
else:
raise HTTPException(status_code=404, detail="Montura no encontrada")
except Exception as e:
print(e)
return []
# busqueda por nombre de montura con like con manejo de errores y con metodo post y clase
class MonturaBusqueda(BaseModel):
nombre_montura: str
# metodo post
@app.post("/monturas/busqueda")
def get_montura_busqueda(montura: MonturaBusqueda):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Montura WHERE nombre_montura LIKE ?",
["%" + montura.nombre_montura + "%"],
)
montura = cursor.fetchall()
if montura:
return montura
else:
raise HTTPException(status_code=404, detail="Montura no encontrada")
except Exception as e:
print(e)
return []
# clase montura para el metodo post
class Montura(BaseModel):
nombre_montura: str
id_gama: int
imagen: str
# metodo post de monturas con manejo de errores
@app.post("/monturas")
def create_montura(montura: Montura):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO Montura (nombre_montura, id_gama, imagen) VALUES (?, ?, ?)",
[
montura.nombre_montura,
montura.id_gama,
montura.imagen,
],
)
conn.commit()
return {"mensaje": "Montura creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la montura"}
# clase montura para el metodo put
class MonturaPut(BaseModel):
id_montura: int
nombre_montura: str
id_gama: int
imagen: str
# metodo put de monturas con manejo de errores
@app.put("/monturas")
def update_montura(montura: MonturaPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Montura SET nombre_montura = ?, id_gama = ?, imagen = ? WHERE id_montura = ?",
[
montura.nombre_montura,
montura.id_gama,
montura.imagen,
montura.id_montura,
],
)
conn.commit()
return {"mensaje": "Montura actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la montura"}
# metodo delete de monturas con manejo de errores
@app.delete("/monturas/{id_montura}")
def delete_montura(id_montura: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM Montura WHERE id_montura = ?", [id_montura])
conn.commit()
return {"mensaje": "Montura eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la montura"}
"""
CREATE TABLE Montura_inventario (
id_montura_inventario INTEGER PRIMARY KEY AUTOINCREMENT,
id_montura INTEGER REFERENCES Montura(id_montura),
precio_unit FLOAT,
stock INTEGER,
codigo VARCHAR
);
"""
# metodo get de monturas_inventario
@app.get("/monturas_inventario")
def get_monturas_inventario():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM Montura_inventario")
monturas_inventario = cursor.fetchall()
return monturas_inventario
except Exception as e:
print(e)
return []
# get de monturas_inventario por id
@app.get("/monturas_inventario/{id_montura_inventario}")
def get_montura_inventario(id_montura_inventario: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Montura_inventario WHERE id_montura_inventario = ?",
[id_montura_inventario],
)
montura_inventario = cursor.fetchone()
if montura_inventario:
return montura_inventario
else:
raise HTTPException(
status_code=404, detail="Montura_inventario no encontrada"
)
except Exception as e:
print(e)
return []
# busqueda por codigo de montura con like con manejo de errores y con metodo get
@app.get("/monturas_inventario/busqueda/{codigo}")
def get_montura_inventario_busqueda(codigo: str):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Montura_inventario WHERE codigo LIKE ?",
["%" + codigo + "%"],
)
montura_inventario = cursor.fetchall()
if montura_inventario:
return montura_inventario
else:
raise HTTPException(
status_code=404, detail="Montura_inventario no encontrada"
)
except Exception as e:
print(e)
return []
# busqueda por codigo de montura con like con manejo de errores y con metodo post y clase
class Montura_inventarioBusqueda(BaseModel):
codigo: str
# metodo post
@app.post("/monturas_inventario/busqueda")
def get_montura_inventario_busqueda(montura_inventario: Montura_inventarioBusqueda):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM Montura_inventario WHERE codigo LIKE ?",
["%" + montura_inventario.codigo + "%"],
)
montura_inventario = cursor.fetchall()
if montura_inventario:
return montura_inventario
else:
raise HTTPException(
status_code=404, detail="Montura_inventario no encontrada"
)
except Exception as e:
print(e)
return []
# clase montura_inventario para el metodo post
class Montura_inventario(BaseModel):
id_montura: int
precio_unit: float
stock: int
codigo: str
# metodo post de monturas_inventario con manejo de errores
@app.post("/monturas_inventario")
def create_montura_inventario(montura_inventario: Montura_inventario):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO Montura_inventario (id_montura, precio_unit, stock, codigo) VALUES (?, ?, ?, ?)",
[
montura_inventario.id_montura,
montura_inventario.precio_unit,
montura_inventario.stock,
montura_inventario.codigo,
],
)
conn.commit()
return {"mensaje": "Montura_inventario creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la montura_inventario"}
# clase montura_inventario para el metodo put
class Montura_inventarioPut(BaseModel):
id_montura_inventario: int
id_montura: int
precio_unit: float
stock: int
codigo: str
# metodo put de monturas_inventario con manejo de errores
@app.put("/monturas_inventario")
def update_montura_inventario(montura_inventario: Montura_inventarioPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE Montura_inventario SET id_montura = ?, precio_unit = ?, stock = ?, codigo = ? WHERE id_montura_inventario = ?",
[
montura_inventario.id_montura,
montura_inventario.precio_unit,
montura_inventario.stock,
montura_inventario.codigo,
montura_inventario.id_montura_inventario,
],
)
conn.commit()
return {"mensaje": "Montura_inventario actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la montura_inventario"}
# metodo delete de monturas_inventario con manejo de errores
@app.delete("/monturas_inventario/{id_montura_inventario}")
def delete_montura_inventario(id_montura_inventario: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"DELETE FROM Montura_inventario WHERE id_montura_inventario = ?",
[id_montura_inventario],
)
conn.commit()
return {"mensaje": "Montura_inventario eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la montura_inventario"}
"""
CREATE TABLE boleta (
id_boleta INTEGER PRIMARY KEY AUTOINCREMENT,
precio_total FLOAT,
estado_recojo VARCHAR
);
"""
# metodo get de boletas
@app.get("/boletas")
def get_boletas():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM boleta")
boletas = cursor.fetchall()
return boletas
except Exception as e:
print(e)
return []
# get de boletas por id
@app.get("/boletas/{id_boleta}")
def get_boleta(id_boleta: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM boleta WHERE id_boleta = ?", [id_boleta])
boleta = cursor.fetchone()
if boleta:
return boleta
else:
raise HTTPException(status_code=404, detail="Boleta no encontrada")
except Exception as e:
print(e)
return []
# clase boleta para el metodo post
class Boleta(BaseModel):
precio_total: float
estado_recojo: str
# metodo post de boletas con manejo de errores
@app.post("/boletas")
def create_boleta(boleta: Boleta):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO boleta (precio_total, estado_recojo) VALUES (?, ?)",
[
boleta.precio_total,
boleta.estado_recojo,
],
)
conn.commit()
return {"mensaje": "Boleta creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la boleta"}
# clase boleta para el metodo put
class BoletaPut(BaseModel):
id_boleta: int
precio_total: float
estado_recojo: str
# metodo put de boletas con manejo de errores
@app.put("/boletas")
def update_boleta(boleta: BoletaPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE boleta SET precio_total = ?, estado_recojo = ? WHERE id_boleta = ?",
[
boleta.precio_total,
boleta.estado_recojo,
boleta.id_boleta,
],
)
conn.commit()
return {"mensaje": "Boleta actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la boleta"}
# metodo delete de boletas con manejo de errores
@app.delete("/boletas/{id_boleta}")
def delete_boleta(id_boleta: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM boleta WHERE id_boleta = ?", [id_boleta])
conn.commit()
return {"mensaje": "Boleta eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la boleta"}
"""
CREATE TABLE montura_pedido (
id_montura_pedido INTEGER PRIMARY KEY AUTOINCREMENT,
id_montura_inventario INTEGER REFERENCES Montura_inventario(id_montura_inventario),
cantidad INTEGER,
precio FLOAT,
id_boleta INTEGER REFERENCES boleta(id_boleta)
);
"""
# metodo get de monturas_pedido
@app.get("/monturas_pedido")
def get_monturas_pedido():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM montura_pedido")
monturas_pedido = cursor.fetchall()
return monturas_pedido
except Exception as e:
print(e)
return []
# get de monturas_pedido por id
@app.get("/monturas_pedido/{id_montura_pedido}")
def get_montura_pedido(id_montura_pedido: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM montura_pedido WHERE id_montura_pedido = ?",
[id_montura_pedido],
)
montura_pedido = cursor.fetchone()
if montura_pedido:
return montura_pedido
else:
raise HTTPException(status_code=404, detail="Montura_pedido no encontrada")
except Exception as e:
print(e)
return []
# clase montura_pedido para el metodo post
class Montura_pedido(BaseModel):
id_montura_inventario: int
cantidad: int
precio: float
id_boleta: int
# metodo post de monturas_pedido con manejo de errores
@app.post("/monturas_pedido")
def create_montura_pedido(montura_pedido: Montura_pedido):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO montura_pedido (id_montura_inventario, cantidad, precio, id_boleta) VALUES (?, ?, ?, ?)",
[
montura_pedido.id_montura_inventario,
montura_pedido.cantidad,
montura_pedido.precio,
montura_pedido.id_boleta,
],
)
conn.commit()
return {"mensaje": "Montura_pedido creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la montura_pedido"}
# clase montura_pedido para el metodo put
class Montura_pedidoPut(BaseModel):
id_montura_pedido: int
id_montura_inventario: int
cantidad: int
precio: float
id_boleta: int
# metodo put de monturas_pedido con manejo de errores
@app.put("/monturas_pedido")
def update_montura_pedido(montura_pedido: Montura_pedidoPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE montura_pedido SET id_montura_inventario = ?, cantidad = ?, precio = ?, id_boleta = ? WHERE id_montura_pedido = ?",
[
montura_pedido.id_montura_inventario,
montura_pedido.cantidad,
montura_pedido.precio,
montura_pedido.id_boleta,
montura_pedido.id_montura_pedido,
],
)
conn.commit()
return {"mensaje": "Montura_pedido actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la montura_pedido"}
# metodo delete de monturas_pedido con manejo de errores
@app.delete("/monturas_pedido/{id_montura_pedido}")
def delete_montura_pedido(id_montura_pedido: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"DELETE FROM montura_pedido WHERE id_montura_pedido = ?",
[id_montura_pedido],
)
conn.commit()
return {"mensaje": "Montura_pedido eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la montura_pedido"}
"""
CREATE TABLE lunas_pedido (
id_lunas_pedido INTEGER PRIMARY KEY AUTOINCREMENT,
id_prescripcion INTEGER REFERENCES Prescripcion(id_prescripcion),
precio FLOAT,
id_boleta INTEGER REFERENCES boleta(id_boleta),
descripcion VARCHAR
);
"""
# metodo get de lunas_pedido
@app.get("/lunas_pedido")
def get_lunas_pedido():
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM lunas_pedido")
lunas_pedido = cursor.fetchall()
return lunas_pedido
except Exception as e:
print(e)
return []
# get de lunas_pedido por id
@app.get("/lunas_pedido/{id_lunas_pedido}")
def get_lunas_pedido(id_lunas_pedido: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM lunas_pedido WHERE id_lunas_pedido = ?", [id_lunas_pedido]
)
luna_pedido = cursor.fetchone()
if luna_pedido:
return luna_pedido
else:
raise HTTPException(status_code=404, detail="Lunas_pedido no encontrada")
except Exception as e:
print(e)
return []
# clase lunas_pedido para el metodo post
class Lunas_pedido(BaseModel):
id_prescripcion: int
precio: float
id_boleta: int
descripcion: str
# metodo post de lunas_pedido con manejo de errores
@app.post("/lunas_pedido")
def create_lunas_pedido(lunas_pedido: Lunas_pedido):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO lunas_pedido (id_prescripcion, precio, id_boleta, descripcion) VALUES (?, ?, ?, ?)",
[
lunas_pedido.id_prescripcion,
lunas_pedido.precio,
lunas_pedido.id_boleta,
lunas_pedido.descripcion,
],
)
conn.commit()
return {"mensaje": "Lunas_pedido creada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al crear la lunas_pedido"}
# clase lunas_pedido para el metodo put
class Lunas_pedidoPut(BaseModel):
id_lunas_pedido: int
id_prescripcion: int
precio: float
id_boleta: int
descripcion: str
# metodo put de lunas_pedido con manejo de errores
@app.put("/lunas_pedido")
def update_lunas_pedido(lunas_pedido: Lunas_pedidoPut):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE lunas_pedido SET id_prescripcion = ?, precio = ?, id_boleta = ?, descripcion = ? WHERE id_lunas_pedido = ?",
[
lunas_pedido.id_prescripcion,
lunas_pedido.precio,
lunas_pedido.id_boleta,
lunas_pedido.descripcion,
lunas_pedido.id_lunas_pedido,
],
)
conn.commit()
return {"mensaje": "Lunas_pedido actualizada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al actualizar la lunas_pedido"}
# metodo delete de lunas_pedido con manejo de errores
@app.delete("/lunas_pedido/{id_lunas_pedido}")
def delete_lunas_pedido(id_lunas_pedido: int):
try:
conn = DatabaseConnection().get_connection()
cursor = conn.cursor()
cursor.execute(
"DELETE FROM lunas_pedido WHERE id_lunas_pedido = ?", [id_lunas_pedido]
)
conn.commit()
return {"mensaje": "Lunas_pedido eliminada correctamente"}
except Exception as e:
print(e)
return {"mensaje": "Error al eliminar la lunas_pedido"}