Spaces:
Build error
Build error
| # 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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"} | |