Descubre una guía paso a paso en la que aprenderás a crear herramientas útiles en Python. Desde la monitorización y gestión de procesos del sistema, pasando por la implementación de hilos para gestionar usuarios y sesiones, hasta la simulación de pedidos en un entorno de comercio electrónico y la integración con APIs REST. ¡Una lectura imprescindible para potenciar tus habilidades de programación!
1. Control y Supervisión de Procesos en el Sistema
En este primer apartado veremos cómo crear una utilidad en Python que nos permita listar los procesos activos del sistema, identificar uno en particular (por ejemplo, el Bloc de notas) y, además, poder finalizar un proceso seleccionado por el usuario. Se hace uso de la librería psutil para interactuar con el sistema.
Paso a paso:
- Listado de Procesos
Recorremos los procesos activos e imprimimos su nombre y PID. Si se detecta el Bloc de notas (usualmente “Notepad.exe”), se mostrará un mensaje especial. - Finalización de un Proceso
Se solicita al usuario que ingrese el identificador (PID) del proceso que desea terminar. El programa intenta finalizarlo y, en caso de error (como falta de permisos), se informa adecuadamente. - (Opcional) Información Adicional
Se puede extender la funcionalidad para mostrar detalles como el uso de CPU y memoria.
A continuación, se muestra un ejemplo de código:
import psutil
import time
def mostrar_procesos():
print("\n--- Lista de Procesos Activos ---")
for proceso in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_info']):
try:
info = proceso.info
pid = info.get('pid')
nombre = info.get('name')
# Mensaje especial para Notepad
if nombre and "Notepad.exe" in nombre:
print(f"{nombre} (PID: {pid}) --> ¡El Bloc de notas está en ejecución!")
else:
print(f"{nombre} (PID: {pid})")
except psutil.NoSuchProcess:
continue
def terminar_proceso():
try:
pid_input = int(input("\nIngrese el PID del proceso que desea finalizar (o 0 para salir): "))
if pid_input == 0:
return False
proceso = psutil.Process(pid_input)
proceso.terminate()
proceso.wait(timeout=3)
print(f"Proceso con PID {pid_input} finalizado correctamente.")
except psutil.NoSuchProcess:
print("Error: No se encontró el proceso.")
except Exception as error:
print(f"Error al intentar finalizar el proceso: {error}")
return True
def main():
while True:
mostrar_procesos()
if not terminar_proceso():
print("Saliendo del programa de gestión de procesos...")
break
time.sleep(1) # Pausa breve para actualizar la lista
if __name__ == "__main__":
main()
Nota: Asegúrate de ejecutar este script con los permisos necesarios, ya que terminar ciertos procesos podría requerir privilegios elevados.
2. Gestión Concurrente de Datos de Usuarios con Hilos
Esta sección aborda cómo trabajar con hilos (threads) para procesar la información de varios usuarios. Usaremos tanto args como kwargs para pasar parámetros a cada hilo, permitiendo una gestión personalizada de la información.
Detalles de la implementación:
- Función de Procesamiento
Se define una función que recibe un identificador de usuario mediante args y detalles como nombre y edad mediante kwargs. La función muestra estos datos. - Creación y Lanzamiento de Hilos
Se crea una lista de usuarios y, para cada uno, se inicia un hilo que ejecuta la función de procesamiento. Se incluye una pausa aleatoria para simular el tiempo de procesamiento y observar la concurrencia. - Sincronización
Se espera a que todos los hilos finalicen antes de continuar.
Ejemplo de código:
import threading
import time
import random
def procesar_usuario(id_usuario, **datos):
# Simulación de tiempo de procesamiento aleatorio
time.sleep(random.uniform(0.5, 2))
nombre = datos.get('nombre', 'Desconocido')
edad = datos.get('edad', 'N/A')
print(f"Usuario ID: {id_usuario}, Nombre: {nombre}, Edad: {edad}")
def main():
# Lista de usuarios: (ID, nombre, edad)
usuarios = [
(1, 'Ana', 30),
(2, 'Carlos', 22),
(3, 'Beatriz', 27),
(4, 'David', 35),
(5, 'Elena', 29)
]
hilos = []
for usuario in usuarios:
id_usuario, nombre, edad = usuario
hilo = threading.Thread(target=procesar_usuario, args=(id_usuario,), kwargs={'nombre': nombre, 'edad': edad})
hilos.append(hilo)
hilo.start()
# Esperar a que todos los hilos finalicen
for hilo in hilos:
hilo.join()
print("Todos los usuarios han sido procesados.")
if __name__ == "__main__":
main()
3. Manejo de Sesiones de Usuario en Entornos Multihilo
En este ejemplo, simulamos la gestión de sesiones en un entorno donde múltiples hilos trabajan de forma simultánea. Se utiliza threading.local() para garantizar que cada hilo almacene su propia información de sesión, evitando interferencias.
¿Cómo lo implementamos?
- Definición de la Clase de Sesión
Creamos una clase que tenga métodos para iniciar y mostrar la sesión del usuario. - Uso de
threading.local()
Utilizamos un objeto local a los hilos para almacenar la instancia de sesión correspondiente a cada hilo. - Ejecución Concurrente
Se lanzan varios hilos, cada uno iniciando sesión con un nombre distinto, asegurando que la información sea exclusiva para cada uno.
Ejemplo de código:
import threading
import time
class SesionUsuario:
def iniciar_sesion(self, nombre_usuario):
self.nombre = nombre_usuario
def mostrar_sesion(self):
print(f"Sesión iniciada para el usuario: {self.nombre}")
# Objeto local para cada hilo
datos_sesion = threading.local()
def gestionar_sesion(nombre_usuario):
# Cada hilo crea su propia instancia de sesión
datos_sesion.sesion = SesionUsuario()
datos_sesion.sesion.iniciar_sesion(nombre_usuario)
# Simular algún tiempo de procesamiento
time.sleep(1)
datos_sesion.sesion.mostrar_sesion()
def main():
nombres = ['Ana', 'Carlos', 'Beatriz', 'David', 'Elena']
hilos = []
for nombre in nombres:
hilo = threading.Thread(target=gestionar_sesion, args=(nombre,))
hilos.append(hilo)
hilo.start()
for hilo in hilos:
hilo.join()
print("Todas las sesiones han sido gestionadas.")
if __name__ == "__main__":
main()
4. Simulación de Pedidos en un Comercio Electrónico con el Patrón Productor-Consumidor
Este proyecto simula el proceso en el que clientes generan pedidos (productores) y empleados los procesan (consumidores). La comunicación se gestiona mediante una cola compartida, garantizando que:
- Los clientes esperen si la cola está llena.
- Los empleados se bloqueen si la cola se encuentra vacía.
Puntos clave de la implementación:
- Configuración de la Cola
Se utilizaqueue.Queuecon una capacidad máxima definida para controlar el flujo de pedidos. - Funciones de Cliente y Empleado
- Clientes: Generan pedidos con un identificador único y los colocan en la cola, simulando demoras aleatorias.
- Empleados: Extraen pedidos de la cola y los procesan, también con tiempos de espera simulados.
- Control del Número Total de Pedidos
La simulación se detiene una vez que se han generado y procesado 15 pedidos en total.
Ejemplo de código:
import threading
import time
import random
from queue import Queue
# Cola compartida con capacidad máxima de 5 pedidos
cola_pedidos = Queue(maxsize=5)
# Variables globales para controlar la simulación
total_pedidos = 0
total_procesados = 0
max_pedidos = 15
lock = threading.Lock()
def cliente(id_cliente):
global total_pedidos
while True:
with lock:
if total_pedidos >= max_pedidos:
break
total_pedidos += 1
pedido = f"Pedido-{total_pedidos}"
# Esperar si la cola está llena
cola_pedidos.put(pedido)
print(f"Cliente {id_cliente}: Generó {pedido}")
time.sleep(random.uniform(1, 2))
def empleado(id_empleado):
global total_procesados
while True:
with lock:
if total_procesados >= max_pedidos:
break
try:
# Esperar a que haya un pedido en la cola
pedido = cola_pedidos.get(timeout=3)
print(f"Empleado {id_empleado}: Procesó {pedido}")
cola_pedidos.task_done()
time.sleep(random.uniform(2, 3))
with lock:
total_procesados += 1
except:
# Timeout o cola vacía
continue
def main():
# Crear 3 hilos de clientes
clientes = [threading.Thread(target=cliente, args=(i,)) for i in range(1, 4)]
# Crear 2 hilos de empleados
empleados = [threading.Thread(target=empleado, args=(i,)) for i in range(1, 3)]
for c in clientes:
c.start()
for e in empleados:
e.start()
for c in clientes:
c.join()
for e in empleados:
e.join()
print("Todos los pedidos han sido procesados.")
if __name__ == "__main__":
main()
Consejo: La sincronización mediante
lockes fundamental para evitar condiciones de carrera al actualizar las variables compartidas.
5. Conexión y Consumo de una API REST con Python
El último proyecto consiste en interactuar con una API pública para extraer y mostrar información de interés. En este ejemplo, se usará la API de OpenWeatherMap, pero puedes adaptar el código a cualquier otra API.
Pasos a seguir:
- Realizar la Solicitud HTTP
Utilizamos la libreríarequestspara enviar una petición GET a la API. - Procesar la Respuesta en Formato JSON
Se extraen datos específicos, como la temperatura, la descripción del clima y el nombre de la ciudad. - Manejo de Errores
Se verifica el código de estado y la existencia de los datos antes de mostrarlos, para asegurar la robustez del programa.
Ejemplo de código:
import requests
def obtener_clima(ciudad, api_key):
url = f"http://api.openweathermap.org/data/2.5/weather?q={ciudad}&appid={api_key}&units=metric&lang=es"
try:
respuesta = requests.get(url)
if respuesta.status_code == 200:
datos = respuesta.json()
# Verificar que las claves necesarias estén en la respuesta
if "main" in datos and "weather" in datos:
temperatura = datos["main"].get("temp")
descripcion = datos["weather"][0].get("description")
print(f"Ciudad: {ciudad}")
print(f"Temperatura: {temperatura}°C")
print(f"Clima: {descripcion}")
else:
print("La respuesta no contiene la información esperada.")
else:
print(f"Error en la solicitud: Código de estado {respuesta.status_code}")
except Exception as error:
print(f"Ocurrió un error al conectar con la API: {error}")
def main():
# Reemplaza 'TU_API_KEY' con tu clave de OpenWeatherMap
api_key = 'TU_API_KEY'
ciudad = input("Ingrese el nombre de la ciudad: ")
obtener_clima(ciudad, api_key)
if __name__ == "__main__":
main()
Recomendación: Regístrate en OpenWeatherMap para obtener una API Key gratuita. Recuerda manejar adecuadamente posibles excepciones o respuestas inesperadas.
Conclusión
Estos proyectos demuestran cómo, con Python, se pueden abordar diferentes desafíos prácticos:
- Monitoreo y gestión de procesos para controlar y finalizar aplicaciones en ejecución.
- Uso de hilos para procesar información de múltiples usuarios y gestionar sesiones de forma aislada.
- Simulación de sistemas concurrentes mediante el patrón productor-consumidor, útil en aplicaciones de comercio electrónico.
- Interacción con APIs REST para integrar datos externos en nuestras aplicaciones.
Cada ejemplo se ha explicado paso a paso, de forma que puedas adaptar y expandir estas soluciones a tus propios proyectos. ¡Experimenta, modifica el código y sigue aprendiendo!
