The 7 things you should NOT use threads for in Python (that every programmer should know)

import threading
import requests
import time

def obtener_datos_api(url, resultados, indice):
    """Each thread makes one request"""
    response = requests.get(url)
    resultados[indice] = response.json()

# WRONG - One thread per request
urls = [f"https://api.ejemplo.com/usuario/{i}" for i in range(100)]
resultados = [None] * 100
threads = []

inicio = time.time()

for i, url in enumerate(urls):
    t = threading.Thread(target=obtener_datos_api, args=(url, resultados, i))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Tiempo con threads: {time.time() - inicio:.2f}s")
print(f"Espacio de direcciones reservado: ~{len(threads) * 8}MB")

import asyncio
import aiohttp
import time

async def obtener_datos_async(session, url):
    """One lightweight coroutine per request"""
    async with session.get(url) as response:
        return await response.json()

async def obtener_todos_los_datos():
    urls = [f"https://api.ejemplo.com/usuario/{i}" for i in range(100)]

    # A single reusable session object
    async with aiohttp.ClientSession() as session:
        # Create every task
        tareas = [obtener_datos_async(session, url) for url in urls]

        # Run them all in parallel
        return await asyncio.gather(*tareas)

# RIGHT - No threads, 100x less memory
inicio = time.time()
resultados = asyncio.run(obtener_todos_los_datos())
print(f"Tiempo con asyncio: {time.time() - inicio:.2f}s")
print(f"Memoria usada: ~2MB")  # Instead of 800MB of reserved stacks

# Bonus: Concurrency control
async def obtener_con_limite():
    """Caps it at 10 simultaneous connections"""
    urls = [f"https://api.ejemplo.com/usuario/{i}" for i in range(100)]

    # Semaphore to limit concurrency
    limite = asyncio.Semaphore(10)

    async def fetch_limitado(session, url):
        async with limite:  # 10 at a time, max
            return await obtener_datos_async(session, url)

    async with aiohttp.ClientSession() as session:
        tareas = [fetch_limitado(session, url) for url in urls]
        return await asyncio.gather(*tareas)

import threading
import time

def procesar_archivo_con_thread(nombre_archivo, resultados):
    """Processes a file in a separate thread"""
    contador_lineas = 0
    suma_valores = 0

    with open(nombre_archivo, 'r') as f:
        for linea in f:
            # Simulate processing
            valores = linea.strip().split(',')
            if valores:
                contador_lineas += 1
                suma_valores += len(valores)

    resultados[nombre_archivo] = {
        'lineas': contador_lineas,
        'suma': suma_valores
    }

# WRONG - One thread per file
archivos = [f"datos_{i}.csv" for i in range(10)]
resultados = {}
threads = []

for archivo in archivos:
    t = threading.Thread(target=procesar_archivo_con_thread, args=(archivo, resultados))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

import asyncio
import aiofiles
from pathlib import Path

async def procesar_archivo_async(nombre_archivo):
    """Processes a file without blocking"""
    contador_lineas = 0
    suma_valores = 0

    async with aiofiles.open(nombre_archivo, 'r') as f:
        async for linea in f:
            # Processing without blocking the event loop
            valores = linea.strip().split(',')
            if valores:
                contador_lineas += 1
                suma_valores += len(valores)

            # Hand control back to the event loop every 1000 lines
            if contador_lineas % 1000 == 0:
                await asyncio.sleep(0)

    return {
        'archivo': nombre_archivo,
        'lineas': contador_lineas,
        'suma': suma_valores
    }

async def procesar_todos_los_archivos():
    archivos = [f"datos_{i}.csv" for i in range(10)]

    # Process them all in parallel
    tareas = [procesar_archivo_async(archivo) for archivo in archivos]
    return await asyncio.gather(*tareas)

# Run it
resultados = asyncio.run(procesar_todos_los_archivos())

# Alternative: Efficient generator for large files
def procesar_archivos_eficiente(archivos):
    """Uses generators for constant memory"""

    def procesar_linea(linea):
        # Your logic here
        return len(linea.strip().split(','))

    for archivo in archivos:
        # Generator that does not load everything into memory
        with open(archivo, 'r') as f:
            # Process in chunks without threads
            resultado = sum(
                procesar_linea(linea)
                for linea in f
                if linea.strip()
            )

            yield archivo, resultado

# Efficient use with no threads and no async
archivos = [f"datos_{i}.csv" for i in range(10)]
for archivo, resultado in procesar_archivos_eficiente(archivos):
    print(f"{archivo}: {resultado}")

from multiprocessing import Pool
import mmap
import os

def procesar_chunk_mmap(args):
    """Processes a file chunk with memory mapping"""
    nombre_archivo, inicio, fin = args

    with open(nombre_archivo, 'r+b') as f:
        # Memory mapping - does not load the file into RAM
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mmapped:
            # Only process the assigned chunk
            chunk = mmapped[inicio:fin]

            # Your processing here
            lineas = chunk.count(b'\n')
            return lineas

def procesar_archivo_grande_paralelo(nombre_archivo, num_procesos=4):
    """Splits a large file into chunks for parallel processing"""

    # Get the file size
    tamano = os.path.getsize(nombre_archivo)
    chunk_size = tamano // num_procesos

    # Build the arguments for each process
    chunks = []
    for i in range(num_procesos):
        inicio = i * chunk_size
        fin = tamano if i == num_procesos - 1 else (i + 1) * chunk_size
        chunks.append((nombre_archivo, inicio, fin))

    # REAL parallel processing
    with Pool(num_procesos) as pool:
        resultados = pool.map(procesar_chunk_mmap, chunks)

    return sum(resultados)

# For a 1GB file - 4x faster than threads
total_lineas = procesar_archivo_grande_paralelo("archivo_enorme.txt")

import tkinter as tk
import threading
import time
import random

class AppConThreads:
    def __init__(self, root):
        self.root = root
        self.label = tk.Label(root, text="Contador: 0")
        self.label.pack()

        self.boton = tk.Button(root, text="Iniciar", command=self.iniciar_contador)
        self.boton.pack()

        self.contador = 0

    def actualizar_gui_desde_thread(self):
        """DANGER - Updating the GUI from a thread"""
        for i in range(100):
            time.sleep(0.1)
            self.contador = i

            # THIS WILL CAUSE RANDOM CRASHES
            self.label.config(text=f"Contador: {self.contador}")

            # Sometimes it works, sometimes it crashes, always unpredictable

    def iniciar_contador(self):
        # WRONG - Thread updating the GUI directly
        t = threading.Thread(target=self.actualizar_gui_desde_thread)
        t.daemon = True
        t.start()

# This will crash at random
root = tk.Tk()
app = AppConThreads(root)
root.mainloop()

import tkinter as tk
import queue
import time
from threading import Thread

class AppSegura:
    def __init__(self, root):
        self.root = root
        self.label = tk.Label(root, text="Contador: 0")
        self.label.pack()

        self.progress = tk.Label(root, text="Progreso: 0%")
        self.progress.pack()

        self.boton = tk.Button(root, text="Iniciar", command=self.iniciar_proceso)
        self.boton.pack()

        # Thread-safe queue for communication
        self.cola_updates = queue.Queue()

        # Start the update loop
        self.procesar_cola()

    def trabajo_en_background(self):
        """Heavy work in a separate thread"""
        for i in range(101):
            time.sleep(0.05)  # Simulate work

            # Send the update through the queue
            self.cola_updates.put({
                'tipo': 'contador',
                'valor': i
            })

            if i % 10 == 0:
                self.cola_updates.put({
                    'tipo': 'progreso',
                    'valor': i
                })

    def procesar_cola(self):
        """Processes updates in the main thread"""
        try:
            while True:
                # Do not block, process everything available
                update = self.cola_updates.get_nowait()

                if update['tipo'] == 'contador':
                    self.label.config(text=f"Contador: {update['valor']}")
                elif update['tipo'] == 'progreso':
                    self.progress.config(text=f"Progreso: {update['valor']}%")

        except queue.Empty:
            pass

        # Schedule the next check
        self.root.after(50, self.procesar_cola)

    def iniciar_proceso(self):
        """Starts background work safely"""
        thread = Thread(target=self.trabajo_en_background)
        thread.daemon = True
        thread.start()

# Modern version with asyncio and tkinter
import asyncio

class AppAsync:
    """asyncio and Tk share a thread, so touching the GUI here is fine"""

    def __init__(self, root):
        self.root = root
        self.viva = True
        self.label = tk.Label(root, text="Estado: Esperando")
        self.label.pack()

        self.boton = tk.Button(
            root,
            text="Iniciar Async",
            command=lambda: asyncio.create_task(self.proceso_async())
        )
        self.boton.pack()

        root.protocol("WM_DELETE_WINDOW", self.cerrar)

    def cerrar(self):
        self.viva = False

    async def proceso_async(self):
        """Async process that does not block the GUI"""
        for i in range(10):
            # Same thread as the Tk loop: this is safe
            self.label.config(text=f"Procesando: {i+1}/10")

            # Hand control back to the event loop
            await asyncio.sleep(0.5)

        self.label.config(text="Completado!")

async def ejecutar_app():
    """No mainloop(): the event loop pumps Tk, all on a single thread.

    That is what makes create_task() work from the button callback:
    by the time Tk calls it, the asyncio loop is already running.
    """
    root = tk.Tk()
    app = AppAsync(root)

    while app.viva:
        root.update()
        await asyncio.sleep(0.01)

    root.destroy()

asyncio.run(ejecutar_app())

# For PyQt/PySide - Use QThread properly
from PyQt5.QtCore import QThread, pyqtSignal, QObject
from PyQt5.QtWidgets import QApplication, QLabel, QPushButton, QVBoxLayout, QWidget

class TrabajadorSeguro(QObject):
    """Worker that emits signals instead of updating the GUI"""
    progreso = pyqtSignal(int)
    completado = pyqtSignal(str)

    def ejecutar(self):
        for i in range(101):
            time.sleep(0.05)
            # Emit signal - thread-safe
            self.progreso.emit(i)

        self.completado.emit("Proceso terminado")

class VentanaSegura(QWidget):
    def __init__(self):
        super().__init__()
        layout = QVBoxLayout()

        self.label = QLabel("0%")
        layout.addWidget(self.label)

        self.boton = QPushButton("Iniciar")
        self.boton.clicked.connect(self.iniciar_trabajo)
        layout.addWidget(self.boton)

        self.setLayout(layout)

        # Set up thread and worker
        self.thread = QThread()
        self.worker = TrabajadorSeguro()
        self.worker.moveToThread(self.thread)

        # Connect signals
        self.thread.started.connect(self.worker.ejecutar)
        self.worker.progreso.connect(self.actualizar_progreso)
        self.worker.completado.connect(self.trabajo_completado)

    def iniciar_trabajo(self):
        self.thread.start()

    def actualizar_progreso(self, valor):
        """Thread-safe update coming from a signal"""
        self.label.setText(f"{valor}%")

    def trabajo_completado(self, mensaje):
        self.label.setText(mensaje)
        self.thread.quit()

import threading
import logging
import time
import random

# WRONG - Shared logger with no protection
logging.basicConfig(
    level=logging.INFO,
    format='%(threadName)s - %(message)s',
    filename='app.log'
)

def trabajador_con_logs(worker_id):
    """Every thread writes logs"""
    for i in range(10):
        # Simulate work
        time.sleep(random.uniform(0.01, 0.1))

        # PROBLEM: the logs get interleaved
        logging.info(f"Worker {worker_id} - Paso {i}")

        # Multiline logs = TOTAL CHAOS
        logging.info(f"""Worker {worker_id} procesando:
        - Item: {i}
        - Estado: Activo
        - Progreso: {i*10}%""")

# Create threads that write logs
threads = []
for i in range(5):
    t = threading.Thread(target=trabajador_con_logs, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# The log file will be unreadable

import logging
import logging.handlers
import queue
import threading
import atexit
import time
import traceback

class LoggerSeguro:
    """Thread-safe logging system backed by a queue"""

    def __init__(self, nombre_archivo='app.log'):
        # Queue for the logs
        self.cola_logs = queue.Queue()

        # Handler that writes from a single thread
        self.queue_handler = logging.handlers.QueueHandler(self.cola_logs)

        # Listener that drains the queue
        file_handler = logging.FileHandler(nombre_archivo)
        file_handler.setFormatter(
            logging.Formatter('%(asctime)s - %(threadName)s - %(levelname)s - %(message)s')
        )

        self.listener = logging.handlers.QueueListener(
            self.cola_logs,
            file_handler
        )

        # Configure the root logger
        self.logger = logging.getLogger()
        self.logger.addHandler(self.queue_handler)
        self.logger.setLevel(logging.INFO)

        # Start the listener
        self.listener.start()

        # Make sure we clean up on exit
        atexit.register(self.listener.stop)

    def get_logger(self, nombre=None):
        """Get a logger for a module"""
        return logging.getLogger(nombre)

# Structured logging system
import json
from datetime import datetime

class LoggerEstructurado:
    """Logger that produces structured JSON"""

    def __init__(self, nombre_archivo='app.jsonl'):
        self.lock = threading.Lock()
        self.archivo = nombre_archivo
        self.buffer = []
        self.buffer_size = 100  # Write every 100 logs

    def log(self, nivel, mensaje, **campos_extra):
        """Thread-safe log with structured data"""

        entrada = {
            'timestamp': datetime.utcnow().isoformat(),
            'nivel': nivel,
            'thread': threading.current_thread().name,
            'mensaje': mensaje,
            **campos_extra
        }

        with self.lock:
            self.buffer.append(entrada)

            if len(self.buffer) >= self.buffer_size:
                self._flush()

    def _flush(self):
        """Write the buffer to disk (must be called holding the lock)"""
        if not self.buffer:
            return

        with open(self.archivo, 'a') as f:
            for entrada in self.buffer:
                f.write(json.dumps(entrada) + '\n')

        self.buffer.clear()

    def info(self, mensaje, **kwargs):
        self.log('INFO', mensaje, **kwargs)

    def error(self, mensaje, **kwargs):
        self.log('ERROR', mensaje, **kwargs)

    def __del__(self):
        """Make sure every log gets written"""
        with self.lock:
            self._flush()

# Correct usage
logger_seguro = LoggerEstructurado()

def trabajador_mejorado(worker_id):
    """Worker with structured logging"""
    for i in range(10):
        # Log with context
        logger_seguro.info(
            "Procesando item",
            worker_id=worker_id,
            item=i,
            progreso=i*10
        )

        try:
            # Simulate work
            resultado = procesar_item(i)

            logger_seguro.info(
                "Item procesado",
                worker_id=worker_id,
                item=i,
                resultado=resultado
            )

        except Exception as e:
            logger_seguro.error(
                "Error procesando item",
                worker_id=worker_id,
                item=i,
                error=str(e),
                traceback=traceback.format_exc()
            )

def procesar_item(item):
    time.sleep(0.1)
    return item * 2

# Alternative: Centralized logging with asyncio
import asyncio
import aiofiles

class LoggerAsync:
    """Async logger with no blocking"""

    def __init__(self, archivo='app_async.log'):
        self.archivo = archivo
        self.cola = asyncio.Queue()
        self.task = None

    async def iniciar(self):
        """Starts the log processor"""
        self.task = asyncio.create_task(self._procesador())

    async def _procesador(self):
        """Drains logs from the queue"""
        async with aiofiles.open(self.archivo, 'a') as f:
            while True:
                try:
                    # Wait for logs with a timeout
                    entrada = await asyncio.wait_for(
                        self.cola.get(),
                        timeout=1.0
                    )

                    # Write asynchronously
                    await f.write(entrada + '\n')
                    await f.flush()

                except asyncio.TimeoutError:
                    continue
                except asyncio.CancelledError:
                    # Flush the remaining logs before closing
                    while not self.cola.empty():
                        entrada = await self.cola.get()
                        await f.write(entrada + '\n')
                    break

    async def log(self, nivel, mensaje):
        """Add a log to the queue"""
        timestamp = datetime.utcnow().isoformat()
        entrada = f"[{timestamp}] {nivel}: {mensaje}"
        await self.cola.put(entrada)

    async def cerrar(self):
        """Close the logger cleanly"""
        if self.task:
            self.task.cancel()
            await self.task

import threading
import time
import random

class CacheRota:
    """Cache with race conditions"""
    def __init__(self):
        self.cache = {}
        self.lock = threading.Lock()

    def obtener_o_calcular(self, clave, funcion_costosa):
        # PROBLEM 1: Check-then-act race condition
        if clave not in self.cache:
            # Between these two lines another thread can add the key
            with self.lock:
                # PROBLEM 2: Double computation
                # Several threads can get here
                resultado = funcion_costosa()
                self.cache[clave] = resultado
                return resultado

        return self.cache[clave]

def operacion_costosa():
    """Simulates an expensive operation"""
    print(f"Thread {threading.current_thread().name} calculando...")
    time.sleep(1)
    return random.randint(1, 100)

# Several threads = several computations for the same key
cache = CacheRota()
threads = []

for i in range(5):
    t = threading.Thread(
        target=lambda: print(cache.obtener_o_calcular("key1", operacion_costosa))
    )
    threads.append(t)
    t.start()

# You will see "calculando..." several times for the same key!

from functools import lru_cache, wraps
import threading
import time

# Solution 1: Built in thread-safe LRU cache
@lru_cache(maxsize=128)
def operacion_con_cache(parametro):
    """lru_cache is thread-safe by default"""
    print(f"Calculando para {parametro}...")
    time.sleep(1)
    return parametro ** 2

# Solution 2: Custom thread-safe cache
class CacheInteligente:
    """Cache with per key locking and lazy loading"""

    def __init__(self, ttl=60):
        self.cache = {}
        self.locks = {}
        self.lock_principal = threading.Lock()
        self.ttl = ttl

    def obtener_o_calcular(self, clave, funcion):
        """Reads from cache or computes, with no double computation"""

        # Fast path - already cached
        if clave in self.cache:
            valor, timestamp = self.cache[clave]
            if time.time() - timestamp < self.ttl:
                return valor

        # Get the lock specific to this key
        with self.lock_principal:
            if clave not in self.locks:
                self.locks[clave] = threading.Lock()
            lock_clave = self.locks[clave]

        # Now only one thread will compute for this key
        with lock_clave:
            # Double-check pattern
            if clave in self.cache:
                valor, timestamp = self.cache[clave]
                if time.time() - timestamp < self.ttl:
                    return valor

            # Compute only once
            print(f"Calculando {clave} en {threading.current_thread().name}")
            valor = funcion()
            self.cache[clave] = (valor, time.time())

            return valor

    def limpiar_expirados(self):
        """Removes expired entries"""
        with self.lock_principal:
            ahora = time.time()
            claves_expiradas = [
                k for k, (_, timestamp) in self.cache.items()
                if ahora - timestamp >= self.ttl
            ]

            for k in claves_expiradas:
                del self.cache[k]
                if k in self.locks:
                    del self.locks[k]

# Solution 3: Cache with asyncio (the best option)
import asyncio
from typing import Dict, Any, Callable, Optional

class CacheAsync:
    """Async cache with no blocking"""

    def __init__(self, ttl: int = 60):
        self.cache: Dict[str, tuple[Any, float]] = {}
        self.calculando: Dict[str, asyncio.Future] = {}
        self.ttl = ttl

    async def obtener_o_calcular(
        self,
        clave: str,
        funcion: Callable,
        *args,
        **kwargs
    ):
        """Reads from cache or computes asynchronously"""

        # Check the cache
        if clave in self.cache:
            valor, timestamp = self.cache[clave]
            if time.time() - timestamp < self.ttl:
                return valor

        # Check whether it is already being computed
        if clave in self.calculando:
            # Wait for the result of the computation in flight
            return await self.calculando[clave]

        # Create a future so the other coroutines can wait on it
        future = asyncio.Future()
        self.calculando[clave] = future

        try:
            # Compute the value
            if asyncio.iscoroutinefunction(funcion):
                valor = await funcion(*args, **kwargs)
            else:
                valor = await asyncio.to_thread(funcion, *args, **kwargs)

            # Store it in the cache
            self.cache[clave] = (valor, time.time())

            # Resolve the future for the coroutines that are waiting
            future.set_result(valor)

            return valor

        except Exception as e:
            future.set_exception(e)
            raise
        finally:
            # Clean up the future
            del self.calculando[clave]

# Decorator to cache any function
def cache_decorator(ttl=60):
    """Decorator that adds a thread-safe cache to any function"""
    cache = {}
    locks = {}
    lock_principal = threading.Lock()

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Build a unique key from the arguments
            clave = f"{func.__name__}_{args}_{kwargs}"

            # Quick check without the lock
            if clave in cache:
                valor, timestamp = cache[clave]
                if time.time() - timestamp < ttl:
                    return valor

            # Get or create the lock for this key
            with lock_principal:
                if clave not in locks:
                    locks[clave] = threading.Lock()
                lock = locks[clave]

            # Compute holding the lock
            with lock:
                # Double-check
                if clave in cache:
                    valor, timestamp = cache[clave]
                    if time.time() - timestamp < ttl:
                        return valor

                # Compute
                valor = func(*args, **kwargs)
                cache[clave] = (valor, time.time())
                return valor

        return wrapper
    return decorator

# Usage
@cache_decorator(ttl=30)
def operacion_costosa(x, y):
    print(f"Calculando {x} + {y}")
    time.sleep(1)
    return x + y

# Test with several threads
def test_cache():
    for _ in range(3):
        resultado = operacion_costosa(5, 3)
        print(f"Resultado: {resultado}")

threads = []
for i in range(10):
    t = threading.Thread(target=test_cache)
    threads.append(t)
    t.start()

# You will only see "Calculando 5 + 3" once!

import subprocess
import threading
import time

def ejecutar_comando_mal(comando, resultados, indice):
    """Runs a command in a thread - PROBLEMATIC"""
    try:
        # PROBLEM: subprocess is not thread-safe on every platform
        resultado = subprocess.run(
            comando,
            shell=True,
            capture_output=True,
            text=True,
            timeout=5
        )
        resultados[indice] = resultado.stdout
    except subprocess.TimeoutExpired:
        resultados[indice] = "Timeout"
    except Exception as e:
        resultados[indice] = f"Error: {e}"

# WRONG - Several threads spawning subprocesses
comandos = [
    "sleep 1 && echo 'Comando 1'",
    "sleep 2 && echo 'Comando 2'",
    "sleep 1 && echo 'Comando 3'",
]

resultados = [None] * len(comandos)
threads = []

for i, cmd in enumerate(comandos):
    t = threading.Thread(target=ejecutar_comando_mal, args=(cmd, resultados, i))
    threads.append(t)
    t.start()

# Problems:
# 1. File descriptor leaks
# 2. Zombie processes
# 3. Lost signals

import asyncio
import sys

async def ejecutar_comando_async(comando):
    """Runs a command asynchronously and safely"""

    # Create the async process
    proceso = await asyncio.create_subprocess_shell(
        comando,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    # Wait for the result with a timeout
    try:
        stdout, stderr = await asyncio.wait_for(
            proceso.communicate(),
            timeout=5.0
        )

        return {
            'comando': comando,
            'codigo': proceso.returncode,
            'stdout': stdout.decode(),
            'stderr': stderr.decode()
        }

    except asyncio.TimeoutError:
        # Kill the process that is not responding
        proceso.kill()
        await proceso.wait()

        return {
            'comando': comando,
            'error': 'Timeout después de 5 segundos'
        }

async def ejecutar_comandos_paralelo(comandos):
    """Runs several commands in parallel"""

    # Limit concurrency so we do not overload the system
    semaforo = asyncio.Semaphore(5)

    async def ejecutar_con_limite(cmd):
        async with semaforo:
            return await ejecutar_comando_async(cmd)

    # Run them all in parallel
    resultados = await asyncio.gather(
        *[ejecutar_con_limite(cmd) for cmd in comandos],
        return_exceptions=True
    )

    return resultados

# Usage
comandos = [
    "echo 'Hola 1' && sleep 1",
    "echo 'Hola 2' && sleep 2",
    "ls -la",
    "pwd",
    "date"
]

resultados = asyncio.run(ejecutar_comandos_paralelo(comandos))

for resultado in resultados:
    if isinstance(resultado, dict):
        print(f"Comando: {resultado.get('comando', 'unknown')}")
        print(f"Salida: {resultado.get('stdout', resultado.get('error', ''))}")

# Alternative solution with concurrent.futures
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import subprocess

class EjecutorComandos:
    """Command runner backed by a worker pool"""

    def __init__(self, max_workers=4):
        # Use ProcessPoolExecutor for full isolation
        self.executor = ProcessPoolExecutor(max_workers=max_workers)

    @staticmethod
    def _ejecutar_comando(comando, timeout=5):
        """Static function to run in a separate process"""
        try:
            resultado = subprocess.run(
                comando,
                shell=True,
                capture_output=True,
                text=True,
                timeout=timeout
            )

            return {
                'comando': comando,
                'exitcode': resultado.returncode,
                'stdout': resultado.stdout,
                'stderr': resultado.stderr,
                'success': resultado.returncode == 0
            }

        except subprocess.TimeoutExpired:
            return {
                'comando': comando,
                'error': f'Timeout después de {timeout} segundos',
                'success': False
            }
        except Exception as e:
            return {
                'comando': comando,
                'error': str(e),
                'success': False
            }

    def ejecutar_paralelo(self, comandos, timeout=5):
        """Runs commands in parallel, safely"""

        # Submit every task
        futuros = [
            self.executor.submit(self._ejecutar_comando, cmd, timeout)
            for cmd in comandos
        ]

        # Collect the results
        resultados = []
        for futuro in futuros:
            try:
                resultado = futuro.result(timeout=timeout + 1)
                resultados.append(resultado)
            except Exception as e:
                resultados.append({
                    'error': f'Error ejecutando: {e}',
                    'success': False
                })

        return resultados

    def cerrar(self):
        """Shuts the executor down cleanly"""
        self.executor.shutdown(wait=True)

# Safe usage
executor = EjecutorComandos(max_workers=4)

comandos = [
    "echo 'Proceso 1'",
    "python -c 'print(2+2)'",
    "ls -la | head -5",
    "sleep 10"  # This one will time out
]

resultados = executor.ejecutar_paralelo(comandos, timeout=3)

for r in resultados:
    if r['success']:
        print(f"✓ {r['comando']}: {r['stdout'].strip()}")
    else:
        print(f"✗ {r['comando']}: {r.get('error', 'Error desconocido')}")

executor.cerrar()

# Advanced handling with pipes and streams
async def ejecutar_con_streaming(comando):
    """Runs a command and processes its output in real time"""

    proceso = await asyncio.create_subprocess_shell(
        comando,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT
    )

    # Read the output line by line
    async for linea in proceso.stdout:
        print(f"[{comando[:20]}...]: {linea.decode().strip()}")

    await proceso.wait()
    return proceso.returncode

# Run a long command with real time output
asyncio.run(ejecutar_con_streaming("for i in {1..5}; do echo $i; sleep 1; done"))

import threading
import sqlite3
import time
import random

# WRONG - Connection shared between threads
conexion = sqlite3.connect('database.db', check_same_thread=False)
cursor = conexion.cursor()

# Create the table
cursor.execute('''
    CREATE TABLE IF NOT EXISTS cuentas (
        id INTEGER PRIMARY KEY,
        saldo REAL
    )
''')
cursor.execute("INSERT OR IGNORE INTO cuentas VALUES (1, 1000)")
conexion.commit()

def transferir_dinero_mal(de_cuenta, a_cuenta, cantidad):
    """Transfer with race conditions"""

    # PROBLEM 1: Inconsistent reads
    cursor.execute("SELECT saldo FROM cuentas WHERE id = ?", (de_cuenta,))
    saldo_origen = cursor.fetchone()[0]

    # PROBLEM 2: Between these lines another thread can modify the data
    time.sleep(random.uniform(0.001, 0.01))  # Simulate latency

    if saldo_origen >= cantidad:
        # PROBLEM 3: Non atomic transactions
        cursor.execute(
            "UPDATE cuentas SET saldo = saldo - ? WHERE id = ?",
            (cantidad, de_cuenta)
        )
        cursor.execute(
            "UPDATE cuentas SET saldo = saldo + ? WHERE id = ?",
            (cantidad, a_cuenta)
        )
        conexion.commit()
        print(f"Transferido {cantidad} de {de_cuenta} a {a_cuenta}")
    else:
        print(f"Saldo insuficiente")

# Create more accounts
for i in range(2, 5):
    cursor.execute("INSERT OR IGNORE INTO cuentas VALUES (?, 1000)", (i,))
conexion.commit()

# Several threads = CHAOS
threads = []
for _ in range(10):
    cuenta_origen = random.randint(1, 4)
    cuenta_destino = random.randint(1, 4)
    while cuenta_destino == cuenta_origen:
        cuenta_destino = random.randint(1, 4)

    cantidad = random.randint(10, 100)

    t = threading.Thread(
        target=transferir_dinero_mal,
        args=(cuenta_origen, cuenta_destino, cantidad)
    )
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# Check the integrity - the total amount of money changed!
cursor.execute("SELECT SUM(saldo) FROM cuentas")
print(f"Saldo total: {cursor.fetchone()[0]} (debería ser 4000)")

import sqlite3
from contextlib import contextmanager
import threading
import queue
import time

class PoolConexionesSeguro:
    """Thread-safe connection pool"""

    def __init__(self, database, max_conexiones=5):
        self.database = database
        self.pool = queue.Queue(maxsize=max_conexiones)

        # Create the initial connections
        for _ in range(max_conexiones):
            conn = self._crear_conexion()
            self.pool.put(conn)

    def _crear_conexion(self):
        """Creates a properly configured connection"""
        conn = sqlite3.connect(self.database)
        conn.execute("PRAGMA journal_mode=WAL")  # Write-Ahead Logging
        conn.execute("PRAGMA busy_timeout=5000")  # 5 second timeout
        conn.row_factory = sqlite3.Row  # Results as dictionaries
        return conn

    @contextmanager
    def obtener_conexion(self):
        """Context manager to take a connection from the pool"""
        conexion = self.pool.get()
        try:
            yield conexion
        finally:
            # Automatic rollback if a transaction is still open
            conexion.rollback()
            self.pool.put(conexion)

    @contextmanager
    def transaccion(self):
        """Context manager for automatic transactions"""
        with self.obtener_conexion() as conn:
            try:
                yield conn
                conn.commit()
            except Exception:
                conn.rollback()
                raise

# Safe operations using the pool
pool = PoolConexionesSeguro('database_segura.db', max_conexiones=3)

def inicializar_db():
    with pool.transaccion() as conn:
        conn.execute('''
            CREATE TABLE IF NOT EXISTS cuentas (
                id INTEGER PRIMARY KEY,
                saldo REAL NOT NULL,
                version INTEGER DEFAULT 0
            )
        ''')

        # Create the initial accounts
        for i in range(1, 5):
            conn.execute(
                "INSERT OR IGNORE INTO cuentas (id, saldo) VALUES (?, ?)",
                (i, 1000.0)
            )

def transferir_dinero_seguro(de_cuenta, a_cuenta, cantidad):
    """Transfer with optimistic concurrency control"""

    with pool.transaccion() as conn:
        # Row level locking, simulating SELECT FOR UPDATE
        cursor = conn.execute(
            "SELECT saldo, version FROM cuentas WHERE id = ?",
            (de_cuenta,)
        )
        row = cursor.fetchone()

        if not row:
            raise ValueError(f"Cuenta {de_cuenta} no existe")

        saldo_actual = row[0]
        version_actual = row[1]

        if saldo_actual < cantidad:
            raise ValueError("Saldo insuficiente")

        # Update with a version check
        result = conn.execute("""
            UPDATE cuentas
            SET saldo = saldo - ?, version = version + 1
            WHERE id = ? AND version = ?
        """, (cantidad, de_cuenta, version_actual))

        if result.rowcount == 0:
            raise ValueError("Conflicto de concurrencia - reintentar")

        # Update the destination account
        conn.execute(
            "UPDATE cuentas SET saldo = saldo + ? WHERE id = ?",
            (cantidad, a_cuenta)
        )

        print(f"✓ Transferido {cantidad} de cuenta {de_cuenta} a {a_cuenta}")

# Version with SQLAlchemy (better for production)
from sqlalchemy import create_engine, Column, Integer, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import QueuePool

Base = declarative_base()

class Cuenta(Base):
    __tablename__ = 'cuentas'

    id = Column(Integer, primary_key=True)
    saldo = Column(Float, nullable=False)
    version = Column(Integer, default=0)

# Configure the engine with a pool
engine = create_engine(
    'sqlite:///database_sqlalchemy.db',
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,  # Check connections before using them
    poolclass=QueuePool
)

# Thread-safe session factory
SessionFactory = scoped_session(sessionmaker(bind=engine))

def transferir_con_sqlalchemy(de_cuenta, a_cuenta, cantidad):
    """Transfer using the SQLAlchemy ORM"""

    session = SessionFactory()

    try:
        # Pessimistic locking
        cuenta_origen = session.query(Cuenta).filter_by(
            id=de_cuenta
        ).with_for_update().first()

        cuenta_destino = session.query(Cuenta).filter_by(
            id=a_cuenta
        ).with_for_update().first()

        if cuenta_origen.saldo >= cantidad:
            cuenta_origen.saldo -= cantidad
            cuenta_destino.saldo += cantidad

            session.commit()
            print(f"✓ SQLAlchemy: Transferido {cantidad}")
        else:
            session.rollback()
            print("✗ Saldo insuficiente")

    except Exception as e:
        session.rollback()
        print(f"Error: {e}")
    finally:
        session.close()

# Async with asyncpg (PostgreSQL)
import asyncio
import asyncpg

class PoolAsync:
    """Async connection pool"""

    def __init__(self):
        self.pool = None

    async def inicializar(self):
        self.pool = await asyncpg.create_pool(
            'postgresql://usuario:password@localhost/db',
            min_size=2,
            max_size=10
        )

    async def transferir_async(self, de_cuenta, a_cuenta, cantidad):
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                # Everything inside one transaction
                saldo = await conn.fetchval(
                    "SELECT saldo FROM cuentas WHERE id = $1 FOR UPDATE",
                    de_cuenta
                )

                if saldo >= cantidad:
                    await conn.execute(
                        "UPDATE cuentas SET saldo = saldo - $1 WHERE id = $2",
                        cantidad, de_cuenta
                    )
                    await conn.execute(
                        "UPDATE cuentas SET saldo = saldo + $1 WHERE id = $2",
                        cantidad, a_cuenta
                    )
                    return True
                return False

# Integrity test
inicializar_db()

threads = []
for _ in range(20):
    t = threading.Thread(
        target=transferir_dinero_seguro,
        args=(
            random.randint(1, 4),
            random.randint(1, 4),
            random.randint(10, 50)
        )
    )
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# Check the integrity
with pool.obtener_conexion() as conn:
    cursor = conn.execute("SELECT SUM(saldo) FROM cuentas")
    total = cursor.fetchone()[0]
    print(f"Saldo total final: {total} (debe ser 4000)")