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
requestysession. - _pila_ctx_aplicacion: Administra el contexto de aplicación, conteniendo
appyg.
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_funcionespara 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():
- Crea un objeto
Rulecon la URL, métodos HTTP y endpoint - Agrega la regla al
mapa_urlde la aplicación - 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:
- Creación de contextos:
- Instancia
CtxSolicitudcon el objetoRequesty datos desession - Instancia
CtxAplicacioncon laAppy el objetog
- Instancia
- 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]}
}
- 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
sessionen cookies
- Ejecución de funciones
- 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:
- Se crean los objetos de contexto (aplicación y solicitud)
- Se apilan en el almacenamiento local por hilo
- Se ejecutan los hooks
before_request - Se ejecuta la función de vista
- Se ejecutan los hooks
after_request - Se destruyen los objetos de contexto