Source code for vstutils.middleware

import time
import logging
from typing import Generator
from contextlib import contextmanager

from django.db import connections
from django.apps import apps
from django.conf import settings
from django.utils import translation
from django.urls import resolve
from django.shortcuts import redirect

from .api.models import Language
from .utils import BaseVstObject

logger = logging.getLogger(settings.VST_PROJECT)

def wrap_connections(wrapper):
    for connection in connections.all():
        yield wrapper
        for connection in connections.all():

class QueryTimingLogger:
    __slots__ = ('queries_time',)

    def __init__(self):
        self.queries_time = 0

    def __call__(self, execute, sql, params, many, context):
        start = time.monotonic()
            return execute(sql, params, many, context)
            self.queries_time += time.monotonic() - start

[docs]class BaseMiddleware(BaseVstObject): """ Middleware base class for handling: * Incoming requests by :meth:`.BaseMiddleware.request_handler()`; * Outgoing response before any calling on server by :meth:`.BaseMiddleware.get_response_handler()`; * Outgoing responses by :meth:`.BaseMiddleware.handler()`. Middleware must be added to `MIDDLEWARE` list in settings. Example: .. sourcecode:: python from vstutils.middleware import BaseMiddleware from django.http import HttpResponse class CustomMiddleware(BaseMiddleware): def request_handler(self, request): # Add header to request request.headers['User-Agent'] = 'Mozilla/5.0' return request def get_response_handler(self, request): if not request.user.is_stuff: # Return 403 HTTP status for non-stuff users. # This request never gets in any view. return HttpResponse( "Access denied!", content_type="text/plain", status_code=403 ) return super().get_response_handler(request) def handler(self, request, response): # Add header to response response['Custom-Header'] = 'Some value' return response """ __slots__ = 'get_response', 'logger' def __init__(self, get_response): self.get_response = get_response self.logger = logger super().__init__() def get_setting(self, value): return self.get_django_settings(value)
[docs] def handler(self, request, response): # nocv # pylint: disable=unused-argument """ The response handler. Method to process responses. :param request: HTTP-request object. :type request: django.http.HttpRequest :param response: HTTP-response object which will be sended to client. :type response: django.http.HttpResponse :return: Handled response object. :rtype: django.http.HttpResponse """ return response
[docs] def request_handler(self, request): # pylint: disable=unused-argument """ The request handler. Called before request is handled by a view. :param request: HTTP-request object which is wrapped from client request. :type request: django.http.HttpRequest :return: Handled request object. :rtype: django.http.HttpRequest """ return request
[docs] def get_response_handler(self, request): """ Entrypoint for breaking or continuing request handling. This function must return `django.http.HttpResponse` object or result of parent class calling. Since the release of 5.3, it has been possible to write this method as asynchronous. This should be used in cases where the middleware makes queries to the database or cache. However, such a middleware should be excluded from bulk requests. .. warning:: Never do asynchronous middleware in dependent chains. They are designed to send independent requests to external sources. Set ``async_capable`` to ``True`` and ``sync_capable`` to ``False`` for such middleware. :param request: HTTP-request object which is wrapped from client request. :type request: django.http.HttpRequest :rtype: django.http.HttpResponse """ return self.get_response(request)
def __call__(self, request): return self.handler( self.request_handler(request), self.get_response_handler(request) )
class TimezoneHeadersMiddleware(BaseMiddleware): __slots__ = () def handler(self, request, response): response['Server-Timezone'] = self.get_setting('TIME_ZONE') response['VSTutils-Version'] = self.get_setting('VSTUTILS_VERSION') return response class ExecuteTimeHeadersMiddleware(BaseMiddleware): __slots__ = () def __duration_handler(self, data): key, value = data if isinstance(value, (list, tuple, map, filter, Generator)): value = ''.join((self.__duration_handler(('', v)) for v in value)) elif isinstance(value, (int, float)): value = f';dur={float(value)}' elif isinstance(value, str) and value: if ' ' in value: value = f'"{value}"' value = f';desc={value}' elif not value: value = '' return f'{key}{value}' def _round_time(self, seconds): return round(seconds * 1000, 2) def get_response_handler(self, request): start_time = time.time() get_response_handler = super().get_response_handler ql = QueryTimingLogger() if not getattr(request, 'is_bulk', False): with wrap_connections(ql): response = get_response_handler(request) else: response = get_response_handler(request) response_durations = getattr(response, 'timings', None) total_time = self._round_time(time.time() - start_time) if getattr(request, 'is_bulk', False): response['Response-Time'] = str(total_time) else: if response_durations: response_durations = f', {", ".join(map(self.__duration_handler, response_durations.items()))}' else: response_durations = "" response_durations += f', db_execution_time;dur={self._round_time(ql.queries_time)}' response['Server-Timing'] = f'total_app;dur={total_time}{response_durations or ""}' return response class LangMiddleware(BaseMiddleware): __slots__ = () def get_lang_object(self, request): set_cookie = True if 'lang' in request.GET: code = request.GET['lang'] elif settings.LANGUAGE_COOKIE_NAME in request.COOKIES: code = request.COOKIES[settings.LANGUAGE_COOKIE_NAME] set_cookie = False else: code = translation.get_language_from_request(request) obj = Language.objects.filter(code=code).first() if obj is not None: return obj, set_cookie return Language.objects.get(code=settings.LANGUAGE_CODE), set_cookie # nocv def get_response_handler(self, request): request.language, set_cookie = self.get_lang_object(request) # type: ignore translation.activate(request.language.code) # type: ignore request.LANGUAGE_CODE = translation.get_language() response = super().get_response_handler(request) if set_cookie: response.set_cookie('lang', request.language.code, domain=settings.SESSION_COOKIE_DOMAIN) # type: ignore if 'Content-Language' not in response: response['Content-Language'] = request.language.code # type: ignore return response class TwoFaMiddleware(BaseMiddleware): redirect_name = 'login' pass_names = ( 'login', 'logout', 'pwa_manifest', 'service_worker', 'offline_gui', 'user_registration', 'terms', ) def request_handler(self, request): request.user.need_twofa = False # type: ignore if request.user.is_authenticated: twofa = getattr(request.user, 'twofa', None) if twofa and not request.session.get('2fa', False): request.user.need_twofa = True # type: ignore return request def check_url_name(self, request): url_name = resolve(request.path).url_name return any([ url_name not in self.pass_names, url_name is not None and url_name.startswith('password_reset') ]) def get_response_handler(self, request): if request.user.need_twofa and self.check_url_name(request): # type: ignore return redirect(self.redirect_name) return super().get_response_handler(request) class FrontendChangesNotifications(BaseMiddleware): __slots__ = ('notificator',) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.notificator = apps.get_app_config('vstutils_api').module.notificator_class([]) def request_handler(self, request): if self.notificator.is_usable(): request.notificator = self.notificator return request def handler(self, request, response): if self.notificator.is_usable(): request.notificator.send() return response