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_classesde 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
Falseenhas_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
userse calcula una sola vez y se cachea. - Los autenticadores se pasan al crear el
Requestdurante eldispatch. - El método
authenticatedebe retornar una tupla(usuario, token)para éxito.Nonepermite 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)