Implementación de Contextos en Flask: Análisis del Código Fuente

Parte 1: Mecanismos de Contexto en Flask

1.1 Almacenamiento Local por Hilo con thraeding.local()

Python proporciona threading.local() como solución nativa para el aislamiento de datos entre hilos. Flask implementa su propia versión mejorada.

Sin aislamiento por hilo:

import time
import threading

class Contador:
    def __init__(self):
        self.valor = 0

recurso_compartido = Contador()

def tarea(indice):
    recurso_compartido.valor = indice
    time.sleep(1)
    print(recurso_compartido.valor)

for indice in range(4):
    hilo = threading.Thread(target=tarea, args=(indice,))
    hilo.start()

Todos los hilos imprimen el mismo valor (el último asignado), ya que comparten la misma referencia al objeto.

Con threading.local():

import time
import threading

datos_locales = threading.local()

def tarea(indice):
    datos_locales.dato = indice
    time.sleep(1)
    print(datos_locales.dato)

for indice in range(4):
    hilo = threading.Thread(target=tarea, args=(indice,))
    hilo.start()

Cada hilo obtiene su propio valor porque threading.local() utiliza el iedntificador único del hilo como clave interna.

1.2 Estructura de Pila (Stack)

Una pila opera bajo el principio LIFO (Last In, First Out). En Python se implementa fácilmente con listas:

contenedor = []
contenedor.push(10)  # Agrega al final
ultimo = contenedor.pop()  # Remueve y retorna el último elemento

Flask emplea pilas para gestionar jerarquías de contextos, permitiendo operaciones de apilado y desapilado durante el ciclo de vida de las solicitudes.

1.3 Reflexión en Programación Orientada a Objetos

La personalización de métodos mágicos permite controlar cómo se almacenan y recuperan los atributos:

class AlmacenDatos:
    def __init__(self):
        object.__setattr__(self, '_datos', {})

    def __setattr__(self, clave, valor):
        self._datos[clave] = valor

    def __getattr__(self, clave):
        return self._datos.get(clave)

instancia = AlmacenDatos()
instancia.nombre = "ejemplo"
print(instancia.nombre)  # Output: ejemplo

1.4 Obtención del Identificador de Hilo

import threading
from threading import get_ident

def obtener_id():
    id_hilo = get_ident()
    print(f"ID del hilo actual: {id_hilo}")

for _ in range(3):
    hilo = threading.Thread(target=obtener_id)
    hilo.start()

1.5 Implementación Personalizada de Almacenamiento Local

import threading

class AlmacenLocal:
    def __init__(self):
        object.__setattr__(self, '_contenedor', {})

    def __setattr__(self, clave, valor):
        id_hilo = threading.get_ident()
        if id_hilo not in self._contenedor:
            self._contenedor[id_hilo] = {}
        self._contenedor[id_hilo][clave] = valor

    def __getattr__(self, clave):
        id_hilo = threading.get_ident()
        datos_hilo = self._contenedor.get(id_hilo, {})
        return datos_hilo.get(clave)

almacen = AlmacenDatos()

def tarea(parametro):
    almacen.dato = parametro
    print(almacen.dato)

for i in range(5):
    hilo = threading.Thread(target=tarea, args=(i,))
    hilo.start()

1.6 Versión Mejorada con Acumulación

Esta variante almacena historial de valores usando listas, retornando siempre el más reciente:

import threading

class AlmacenConHistorial:
    def __init__(self):
        object.__setattr__(self, '_contenedor', {})

    def __setattr__(self, clave, valor):
        id_hilo = threading.get_ident()
        if id_hilo not in self._contenedor:
            self._contenedor[id_hilo] = {}
        if clave not in self._contenedor[id_hilo]:
            self._contenedor[id_hilo][clave] = []
        self._contenedor[id_hilo][clave].append(valor)

    def __getattr__(self, clave):
        id_hilo = threading.get_ident()
        datos_hilo = self._contenedor.get(id_hilo, {})
        valores = datos_hilo.get(clave, [])
        return valores[-1] if valores else None

almacen = AlmacenConHistorial()

def tarea(parametro):
    almacen.info = parametro
    print(almacen.info)

for i in range(5):
    hilo = threading.Thread(target=tarea, args=(i,))
    hilo.start()

Cada hilo mantiene su propia lista de valores. Al consultar almacen.info, se retorna el último elemento insertado por ese hilo específico.

1.7 Implementación de Local en el Código Fuente de Flask

from threading import get_ident

class Local:
    __slots__ = ("__almacenamiento__", "__funcion_id__")

    def __init__(self):
        object.__setattr__(self, "__almacenamiento__", {})
        object.__setattr__(self, "__funcion_id__", get_ident)

    def __iter__(self):
        return iter(self.__almacenamiento__.items())

    def __getattr__(self, nombre):
        try:
            return self.__almacenamiento__[self.__funcion_id__()][nombre]
        except KeyError:
            raise AttributeError(nombre)

    def __setattr__(self, nombre, valor):
        id_actual = self.__funcion_id__()
        almacen = self.__almacenamiento__
        if id_actual not in almacen:
            almacen[id_actual] = {}
        almacen[id_actual][nombre] = valor

    def __delattr__(self, nombre):
        try:
            del self.__almacenamiento__[self.__funcion_id__()][nombre]
        except KeyError:
            raise AttributeError(nombre)

1.8 LocalStack: Gestión de Contexto con Pila

class PilaLocal:
    def __init__(self):
        self._almacen = Local()

    def apilar(self, elemento):
        """Inserta un elemento en la cima de la pila"""
        pila = getattr(self._almacen, "datos", None)
        if pila is None:
            self._almacen.datos = pila = []
        pila.append(elemento)
        return pila

    def desapilar(self):
        pila = getattr(self._almacen, "datos", None)
        if pila is None:
            return None
        if len(pila) == 1:
            return pila[-1]
        return pila.pop()

    @property
    def cima(self):
        try:
            return self._almacen.datos[-1]
        except (AttributeError, IndexError):
            return None

mi_pila = PilaLocal()
mi_pila.apilar('primer_elemento')
mi_pila.apilar('segundo_elemento')
print(mi_pila.cima)  # Output: segundo_elemento
mi_pila.desapilar()
print(mi_pila.cima)  # Output: primer_elemento

1.9 Patrón Singleton en Python

El patrón Singleton garantiza que una clase tenga una única instancia. Flask utiliza este concepto internamente.

Implementación con módulo:

# configuracion.py
class Configuracion:
    def obtener_valor(self):
        return "valor_unico"

config_global = Configuracion()

Implementación con __new__:

class Singleton:
    _instancia = None

    def __new__(cls, *args, **kwargs):
        if cls._instancia is None:
            cls._instancia = super().__new__(cls)
        return cls._instancia

a = Singleton()
b = Singleton()
print(a is b)  # True

Implementación con decorador:

from functools import wraps

def singleton(cls):
    instancias = {}

    @wraps(cls)
    def obtener_instancia(*args, **kwargs):
        if cls not in instancias:
            instancias[cls] = cls(*args, **kwargs)
        return instancias[cls]

    return obtener_instancia

@singleton
class MiClase:
    pass

Implementación con metaclase:

class MetaSingleton(type):
    _instancias = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instancias:
            cls._instancias[cls] = super().__call__(*args, **kwargs)
        return cls._instancias[cls]

class Aplicacion(metaclass=MetaSingleton):
    pass

1.10 Uso de LocalStack en el Código Fuente de Flask

Flask mantiene dos instancias de LocalStack para gestionar contextos:

# Estructura interna de _pila_ctx_solicitud
__almacenamiento__ = {
    1111: {'pila': [CtxSolicitud(solicitud, sesion)]},
    1234: {'pila': [CtxSolicitud(solicitud, sesion)]},
}
_pila_ctx_solicitud = LocalStack()

# Estructura interna de _pila_ctx_aplicacion
__almacenamiento__ = {
    1111: {'pila': [CtxAplicacion(aplicacion, g)]},
    1234: {'pila': [CtxAplicacion(aplicacion, g)]},
}
_pila_ctx_aplicacion = LocalStack()

  • _pila_ctx_solicitud: Administra el contexto de cada solicitud HTTP, conteniendo request y session.
  • _pila_ctx_aplicacion: Administra el contexto de aplicación, conteniendo app y g.

Parte 2: Aálisis del Flujo del Código Fuente

2.1 Inicialización de la Aplicación

Al crear una instancia de Flask, ocurren varios procesos internos:

mi_app = Flask(__name__)

La inicialización incluye:

  • Configuración de rutas estáticas y carpetas de plantillas
  • Creación del diccionario vistas_funciones para mapeo de endpoints
  • Instanciación del mapa de URL (url_map) basado en Werkzeug
  • Registro de la ruta para archivos estáticos
class Flask:
    regla_url_clase = Rule
    mapa_url_clase = Map

    def __init__(self, ...):
        self.ruta_archivos_estaticos
        self.carpeta_estatica
        self.carpeta_plantillas
        self.vistas_funciones = {}
        self.mapa_url = self.mapa_url_clase()

Configuración:

mi_app.config.from_object('configuracion.modulo')

Este proceso lee todos los pares clave-valor del archivo de configuración y los almacena en el objeto Config (un diccionario), que luego se asigna a mi_app.config.

Registro de rutas:

@mi_app.route('/inicio')
def vista_inicio():
    return 'Bienvenido'

Internamente se ejecuta add_url_rule():

  1. Crea un objeto Rule con la URL, métodos HTTP y endpoint
  2. Agrega la regla al mapa_url de la aplicación
  3. Registra la relación endpoint → función en vistas_funciones

2.2 Flujo de una Solicitud Entrante

Cuando llega una solicitud HTTP, Flask ejecuta estos pasos:

  1. Creación de contextos:
    • Instancia CtxSolicitud con el objeto Request y datos de session
    • Instancia CtxAplicacion con la App y el objeto g
  2. Apilado en Local: Ambos contextos se insertan en sus respectivas pilas dentro del almacenamiento local por hilo: ``` { id_hilo: {"pila": [ctx_solicitud]} } { id_hilo: {"pila": [ctx_aplicacion]} }
  3. Ejecución del ciclo:
    • Ejecución de funciones before_request
    • Ejecución de la función de vista
    • Ejecución de funciones after_request
    • Encriptación de session en cookies
  4. Destrucción: Los contextos se eliminan de las pilas al finalizar la respuesta

2.3 LocalProxy: Acceso Transparente a los Contextos

LocalProxy es una clase proxy de Werkzeug que permite acceso dinámico y seguro a los datos del contexto actual:

import functools

class ProxyLocal:
    def __init__(self, almacen_local):
        object.__setattr__(self, "_referencia", almacen_local)

    def __setitem__(self, clave, valor):
        self._obtener_objeto_actual()[clave] = valor

    def __getattr__(self, nombre):
        return getattr(self._obtener_objeto_actual(), nombre)

    def _obtener_objeto_actual(self):
        return self._referencia()


def _buscar_en_solicitud(nombre):
    tope = _pila_ctx_solicitud.cima
    if tope is None:
        raise RuntimeError("Fuera del contexto de solicitud")
    return getattr(tope, nombre)

sesion = ProxyLocal(functools.partial(_buscar_en_solicitud, "sesion"))
solicitud = ProxyLocal(functools.partial(_buscar_en_solicitud, "solicitud"))

En Flask 3.x, la implementación utiliza ContextVar de Python:

from werkzeug.local import LocalProxy
from contextvars import ContextVar

var_ctx_aplicacion = ContextVar("flask.app_ctx")
var_ctx_solicitud = ContextVar("flask.request_ctx")

aplicacion_actual = LocalProxy(var_ctx_aplicacion, "app")
g = LocalProxy(var_ctx_aplicacion, "g")
solicitud = LocalProxy(var_ctx_solicitud, "request")
sesion = LocalProxy(var_ctx_solicitud, "session")

2.4 El Objeto g: Variables Globales por Solicitud

g almacena datos que persisten durante todo el ciclo de una solicitud HTTP. Es ideal para compartir información entre funciones decoradas con before_request y las vistas:

from flask import Flask, g

mi_app = Flask(__name__)

@mi_app.before_request
def preparar_datos():
    g.usuario_actual = obtener_usuario()
    g.config_extra = {"modo": "produccion"}

@mi_app.route('/perfil')
def mostrar_perfil():
    nombre = g.usuario_actual.nombre
    return f'Perfil de {nombre}'

@mi_app.route('/dashboard')
def panel_control():
    modo = g.config_extra["modo"]
    return f'Panel en modo {modo}'

if __name__ == '__main__':
    mi_app.run()

Durante la ejecución de g.usuario_actual, Flask internamente consulta g._obtener_objeto_actual(), que retorna el contexto de aplicación actual y accede al atributo usuario_actual.


Resumen del Flujo

Fase de arranque: Se cargan decoradores especiales, rutas y configuración, encapsulándolos en el objeto Flask.

Fase de solicitud:

  1. Se crean los objetos de contexto (aplicación y solicitud)
  2. Se apilan en el almacenamiento local por hilo
  3. Se ejecutan los hooks before_request
  4. Se ejecuta la función de vista
  5. Se ejecutan los hooks after_request
  6. Se destruyen los objetos de contexto

Etiquetas: Flask Python threading-local context-vars werkzeug

Publicado el 7-1 17:14