Python: De Scripting a Ingeniería AI/ML

Sintaxis Moderna y Rendimiento

Python 3.12 introdujo mensajes de error mejorados, GIL por intérprete (experimental), la sentencia type para alias de tipos y mejoras en f-strings. Python 3.13 agrega un compilador JIT (experimental) y avances en free-threading. Estas versiones se enfocan en rendimiento y experiencia del desarrollador.

# Python 3.12+: type statement for cleaner type aliases
from collections.abc import Sequence, Callable

type Vector = list[float]
type Matrix = list[Vector]
type Transformer[T] = Callable[[T], T]
type JSONValue = str | int | float | bool | None | list["JSONValue"] | dict[str, "JSONValue"]

# Improved f-strings (3.12): any expression allowed, including quotes
data = {"name": "Jose", "weight_kg": 78.5}
msg = f"User {data['name']} weighs {data['weight_kg']:.1f}kg"

# Structural pattern matching (3.10+)
def process_event(event: dict) -> str:
    match event:
        case {"type": "weight", "value": float(v)} if v > 0:
            return f"Weight: {v}kg"
        case {"type": "body_fat", "value": float(v)}:
            return f"Body fat: {v}%"
        case {"type": str(t), **rest}:
            return f"Unknown metric: {t}"
        case _:
            return "Invalid event"

# Exception groups and except* (3.11+)
async def fetch_all(urls: list[str]) -> list[str]:
    results: list[str] = []
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(url)) for url in urls]
    except* ConnectionError as eg:
        for exc in eg.exceptions:
            logger.warning(f"Connection failed: {exc}")
    except* TimeoutError as eg:
        for exc in eg.exceptions:
            logger.warning(f"Timeout: {exc}")
    return [t.result() for t in tasks if not t.cancelled()]

Automatización de Sistemas y Daemons

Python se destaca en automatización de sistemas: procesamiento de archivos, integración con APIs, pipelines de datos y daemons de larga ejecución. La biblioteca estándar proporciona subprocess, pathlib, asyncio y json para la mayoría de tareas.

#!/usr/bin/env python3
"""Xiaomi health data export daemon.
Automates data export from Xiaomi Health app via headless AVD,
processes exported body composition data, and syncs to the
health dashboard API.
"""
import asyncio
import subprocess
import json
import logging
from datetime import datetime, timezone
from pathlib import Path

logger = logging.getLogger(__name__)

AVD_NAME = "xiaomi_health"
EXPORT_PATH = Path("/tmp/xiaomi_export/health_data.json")

async def trigger_export() -> dict:
    """Use ADB + uiautomator to export data from Xiaomi Health app."""
    # Wake the headless AVD and launch the app
    subprocess.run(["adb", "shell", "input", "keyevent", "KEYCODE_WAKEUP"])
    subprocess.run(["adb", "shell", "am", "start", "-n",
                     "com.mi.health/.MainAct"])
    await asyncio.sleep(3)

    # Navigate to export via uiautomator
    subprocess.run(["adb", "shell", "uiautomator", "dump"])
    subprocess.run(["adb", "shell", "input", "tap", "540", "1200"])
    await asyncio.sleep(2)

    # Pull exported data
    subprocess.run(["adb", "pull", "/sdcard/health_export.json",
                     str(EXPORT_PATH)])
    with open(EXPORT_PATH) as f:
        return json.load(f)

async def sync_loop(interval: int = 3600) -> None:
    """Main daemon loop: export data, process, sync to API."""
    while True:
        try:
            data = await trigger_export()
            processed = process_body_composition(data)
            await push_to_api(processed)
            logger.info(f"Synced: {processed['weight_kg']}kg")
        except Exception as e:
            logger.error(f"Sync failed: {e}")
        await asyncio.sleep(interval)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(sync_loop())

Herramientas CLI con Click, Typer y Rich

Click proporciona parsing de argumentos CLI basado en decoradores. Typer se construye sobre Click con interfaces basadas en type hints. Rich agrega salida formateada con tablas, barras de progreso y syntax highlighting. Juntos crean herramientas de línea de comandos de nivel profesional.

import click
from rich.console import Console
from rich.table import Table
from rich.progress import track

console = Console()

@click.group()
@click.option("--verbose", "-v", is_flag=True)
@click.pass_context
def cli(ctx, verbose):
    ctx.ensure_object(dict)
    ctx.obj["verbose"] = verbose

@cli.command()
@click.argument("directory", type=click.Path(exists=True))
@click.option("--output-format", "-f", type=click.Choice(["json", "csv"]), default="json")
def process(directory, output_format):
    """Process all data files in DIRECTORY."""
    files = list(Path(directory).glob("*.dat"))
    table = Table(title=f"Processing {len(files)} files")
    table.add_column("File", style="cyan")
    table.add_column("Records", justify="right")
    table.add_column("Status", style="green")

    for f in track(files, description="Processing..."):
        records = parse_data_file(f)
        table.add_row(f.name, str(len(records)), "OK")

    console.print(table)

# Typer: type-hint-driven CLI (built on Click)
import typer
from pathlib import Path
from enum import Enum

app = typer.Typer(help="Health data CLI tool.")

class OutputFormat(str, Enum):
    json = "json"
    csv = "csv"

@app.command()
def export(
    directory: Path = typer.Argument(..., exists=True, help="Data directory"),
    output_format: OutputFormat = typer.Option(OutputFormat.json, "--format", "-f"),
    days: int = typer.Option(30, "--days", "-d", min=1, max=365),
):
    """Export health metrics from DIRECTORY."""
    console.print(f"[bold]Exporting {days} days in {output_format.value}[/bold]")
    data = load_metrics(directory, days=days)
    write_output(data, fmt=output_format.value)

if __name__ == "__main__":
    app()

FastAPI: Framework Async Moderno para APIs

FastAPI (v0.140.0 a julio 2026) aprovecha los type hints de Python para validación automática de requests, serialización y documentación OpenAPI. Construido sobre Starlette y Pydantic v2, soporta inyección de dependencias, endpoints async y genera documentación interactiva de API automáticamente. Requiere Python 3.10+. Las últimas funcionalidades incluyen streaming JSON Lines y verificación estricta de content-type.

from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel, Field
from datetime import datetime, timedelta, timezone
from typing import Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI(title="Health API", version="2.0.0")

class BodyMetrics(BaseModel):
    weight_kg: float = Field(..., gt=0, lt=500, description="Weight in kilograms")
    body_fat_pct: Optional[float] = Field(None, ge=0, le=100)
    muscle_mass_kg: Optional[float] = Field(None, gt=0)
    recorded_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    model_config = {
        "json_schema_extra": {
            "example": {"weight_kg": 78.5, "body_fat_pct": 15.2, "muscle_mass_kg": 35.1}
        }
    }

class MetricsResponse(BaseModel):
    id: int
    metrics: BodyMetrics
    trend: str  # "up", "down", "stable"

# Dependency injection: reusable auth + DB session
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    payload = decode_jwt(token)
    user = await user_repo.get(payload["sub"])
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    return user

@app.post("/api/metrics", response_model=MetricsResponse, status_code=status.HTTP_201_CREATED)
async def create_metrics(
    metrics: BodyMetrics,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """Record new body composition metrics."""
    record = MetricsRecord(user_id=user.id, **metrics.model_dump())
    db.add(record)
    await db.commit()
    await db.refresh(record)
    trend = await calculate_trend(db, user.id)
    return MetricsResponse(id=record.id, metrics=metrics, trend=trend)

@app.get("/api/metrics", response_model=list[MetricsResponse])
async def list_metrics(
    days: int = 30,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """List metrics for the last N days."""
    since = datetime.now(timezone.utc) - timedelta(days=days)
    result = await db.execute(
        select(MetricsRecord)
        .where(MetricsRecord.user_id == user.id, MetricsRecord.recorded_at >= since)
        .order_by(MetricsRecord.recorded_at.desc())
    )
    return result.scalars().all()

Flask: Liviano y Flexible

Flask es un micro-framework con un núcleo simple y un rico ecosistema de extensiones. Es ideal para APIs pequeñas, webhooks y servicios donde no se requiere soporte async completo.

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from functools import wraps
import jwt

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///health.db"
db = SQLAlchemy(app)

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        try:
            payload = jwt.decode(token, app.config["SECRET_KEY"], algorithms=["HS256"])
            request.user_id = payload["user_id"]
        except jwt.InvalidTokenError:
            return jsonify({"error": "Invalid token"}), 401
        return f(*args, **kwargs)
    return decorated

@app.route("/api/webhook/scale", methods=["POST"])
@require_auth
def scale_webhook():
    """Receive scale data from IoT device."""
    data = request.get_json()
    record = Measurement(user_id=request.user_id, **data)
    db.session.add(record)
    db.session.commit()
    return jsonify({"id": record.id, "status": "recorded"}), 201

Análisis de Datos con pandas

pandas proporciona DataFrames de alto rendimiento para manipulación, limpieza, agregación y análisis de datos. Combinado con numpy para operaciones numéricas, forma la base de la ciencia de datos en Python.

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# Load and clean health data
df = pd.read_csv("health_metrics.csv", parse_dates=["recorded_at"])
df = df.dropna(subset=["weight_kg"])
df = df[df["weight_kg"].between(40, 200)]  # remove outliers

# Rolling averages for trend analysis
df["weight_7d_avg"] = df["weight_kg"].rolling(window=7, min_periods=3).mean()
df["body_fat_7d_avg"] = df["body_fat_pct"].rolling(window=7, min_periods=3).mean()

# Weekly aggregation
weekly = df.resample("W", on="recorded_at").agg({
    "weight_kg": ["mean", "min", "max", "std"],
    "body_fat_pct": "mean",
    "muscle_mass_kg": "mean",
})

# Correlation analysis
correlations = df[["weight_kg", "body_fat_pct", "muscle_mass_kg",
                   "sleep_hours", "calories_in"]].corr()

# Export for dashboard
dashboard_data = df.tail(90).to_json(orient="records", date_format="iso")

# Pivot table: average metrics by day of week
pivot = df.pivot_table(
    values=["weight_kg", "calories_in"],
    index=df["recorded_at"].dt.day_name(),
    aggfunc={"weight_kg": "mean", "calories_in": ["mean", "sum"]},
)

Visualización con matplotlib

matplotlib es la biblioteca de graficación fundamental de Python. Combinada con pandas, produce gráficos de calidad profesional para análisis de tendencias, distribuciones y dashboards.

import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import numpy as np

fig, axes = plt.subplots(2, 1, figsize=(12, 8), sharex=True)

# Weight trend with rolling average
ax1 = axes[0]
ax1.scatter(df["recorded_at"], df["weight_kg"], alpha=0.3, s=10, label="Daily")
ax1.plot(df["recorded_at"], df["weight_7d_avg"], color="tab:blue", linewidth=2, label="7-day avg")
ax1.fill_between(df["recorded_at"], df["weight_7d_avg"] - df["weight_kg"].std(),
                 df["weight_7d_avg"] + df["weight_kg"].std(), alpha=0.1)
ax1.set_ylabel("Weight (kg)")
ax1.legend()
ax1.grid(True, alpha=0.3)

# Body fat percentage
ax2 = axes[1]
ax2.plot(df["recorded_at"], df["body_fat_7d_avg"], color="tab:orange", linewidth=2)
ax2.set_ylabel("Body Fat (%)")
ax2.set_xlabel("Date")
ax2.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig("health_trends.png", dpi=150, bbox_inches="tight")

Fine-Tuning LoRA/QLoRA con Unsloth

LoRA (Low-Rank Adaptation) hace fine-tuning de modelos de lenguaje grandes entrenando pequeñas matrices adaptadoras en vez de todos los parámetros. QLoRA agrega cuantización de 4 bits, haciendo posible fine-tunear modelos de 70B+ en GPUs de consumidor. Unsloth optimiza el loop de entrenamiento para 2x de velocidad.

from unsloth import FastLanguageModel
from trl import SFTTrainer
from transformers import TrainingArguments
from datasets import load_dataset

# Load base model with 4-bit quantization (QLoRA)
model, tokenizer = FastLanguageModel.from_pretrained(
    model_name="meta-llama/Llama-3.1-8B-Instruct",
    max_seq_length=4096,
    dtype=None,  # auto-detect
    load_in_4bit=True,
    device_map="auto",
)

# Add LoRA adapters
model = FastLanguageModel.get_peft_model(
    model,
    r=64,                  # LoRA rank
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                     "gate_proj", "up_proj", "down_proj"],
    lora_alpha=128,
    lora_dropout=0.05,
    bias="none",
    use_gradient_checkpointing="unsloth",  # 60% less VRAM
)

# Prepare dataset
dataset = load_dataset("json", data_files="training_data.jsonl", split="train")

def format_prompt(example):
    return {
        "text": f"<|system|>You are a helpful health advisor.<|end|>\n"
                f"<|user|>{example['question']}<|end|>\n"
                f"<|assistant|>{example['answer']}<|end|>"
    }

dataset = dataset.map(format_prompt)

# Training configuration
trainer = SFTTrainer(
    model=model,
    tokenizer=tokenizer,
    train_dataset=dataset,
    dataset_text_field="text",
    max_seq_length=4096,
    args=TrainingArguments(
        per_device_train_batch_size=4,
        gradient_accumulation_steps=8,
        warmup_steps=100,
        num_train_epochs=3,
        learning_rate=2e-5,
        fp16=True,
        logging_steps=10,
        output_dir="./output",
        optim="adamw_8bit",
        seed=42,
    ),
)

trainer.train()

# Save LoRA adapter (only ~100MB vs ~16GB full fp16 model)
model.save_pretrained("./health-advisor-lora")
tokenizer.save_pretrained("./health-advisor-lora")

Ecosistema Hugging Face

Hugging Face proporciona la biblioteca transformers para carga de modelos, datasets para procesamiento de datos, peft para fine-tuning eficiente en parámetros, y el Hub para compartir modelos. Es el ecosistema estándar para NLP y AI multimodal moderna.

from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from peft import PeftModel, LoraConfig, get_peft_model
from datasets import load_dataset, DatasetDict
from huggingface_hub import HfApi

# Load and merge LoRA adapter with base model
base_model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B-Instruct",
    torch_dtype="auto",
    device_map="auto",
)
model = PeftModel.from_pretrained(base_model, "./health-advisor-lora")
model = model.merge_and_unload()  # merge LoRA weights into base

# Push merged model to Hugging Face Hub
model.push_to_hub("josenobile/health-advisor-merged")
tokenizer.push_to_hub("josenobile/health-advisor-merged")

# Dataset processing pipeline
dataset = load_dataset("csv", data_files="health_qa.csv")
dataset = dataset["train"].train_test_split(test_size=0.1, seed=42)

def tokenize(examples):
    return tokenizer(
        examples["text"],
        truncation=True,
        max_length=2048,
        padding="max_length",
    )

tokenized = dataset.map(tokenize, batched=True, num_proc=4)

# Quick inference with pipeline API
pipe = pipeline("text-generation", model=model, tokenizer=tokenizer)
result = pipe("What exercises help reduce body fat?", max_new_tokens=256)

Inferencia y Servicio de Modelos

Sirve modelos fine-tuneados via FastAPI con inferencia en batch, respuestas streaming y gestión de memoria GPU. vLLM proporciona servicio optimizado con PagedAttention para alto throughput.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from unsloth import FastLanguageModel
import torch
import asyncio

app = FastAPI()

model, tokenizer = FastLanguageModel.from_pretrained(
    "./health-advisor-lora",
    max_seq_length=4096,
    load_in_4bit=True,
)
FastLanguageModel.for_inference(model)

@app.post("/api/chat")
async def chat(request: ChatRequest):
    prompt = format_chat_prompt(request.messages)
    inputs = tokenizer(prompt, return_tensors="pt").to(model.device)

    with torch.no_grad():
        outputs = model.generate(
            **inputs,
            max_new_tokens=1024,
            temperature=request.temperature or 0.7,
            top_p=0.9,
            do_sample=True,
            repetition_penalty=1.15,
        )

    response = tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
    return {"response": response, "model": "health-advisor-lora"}

Patrones asyncio

asyncio es el framework integrado de Python para código concurrente con I/O. TaskGroup (3.11+) reemplaza a gather para concurrencia estructurada con manejo adecuado de errores. Se combina con asyncio.Queue para patrones productor-consumidor.

import asyncio
from asyncio import TaskGroup, Queue
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

# Structured concurrency with TaskGroup (3.11+)
async def fetch_all_metrics(user_ids: list[int]) -> list[dict]:
    results: list[dict] = []
    async with TaskGroup() as tg:
        for uid in user_ids:
            tg.create_task(fetch_user_metrics(uid, results))
    return results

# Producer-consumer with bounded queue
async def producer(queue: Queue[dict], source: AsyncIterator[dict]) -> None:
    async for item in source:
        await queue.put(item)
    await queue.put(None)  # sentinel

async def consumer(queue: Queue[dict], batch_size: int = 50) -> None:
    batch: list[dict] = []
    while True:
        item = await queue.get()
        if item is None:
            break
        batch.append(item)
        if len(batch) >= batch_size:
            await flush_batch(batch)
            batch.clear()
    if batch:
        await flush_batch(batch)

# Async context manager for resource cleanup
@asynccontextmanager
async def managed_connection(url: str):
    conn = await create_connection(url)
    try:
        yield conn
    finally:
        await conn.close()

aiohttp: Cliente y Servidor HTTP Async

aiohttp proporciona tanto un cliente HTTP async para hacer requests concurrentes como un servidor web async. Su pool de conexiones y reutilización de sesiones lo hacen eficiente para integraciones con APIs y web scraping a escala.

import aiohttp
import asyncio
from aiohttp import ClientTimeout

async def fetch_health_apis(endpoints: list[str]) -> list[dict]:
    """Fetch data from multiple health APIs concurrently."""
    timeout = ClientTimeout(total=30, connect=5)
    results: list[dict] = []

    async with aiohttp.ClientSession(timeout=timeout) as session:
        async def fetch_one(url: str) -> dict:
            async with session.get(url) as resp:
                resp.raise_for_status()
                return await resp.json()

        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(ep)) for ep in endpoints]

        results = [t.result() for t in tasks]
    return results

# Rate-limited concurrent fetcher
async def rate_limited_fetch(
    urls: list[str],
    max_concurrent: int = 10,
    delay: float = 0.1,
) -> list[dict]:
    semaphore = asyncio.Semaphore(max_concurrent)
    async with aiohttp.ClientSession() as session:
        async def fetch(url: str) -> dict:
            async with semaphore:
                await asyncio.sleep(delay)
                async with session.get(url) as resp:
                    return await resp.json()

        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(u)) for u in urls]
        return [t.result() for t in tasks]

pytest: Framework de Testing Moderno

pytest es el framework de testing estándar de Python, ofreciendo fixtures para setup y teardown, parametrize para tests basados en datos, y un rico ecosistema de plugins. Se usa pytest-asyncio para testing de código async y pytest-cov para reportes de cobertura.

import pytest
from httpx import AsyncClient, ASGITransport
from unittest.mock import AsyncMock, patch
from health_api.main import app
from health_api.models import BodyMetrics

# Async fixture for test client
@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

# Parametrized test: multiple inputs, one test function
@pytest.mark.parametrize("weight,expected_bmi", [
    (70.0, 23.1),
    (85.0, 28.1),
    (55.0, 18.2),
])
def test_bmi_calculation(weight: float, expected_bmi: float):
    bmi = calculate_bmi(weight, height_m=1.74)
    assert bmi == pytest.approx(expected_bmi, abs=0.1)

# Async test with mocked dependency
@pytest.mark.asyncio
async def test_create_metrics(client: AsyncClient):
    payload = {"weight_kg": 78.5, "body_fat_pct": 15.2}
    with patch("health_api.deps.get_current_user", return_value=mock_user):
        resp = await client.post("/api/metrics", json=payload)
    assert resp.status_code == 201
    data = resp.json()
    assert data["metrics"]["weight_kg"] == 78.5
    assert data["trend"] in ("up", "down", "stable")

# Fixture with database teardown
@pytest.fixture
async def db_session():
    async with async_session_factory() as session:
        yield session
        await session.rollback()

# Custom marker for slow integration tests
@pytest.mark.slow
@pytest.mark.asyncio
async def test_full_sync_pipeline(db_session):
    """End-to-end test: BLE read, transform, store, verify."""
    raw = generate_mock_ble_data(weight=80.0, impedance=500)
    metrics = parse_ble_data(raw)
    record = await store_metrics(db_session, metrics)
    assert record.id is not None
    assert record.weight_kg == pytest.approx(80.0, abs=0.1)

Cobertura y Configuración de CI

Se usa pytest-cov para medición de cobertura y se aplican umbrales mínimos en CI. Se combina con ruff para linting y mypy para type checking en un solo pipeline de testing.

# pyproject.toml - pytest and coverage configuration
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
markers = [
    "slow: marks tests as slow (deselect with '-m not slow')",
]
addopts = "--strict-markers -ra --cov=health_api --cov-report=term-missing"

[tool.coverage.run]
source = ["health_api"]
branch = true
omit = ["*/tests/*", "*/migrations/*"]

[tool.coverage.report]
fail_under = 85
show_missing = true
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "if __name__ == .__main__.",
]

# CI pipeline (GitHub Actions snippet)
# - run: |
#     poetry run ruff check .
#     poetry run mypy .
#     poetry run pytest --cov-fail-under=85

BeautifulSoup: Parsing Rápido

BeautifulSoup combinado con requests o aiohttp es el enfoque más simple para scraping de HTML estructurado. Sobresale en scripts puntuales y crawls pequeños donde un framework completo sería excesivo.

import requests
from bs4 import BeautifulSoup
from dataclasses import dataclass

@dataclass
class NutritionInfo:
    name: str
    calories: int
    protein_g: float
    carbs_g: float
    fat_g: float

def scrape_nutrition(food: str) -> NutritionInfo | None:
    """Scrape nutritional data for a food item."""
    resp = requests.get(
        "https://example.com/nutrition/search",
        params={"q": food},
        headers={"User-Agent": "HealthBot/1.0"},
        timeout=10,
    )
    resp.raise_for_status()

    soup = BeautifulSoup(resp.text, "html.parser")
    result = soup.select_one(".nutrition-result")
    if not result:
        return None

    return NutritionInfo(
        name=result.select_one("h2").get_text(strip=True),
        calories=int(result.select_one("[data-nutrient='calories']").text),
        protein_g=float(result.select_one("[data-nutrient='protein']").text),
        carbs_g=float(result.select_one("[data-nutrient='carbs']").text),
        fat_g=float(result.select_one("[data-nutrient='fat']").text),
    )

Scrapy: Crawling de Nivel Industrial

Scrapy es un framework de crawling completo con scheduling de requests integrado, pipelines de middleware, auto-throttling y exportación a JSON/CSV/bases de datos. Se usa para crawls a gran escala y multi-página con extracción de datos estructurados.

import scrapy
from scrapy.loader import ItemLoader
from itemloaders.processors import TakeFirst, MapCompose

class ExerciseItem(scrapy.Item):
    name = scrapy.Field()
    muscle_group = scrapy.Field()
    difficulty = scrapy.Field()
    description = scrapy.Field()

class ExerciseSpider(scrapy.Spider):
    name = "exercises"
    start_urls = ["https://example.com/exercises/"]
    custom_settings = {
        "CONCURRENT_REQUESTS": 8,
        "DOWNLOAD_DELAY": 0.5,
        "AUTOTHROTTLE_ENABLED": True,
        "FEEDS": {"exercises.json": {"format": "json", "overwrite": True}},
    }

    def parse(self, response):
        for card in response.css(".exercise-card"):
            loader = ItemLoader(item=ExerciseItem(), selector=card)
            loader.default_output_processor = TakeFirst()
            loader.add_css("name", "h3::text")
            loader.add_css("muscle_group", ".muscle-tag::text")
            loader.add_css("difficulty", ".difficulty::attr(data-level)")
            loader.add_css("description", ".desc::text")
            yield loader.load_item()

        next_page = response.css("a.next-page::attr(href)").get()
        if next_page:
            yield response.follow(next_page, self.parse)

Poetry: Gestión Moderna de Dependencias

Poetry proporciona builds deterministas con lockfile, gestión de entornos virtuales, resolución de dependencias y publicación de paquetes. El archivo pyproject.toml es la fuente única de verdad para la configuración del proyecto.

# pyproject.toml
[tool.poetry]
name = "health-api"
version = "2.1.0"
description = "Health metrics API and data pipeline"
authors = ["Jose Nobile <[email protected]>"]
python = "^3.12"

[tool.poetry.dependencies]
python = "^3.12"
fastapi = "^0.115.0"
uvicorn = {extras = ["standard"], version = "^0.34.0"}
sqlalchemy = {extras = ["asyncio"], version = "^2.0.36"}
pydantic = "^2.10.0"
pandas = "^2.2.0"
numpy = "^2.1.0"

[tool.poetry.group.dev.dependencies]
pytest = "^8.3.0"
pytest-asyncio = "^0.24.0"
pytest-cov = "^6.0.0"
mypy = "^1.13.0"
pyright = "^1.1.390"
ruff = "^0.8.0"

[tool.poetry.group.ml.dependencies]
torch = "^2.5.0"
unsloth = {extras = ["colab-new"], version = "^2024.12"}
transformers = "^4.46.0"
datasets = "^3.1.0"
trl = "^0.12.0"

[tool.poetry.scripts]
api = "health_api.main:start"
sync = "health_api.sync:main"

# Commands
# poetry install              -> install all deps
# poetry install --with ml    -> include ML deps
# poetry lock                 -> regenerate lockfile
# poetry run api              -> run the API
# poetry build                -> build package

uv: Gestor de Paquetes Rápido para Python

uv (v0.11.32 a julio 2026) es un instalador y resolvedor de paquetes Python basado en Rust que es 10-100x más rápido que pip. Soporta pyproject.toml, lockfiles, entornos virtuales y gestión de versiones de Python en una sola herramienta. Se ha convertido en el estándar de facto para nuevos proyectos Python en 2026.

# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create project with uv
uv init health-api
cd health-api

# Add dependencies (resolves and locks automatically)
uv add fastapi uvicorn[standard] sqlalchemy[asyncio] pydantic
uv add --dev pytest pytest-asyncio mypy ruff

# Sync environment from lockfile (fast, deterministic)
uv sync

# Run commands inside the managed environment
uv run uvicorn health_api.main:app --reload
uv run pytest --cov

# Pin Python version for the project
uv python pin 3.12

# Compile requirements.txt from pyproject.toml (for Docker)
uv pip compile pyproject.toml -o requirements.txt

# Install into current venv (pip-compatible mode)
uv pip install -r requirements.txt

Entornos Virtuales y pip

Para proyectos más simples, venv + pip + requirements.txt sigue siendo el enfoque más portable. Se fijan versiones exactas con pip freeze y se separan dependencias de dev de producción.

# Create and activate virtual environment
python3.12 -m venv .venv
source .venv/bin/activate

# Install with constraints
pip install -r requirements.txt -c constraints.txt

# requirements.txt (pinned)
fastapi==0.115.6
uvicorn[standard]==0.34.0
pydantic==2.10.3
sqlalchemy[asyncio]==2.0.36

# requirements-dev.txt
-r requirements.txt
pytest==8.3.4
pytest-cov==6.0.0
mypy==1.13.0
ruff==0.8.3

# Dockerfile multi-stage build
FROM python:3.12-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

FROM python:3.12-slim
COPY --from=builder /install /usr/local
COPY . /app
WORKDIR /app
CMD ["uvicorn", "health_api.main:app", "--host", "0.0.0.0", "--port", "8000"]

Anotaciones de Tipos Avanzadas

El sistema de tipos de Python soporta genéricos, protocolos (structural typing), tipos literales, overloads y TypeGuard. Las anotaciones de tipos son validadas por mypy y pyright en tiempo de desarrollo, capturando bugs antes del runtime.

from typing import Protocol, TypeVar, Generic, overload, Literal
from collections.abc import Sequence, Callable, AsyncIterator

T = TypeVar("T")
T_co = TypeVar("T_co", covariant=True)

# Protocol (structural typing - like Go interfaces)
class Measurable(Protocol):
    @property
    def value(self) -> float: ...
    @property
    def unit(self) -> str: ...
    def to_base_unit(self) -> float: ...

# Generic repository
class Repository(Generic[T]):
    async def get(self, id: int) -> T | None: ...
    async def list(self, limit: int = 100, offset: int = 0) -> Sequence[T]: ...
    async def create(self, entity: T) -> T: ...
    async def update(self, id: int, entity: T) -> T: ...
    async def delete(self, id: int) -> bool: ...

# Overloaded function signatures
@overload
def parse_metric(raw: str, kind: Literal["weight"]) -> WeightMetric: ...
@overload
def parse_metric(raw: str, kind: Literal["body_fat"]) -> BodyFatMetric: ...
def parse_metric(raw: str, kind: str) -> WeightMetric | BodyFatMetric:
    if kind == "weight":
        return WeightMetric.from_raw(raw)
    return BodyFatMetric.from_raw(raw)

# Callback types with ParamSpec
from typing import ParamSpec
P = ParamSpec("P")

def retry(max_attempts: int = 3) -> Callable[[Callable[P, T]], Callable[P, T]]:
    def decorator(func: Callable[P, T]) -> Callable[P, T]:
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception:
                    if attempt == max_attempts - 1:
                        raise
            raise RuntimeError("unreachable")
        return wrapper
    return decorator

Configuración de mypy y pyright

La configuración estricta de mypy y pyright captura errores de tipos temprano. Se integran ambos en CI/CD para prevenir regresiones de tipos. Los overrides por módulo manejan bibliotecas de terceros sin type stubs.

# pyproject.toml - mypy configuration
[tool.mypy]
python_version = "3.12"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_any_generics = true
check_untyped_defs = true
no_implicit_reexport = true
warn_redundant_casts = true
warn_unused_ignores = true
show_error_codes = true
pretty = true

[[tool.mypy.overrides]]
module = ["unsloth.*", "trl.*", "bleak.*"]
ignore_missing_imports = true

# pyright configuration
[tool.pyright]
pythonVersion = "3.12"
typeCheckingMode = "strict"
reportMissingTypeStubs = "warning"
reportUnusedImport = "error"
reportUnusedVariable = "error"

# ruff configuration (linter + formatter)
[tool.ruff]
target-version = "py312"
line-length = 100

[tool.ruff.lint]
select = ["E", "F", "I", "N", "W", "UP", "B", "A", "SIM", "TCH"]
ignore = ["E501"]

Python 3.13 (Octubre 2024)

Python 3.13 es una versión histórica con dos características experimentales que señalan el futuro del lenguaje: un build free-threading (python3.13t) que desactiva el GIL para paralelismo multi-core real, y un compilador JIT experimental para mejor rendimiento en runtime. También incluye un REPL interactivo mejorado con edición multilínea y syntax highlighting, y soporte inicial Tier 3 para plataformas iOS y Android.python3.13t) that disables the GIL for true multi-core parallelism, and an experimental JIT compiler for improved runtime performance. It also ships an improved interactive REPL with multi-line editing and syntax highlighting, and initial Tier 3 support for iOS and Android platforms.

# Python 3.13: Free-threading experimental build
# Install the free-threaded build: python3.13t
import threading
import time

def cpu_bound(n: int) -> int:
    """CPU-intensive work that benefits from true parallelism."""
    return sum(i * i for i in range(n))

# With GIL (python3.13): threads run one at a time for CPU work
# With free-threading (python3.13t): threads run on separate cores
threads = [threading.Thread(target=cpu_bound, args=(10_000_000,)) for _ in range(4)]
start = time.perf_counter()
for t in threads: t.start()
for t in threads: t.join()
print(f"Elapsed: {time.perf_counter() - start:.2f}s")
# python3.13:  ~4.0s (serial due to GIL)
# python3.13t: ~1.1s (parallel on 4 cores)

Python 3.14 (Octubre 2025)

Python 3.14 introduce template string literals (t-strings, PEP 750), evaluación diferida de anotaciones (PEP 649/749) y mejoras significativas en free-threading. Los t-strings son la característica nueva más visible: lucen como f-strings pero producen objetos Template en vez de strings, habilitando interpolación segura para SQL, HTML y otros contextos donde importa la prevención de inyección. La última versión de mantenimiento es Python 3.14.6 (10 de junio de 2026). Python 3.15 ya está en beta (3.15.0b4, 18 de julio de 2026, la última beta planificada) -- con features congeladas desde la beta 1 el 7 de mayo de 2026, RC1 prevista para el 4 de agosto y la versión final 3.15.0 el 1 de octubre de 2026. Los builds free-threaded ahora están oficialmente soportados en 3.14.

# Python 3.14: Template strings (t-strings, PEP 750)
from string.templatelib import Template

name = "Jose"
greeting: Template = t"Hello, {name}!"
# greeting is a Template object, NOT a string
# Libraries can inspect the structure for safe interpolation

# Use case: SQL injection prevention
user_input = "'; DROP TABLE users; --"
query = t"SELECT * FROM users WHERE name = {user_input}"
# A SQL library processes the Template safely, parameterizing the value

# Deferred annotations (PEP 649/749) - annotations evaluated lazily
from __future__ import annotations  # no longer needed in 3.14

class Tree:
    left: Tree | None   # works without quotes -- evaluated lazily
    right: Tree | None
    value: int

Free-Threading en Profundidad: La Remoción del GIL

La remoción del Global Interpreter Lock (GIL) es el cambio más grande en CPython en décadas. En Python 3.13, el free-threading es experimental con una penalidad de 10-15% en rendimiento single-thread. En Python 3.14, esta penalidad baja a 5-10% gracias a un garbage collector incremental thread-safe y locks por objeto en tipos built-in. Python 3.15 (con features congeladas en mayo de 2026) mantiene el free-threading como un build opt-in separado; el cambio de default sigue en discusión para una versión posterior. Para usarlo: instala el build free-threaded (python3.14t), y asegúrate de que tus extensiones C sean thread-safe. La mayoría del código Python puro funciona sin cambios.python3.14t), and ensure your C extensions are thread-safe. Most pure Python code works unchanged.

Python 3.15 Beta: El Compilador JIT Finalmente Entrega Ganancias de Velocidad

Python 3.15 (ya en beta, con features congeladas desde mayo de 2026 y versión final prevista para el 1 de octubre de 2026) marca la primera vez que el compilador JIT experimental supera consistentemente al intérprete. En macOS AArch64, el JIT es 11-12% más rápido que el intérprete tail-calling; en x86_64 Linux, entrega mejoras de 5-6% sobre el intérprete estándar. El JIT usa una estrategia de compilación copy-and-patch que estuvo enfocada en infraestructura en 3.13-3.14, pero 3.15 se beneficia de una base de contribuidores más amplia y generación de código optimizada. El intérprete adaptativo especializante ahora está completamente habilitado en builds free-threaded, cerrando casi por completo la brecha de rendimiento single-threaded entre los modos GIL y free-threaded.

# Free-threading: real multi-core parallelism in Python 3.14t
import threading
from concurrent.futures import ThreadPoolExecutor

# Per-object locks on built-in types ensure thread safety
shared_dict: dict[str, int] = {}
lock = threading.Lock()  # still recommended for compound operations

def process_batch(batch_id: int, data: list[float]) -> dict:
    """Process a batch of data on a dedicated core."""
    result = heavy_computation(data)  # truly parallel with python3.14t
    with lock:
        shared_dict[f"batch_{batch_id}"] = len(result)
    return result

# ThreadPoolExecutor with free-threading uses real OS threads on separate cores
with ThreadPoolExecutor(max_workers=8) as pool:
    futures = [pool.submit(process_batch, i, chunk) for i, chunk in enumerate(chunks)]
    results = [f.result() for f in futures]

# Check if running free-threaded: import sys; sys._is_gil_enabled()