Spaces:
Running
Running
File size: 2,920 Bytes
a79e795 cc5ef58 a79e795 cc5ef58 a79e795 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
from library.librerias import *
from models.empleados import Empleados
router = APIRouter(
prefix="/empleados",
tags=["Empleados"],
responses={404: {"description": "No encontrado"}},
)
"""CREATE TABLE empleados (
ID_Empleado INTEGER PRIMARY KEY AUTOINCREMENT,
Nombre TEXT,
Apellido TEXT,
Sueldo REAL,
Fecha_Inicio DATE
);"""
@router.get("/")
def get_empleados():
try:
with DatabaseConnection().get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM empleados")
empleados = cursor.fetchall()
return empleados
except Exception as e:
print(e)
return []
@router.post("/")
def post_empleado(empleado: Empleados):
try:
with DatabaseConnection().get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO empleados (Nombre, Apellido, Sueldo, Fecha_Inicio) VALUES (?, ?, ?, ?)",
(
empleado.Nombre,
empleado.Apellido,
empleado.Sueldo,
empleado.Fecha_Inicio,
),
)
conn.commit()
return {"message": "Empleado creado"}
except Exception as e:
print(e)
return []
@router.put("/")
def put_empleado(empleado: Empleados):
try:
with DatabaseConnection().get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"UPDATE empleados SET Nombre = ?, Apellido = ?, Sueldo = ?, Fecha_Inicio = ? WHERE ID_Empleado = ?",
(
empleado.Nombre,
empleado.Apellido,
empleado.Sueldo,
empleado.Fecha_Inicio,
empleado.ID_Empleado,
),
)
conn.commit()
return {"message": "Empleado actualizado"}
except Exception as e:
print(e)
return []
# search by id_empleado in empleados table
@router.get("/{ID_Empleado}")
def get_empleado_by_id(ID_Empleado: int):
try:
with DatabaseConnection().get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"SELECT * FROM empleados WHERE ID_Empleado = ?", (ID_Empleado,)
)
empleado = cursor.fetchall()
return empleado
except Exception as e:
print(e)
return []
# search by nombre in empleados table
@router.get("/nombre/{Nombre}")
def get_empleado_by_name(Nombre: str):
try:
with DatabaseConnection().get_connection() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM empleados WHERE Nombre = ?", (Nombre,))
empleado = cursor.fetchall()
return empleado
except Exception as e:
print(e)
return []
|