Análisis del código fuente de las tres verificaciones principales de DRF y captura de excepciones

Origen de las tres verificaciones en DRF

En el método dispatch de APIView, antes de distribuir la solicitud según el método HTTP, se ejecuta la función initial. Esta función contiene las tres verificaciones principales y tanto initial como la distribución de la vista están dentro del mismo bloque try para capturar excepciones. Si alguna verificación falla y lanza una excepción, la vista no se ejecutará.

# Método initial en APIView
def initial(self, request, *args, **kwargs):
    # ... código para resolver codificación y control de versiones
    self.perform_authentication(request)
    self.check_permissions(request)
    self.check_throttles(request)

# Método dispatch en APIView
def dispatch(self, request, *args, **kwargs):
    # ...
    try:
        self.initial(request, *args, **kwargs)
        # código de distribución de vista
    except Exception as exc:
        response = self.handle_exception(exc)  # Maneja excepciones de verificaciones y vista
    # ...

Análisis de la verificación de permisos

El método check_permissions itera sobre los objetos de permiso obtenidos de la configuración de la vista o globalmente. Si algún permiso falla, se lanza una excepción.

def check_permissions(self, request):
    # Obtiene lista de objetos de permiso inicializados
    permisos = self.get_permissions()
    for permiso in permisos:
        # Ejecuta el método has_permission de cada permiso
        if not permiso.has_permission(request, self):
            self.permission_denied(
                request,
                message=getattr(permiso, 'message', None),
                code=getattr(permiso, 'code', None)
            )

Puntos clave:

  • La lista permission_classes de la vista se procesa para crear objetos de permiso. Si no se define, se usa la configuración global.
  • Si algún objeto de permiso retorna False en has_permission, se interrumpe el flujo.

Análisis de la verificación de autenticación

El método perform_authentication accede al atributo user del request. Esto dispara la autenticación real.

def perform_authentication(self, request):
    # Se accede al usuario (propiedad computada)
    request.user

El objeto request es una instancia de la clase Request de DRF. La propiedad user ejecuta la autenticación la primera vez.

@property
def user(self):
    if not hasattr(self, '_user'):
        with wrap_attributeerrors():
            self._authenticate()
    return self._user

def _authenticate(self):
    # Itera sobre los autenticadores configurados
    for auth_obj in self.authenticators:
        try:
            credenciales = auth_obj.authenticate(self)
        except exceptions.APIException:
            self._not_authenticated()
            raise

        if credenciales is not None:
            self._authenticator = auth_obj
            self.user, self.auth = credenciales
            return

    self._not_authenticated()

Flujo de autenticación:

  • La propiedad user se calcula una sola vez y se cachea.
  • Los autenticadores se pasan al crear el Request durante el dispatch.
  • El método authenticate debe retornar una tupla (usuario, token) para éxito. None permite probar el siguiente autenticador.

Análisis de la verificación de frecuencia

El método check_throttles verifica los límites de frecuencia usando objetos de throttle.

def check_throttles(self, request):
    duraciones_pendientes = []
    # Obtiene objetos de throttle configurados
    throttles = self.get_throttles()
    for throttle in throttles:
        if not throttle.allow_request(request, self):
            duraciones_pendientes.append(throttle.wait())

    if duraciones_pendientes:
        duraciones = [d for d in duraciones_pendientes if d is not None]
        duracion_max = max(duraciones, default=None)
        self.throttled(request, duracion_max)

Implementación personalizada de allow_request

class ThrottlePersonalizado(BaseThrottle):
    registro_accesos = {}

    def allow_request(self, request, vista):
        ip_cliente = request.META.get('REMOTE_ADDR')
        ahora = time.time()

        if ip_cliente not in self.registro_accesos:
            self.registro_accesos[ip_cliente] = [ahora]
            return True

        historial = self.registro_accesos[ip_cliente]
        # Mantener solo accesos de los últimos 60 segundos
        historial_filtrado = [t for t in historial if ahora - t < 60]
        historial_filtrado.append(ahora)
        self.registro_accesos[ip_cliente] = historial_filtrado

        # Límite de 5 solicitudes por minuto
        return len(historial_filtrado) <= 5

    def wait(self):
        tiempo_primero = self.history[0]
        return 60 - (time.time() - tiempo_primero)

Análisis de SimpleRateThrottle

Esta clase base permite configurar límites mediante una tasa (ej: '5/h').

# En SimpleRateThrottle
def allow_request(self, request, vista):
    if self.rate is None:
        return True

    self.key = self.get_cache_key(request, vista)
    if self.key is None:
        return True

    historial = self.cache.get(self.key, [])
    ahora = self.timer()

    # Eliminar registros fuera del tiempo definido
    while historial and historial[-1] <= ahora - self.duracion:
        historial.pop()

    if len(historial) >= self.num_solicitudes:
        return self.throttle_failure()
    
    return self.throttle_success()

Connfiguración de la tasa:

def __init__(self):
    if not getattr(self, 'rate', None):
        self.rate = self.obtener_tasa()
    self.num_solicitudes, self.duracion = self.parsear_tasa(self.rate)

def obtener_tasa(self):
    if not getattr(self, 'scope', None):
        raise ImproperlyConfigured("Debe definir .scope o .rate")
    return self.TASAS_THROTTLE[self.scope]

def parsear_tasa(self, tasa):
    if tasa is None:
        return (None, None)
    num, periodo = tasa.split('/')
    num_solicitudes = int(num)
    duracion_map = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}
    return (num_solicitudes, duracion_map[periodo[0]])

Manejo de excepciones en DRF

El método handle_exception en APIView captura excepciones de las verificaciones y la vista.

def handle_exception(self, excepcion):
    if isinstance(excepcion, (exceptions.NotAuthenticated,
                              exceptions.AuthenticationFailed)):
        auth_header = self.get_authenticate_header(self.request)
        if auth_header:
            excepcion.auth_header = auth_header
        else:
            excepcion.status_code = status.HTTP_403_FORBIDDEN

    # Obtener manejador de excepciones configurado
    manejador = self.get_exception_handler_context()
    contexto = {
        'vista': self,
        'args': getattr(self, 'args', ()),
        'kwargs': getattr(self, 'kwargs', {}),
        'request': getattr(self, 'request', None)
    }
    respuesta = manejador(excepcion, contexto)
    
    if respuesta is None:
        self.raise_uncaught_exception(excepcion)
    
    respuesta.exception = True
    return respuesta

Personalización del manejador de excepciones

Se puede configurar un manejador personalizado en los ajustes de DRF:

# En settings.py
REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'mi_app.manejo_excepciones.personalizado'
}

# mi_app/manejo_excepciones.py
from rest_framework.views import exception_handler

def personalizado(excepcion, contexto):
    # Registrar información de la excepción
    solicitud = contexto.get('request')
    info_extra = {
        'usuario': getattr(solicitud, 'user', None),
        'metodo': getattr(solicitud, 'method', None),
        'ruta': getattr(solicitud, 'path', None)
    }
    # ... lógica de registro ...
    
    return exception_handler(excepcion, contexto)

Paginación en ViewSets

Ejemplo de paginación manual dentro de una vista:

class LibrosViewSet(viewsets.ViewSet):
    paginador = PaginadorComun()

    def listar(self, request):
        libros = Libro.objects.all()
        libros_paginados = self.paginador.paginate_queryset(libros, request, self)
        serializer = LibroSerializer(instance=libros_paginados, many=True)
        return Response(serializer.data)

Etiquetas: django-rest-framework autenticación permisos throttling excepciones

Publicado el 5-29 16:15