Egor fureunoir Gorbunov
d616d2e18b
Fixes: 1) Removed unused imports including `Order` from `payments/models.py` and `health_check.contrib.redis` from settings; 2) Fixed inconsistent formatting and parameter alignment in method definitions; 3) Corrected type annotations and adjusted verbose text handling; Extra: Cleaned up formatting in docstrings, comments, and JSON field help text across multiple files for better code style compliance and readability.
109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
import logging
|
|
import traceback
|
|
from os import getenv
|
|
|
|
from constance import config
|
|
from django.contrib.auth.models import AnonymousUser
|
|
from django.core.exceptions import BadRequest, DisallowedHost
|
|
from django.http import HttpResponseForbidden
|
|
from django.middleware.common import CommonMiddleware
|
|
from django.middleware.locale import LocaleMiddleware
|
|
from django.shortcuts import redirect
|
|
from django.utils import translation
|
|
from rest_framework_simplejwt.authentication import JWTAuthentication
|
|
from rest_framework_simplejwt.exceptions import InvalidToken
|
|
from sentry_sdk import capture_exception
|
|
|
|
from evibes.settings import DEBUG
|
|
|
|
logger = logging.getLogger("django.request")
|
|
|
|
|
|
class CustomCommonMiddleware(CommonMiddleware):
|
|
def process_request(self, request):
|
|
try:
|
|
return super().process_request(request)
|
|
except DisallowedHost:
|
|
return redirect(f"https://api.{config.BASE_DOMAIN}")
|
|
|
|
|
|
class CustomLocaleMiddleware(LocaleMiddleware):
|
|
def process_request(self, request):
|
|
lang = translation.get_language_from_request(request)
|
|
parts = lang.replace("_", "-").split("-")
|
|
if len(parts) == 2:
|
|
lang_code = parts[0].lower()
|
|
region = parts[1].upper()
|
|
normalized = f"{lang_code}-{region}"
|
|
else:
|
|
normalized = lang
|
|
|
|
translation.activate(normalized)
|
|
request.LANGUAGE_CODE = normalized
|
|
|
|
|
|
# noinspection PyShadowingBuiltins
|
|
class GrapheneJWTAuthorizationMiddleware:
|
|
def resolve(self, next, root, info, **args):
|
|
context = info.context
|
|
|
|
user = self.get_jwt_user(context)
|
|
|
|
info.context.user = user
|
|
|
|
return next(root, info, **args)
|
|
|
|
@staticmethod
|
|
def get_jwt_user(request):
|
|
jwt_authenticator = JWTAuthentication()
|
|
try:
|
|
user, _ = jwt_authenticator.authenticate(request)
|
|
except InvalidToken:
|
|
user = AnonymousUser()
|
|
except TypeError:
|
|
user = AnonymousUser()
|
|
return user
|
|
|
|
|
|
class BlockInvalidHostMiddleware:
|
|
def __init__(self, get_response):
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request):
|
|
allowed_hosts = [
|
|
"app:8000",
|
|
"worker:8000",
|
|
"beat:8000",
|
|
"localhost:8000",
|
|
"api.localhost:8000",
|
|
"b2b.localhost:8000",
|
|
"127.0.0.1:8000",
|
|
"api.127.0.0.1:8000",
|
|
"b2b.127.0.0.1:8000",
|
|
]
|
|
|
|
if DEBUG:
|
|
allowed_hosts += ["*"]
|
|
else:
|
|
allowed_hosts += getenv("ALLOWED_HOSTS").split(" ")
|
|
if not hasattr(request, "META"):
|
|
return BadRequest("Invalid Request")
|
|
if (
|
|
request.META.get("HTTP_HOST") not in allowed_hosts
|
|
and "*" not in allowed_hosts
|
|
):
|
|
return HttpResponseForbidden("Invalid Host Header")
|
|
return self.get_response(request)
|
|
|
|
|
|
# noinspection PyShadowingBuiltins
|
|
class GrapheneLoggingErrorsDebugMiddleware:
|
|
def resolve(self, next, root, info, **args):
|
|
try:
|
|
return next(root, info, **args)
|
|
except Exception as e:
|
|
logger.error("Error occurred in GraphQL execution:", exc_info=True)
|
|
if bool(int(getenv("DEBUG"))):
|
|
logger.error(traceback.format_exc())
|
|
capture_exception(e)
|
|
raise e
|