schon/core/views.py
Egor fureunoir Gorbunov 856f2ff516 Features: 1) Add app_name attribute in multiple urls.py files across apps to support namespacing;
Fixes: 1) Simplify Prometheus and GraphQL path definitions in `evibes/api_urls.py`;

Extra: 1) Add line breaks across multiple files for improved code readability.
2025-06-29 20:03:33 +03:00

543 lines
19 KiB
Python

import mimetypes
import os
import requests
from django.contrib.sitemaps.views import index as _sitemap_index_view
from django.contrib.sitemaps.views import sitemap as _sitemap_detail_view
from django.core.cache import cache
from django.core.exceptions import BadRequest
from django.http import FileResponse, Http404, JsonResponse
from django.shortcuts import redirect
from django.utils.encoding import force_str
from django.utils.http import urlsafe_base64_decode
from django.utils.translation import gettext_lazy as _
from django_ratelimit.decorators import ratelimit
from djangorestframework_camel_case.render import CamelCaseJSONRenderer
from djangorestframework_camel_case.util import camelize
from drf_spectacular.utils import extend_schema_view
from drf_spectacular.views import SpectacularRedocView, SpectacularSwaggerView
from graphene_file_upload.django import FileUploadGraphQLView
from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.renderers import MultiPartRenderer
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework_xml.renderers import XMLRenderer
from rest_framework_yaml.renderers import YAMLRenderer
from sentry_sdk import capture_exception
from core.docs.drf.views import (
BUY_AS_BUSINESS_SCHEMA,
CACHE_SCHEMA,
CONTACT_US_SCHEMA,
LANGUAGE_SCHEMA,
PARAMETERS_SCHEMA,
REQUEST_CURSED_URL_SCHEMA,
SEARCH_SCHEMA,
)
from core.elasticsearch import process_query
from core.models import DigitalAssetDownload, Order
from core.serializers import (
BuyAsBusinessOrderSerializer,
CacheOperatorSerializer,
ContactUsSerializer,
LanguageSerializer,
)
from core.utils import get_project_parameters, is_url_safe
from core.utils.caching import web_cache
from core.utils.emailing import contact_us_email
from core.utils.languages import get_flag_by_language
from evibes import settings
from evibes.settings import LANGUAGES
from payments.serializers import TransactionProcessSerializer
def sitemap_index(request, *args, **kwargs):
"""
Handles the request for the sitemap index and returns an XML response. It ensures the response includes
the appropriate content type header for XML.
Args:
request: The HTTP request object.
*args: Additional positional arguments passed to the view.
**kwargs: Additional keyword arguments passed to the view.
Returns:
A response object containing the sitemap index in XML format, with the proper content type set as
"application/xml; charset=utf-8".
"""
response = _sitemap_index_view(request, *args, **kwargs)
response["Content-Type"] = "application/xml; charset=utf-8"
return response
def sitemap_detail(request, *args, **kwargs):
"""
Handles the detailed view response for a sitemap. This function processes
the request, fetches the appropriate sitemap detail response, and sets the
Content-Type header for XML responses.
Args:
request: An HTTP request object containing request metadata, such as
headers and HTTP method.
*args: Additional positional arguments provided dynamically to the
underlying sitemap detail view function.
**kwargs: Additional keyword arguments provided dynamically to the
underlying sitemap detail view function.
Returns:
HttpResponse: A response object with content representing the requested
sitemap details. The Content-Type header is explicitly set to
"application/xml; charset=utf-8".
"""
response = _sitemap_detail_view(request, *args, **kwargs)
response["Content-Type"] = "application/xml; charset=utf-8"
return response
class CustomGraphQLView(FileUploadGraphQLView):
"""
A custom GraphQL view class that extends the functionality of FileUploadGraphQLView.
This class serves as a customization extension of FileUploadGraphQLView that allows modification
or enhancement of specific behaviors, particularly the context handling for GraphQL requests.
Attributes
----------
None
"""
def get_context(self, request):
return request
class CustomSwaggerView(SpectacularSwaggerView):
"""
CustomSwaggerView is a subclass of SpectacularSwaggerView.
This class overrides the `get_context_data` method to
add extra context to the response. It modifies the context by
including the absolute URI of the current request as the `script_url`.
This can be useful in scenarios where the script or reference
URL needs to be dynamically generated and included in the context.
"""
def get_context_data(self, **kwargs):
# noinspection PyUnresolvedReferences
context = super().get_context_data(**kwargs)
context["script_url"] = self.request.build_absolute_uri()
return context
class CustomRedocView(SpectacularRedocView):
"""
CustomRedocView provides a customized version of the SpectacularRedocView.
This class extends the SpectacularRedocView to include additional
functionality, such as dynamically setting the `script_url` in the
context data. It is designed to be used where customized behavior
for rendering ReDoc UI is required, specifically adapting the script
URL for the current request environment.
"""
def get_context_data(self, **kwargs):
# noinspection PyUnresolvedReferences
context = super().get_context_data(**kwargs)
context["script_url"] = self.request.build_absolute_uri()
return context
@extend_schema_view(**LANGUAGE_SCHEMA)
class SupportedLanguagesView(APIView):
"""
Handles retrieving the list of supported languages.
This class provides an endpoint to return available languages information.
It is configured with relevant serializers, permission classes, and renderers
for flexibility in response formats and access permissions. The endpoint
supports retrieving the list of languages with their respective codes, names,
and flags.
Attributes:
serializer_class (Serializer): Serializer used for formatting the response data.
permission_classes (list): Permissions applied to restrict the endpoint access.
renderer_classes (list): Renderers available for formatting response output.
Methods:
get(self, request): Retrieves the list of supported languages.
"""
serializer_class = LanguageSerializer
permission_classes = [
AllowAny,
]
renderer_classes = [
CamelCaseJSONRenderer,
MultiPartRenderer,
XMLRenderer,
YAMLRenderer,
]
def get(self, request):
return Response(
data=self.serializer_class(
[
{
"code": lang[0],
"name": lang[1],
"flag": get_flag_by_language(lang[0]),
}
for lang in LANGUAGES
],
many=True,
).data,
status=status.HTTP_200_OK,
)
@extend_schema_view(**PARAMETERS_SCHEMA)
class WebsiteParametersView(APIView):
"""
Handles operations related to website parameters.
This class is a Django Rest Framework view that allows clients to retrieve
the parameters of a website. It uses different renderers to present the data
in various formats. The view is publicly accessible.
Attributes
----------
serializer_class
A placeholder for a DRF serializer, it is set to None since no serializer
is explicitly used in this view.
permission_classes
A list indicating the permissions required to access this view. In this case,
`AllowAny`, meaning the view is open to everyone.
renderer_classes
A list of renderers available for this view, supporting CamelCase JSON,
multipart forms, XML, and YAML formats.
Methods
-------
get(request)
Handles HTTP GET requests to fetch website parameters.
"""
serializer_class = None
permission_classes = [
AllowAny,
]
renderer_classes = [
CamelCaseJSONRenderer,
MultiPartRenderer,
XMLRenderer,
YAMLRenderer,
]
def get(self, request):
return Response(data=camelize(get_project_parameters()), status=status.HTTP_200_OK)
@extend_schema_view(**CACHE_SCHEMA)
class CacheOperatorView(APIView):
"""
View for managing cache operations.
This class provides an API view for handling cache operations such as setting cache
data with a specified key and timeout. It leverages multiple renderer classes for
serializing outputs in various formats.
Attributes:
serializer_class (type): Serializer to validate and deserialize input data.
permission_classes (list): List of permission classes to apply access
restrictions.
renderer_classes (list): List of renderer classes to serialize the output
in desired formats.
Methods:
post(request, *args, **kwargs): Handles HTTP POST requests to set cache
data based on the provided key and timeout.
"""
serializer_class = CacheOperatorSerializer
permission_classes = [
AllowAny,
]
renderer_classes = [
CamelCaseJSONRenderer,
MultiPartRenderer,
XMLRenderer,
YAMLRenderer,
]
def post(self, request, *args, **kwargs):
return Response(
data=web_cache(
request,
request.data.get("key"),
request.data.get("data"),
request.data.get("timeout"),
),
status=status.HTTP_200_OK,
)
@extend_schema_view(**CONTACT_US_SCHEMA)
class ContactUsView(APIView):
"""
Handles contact us form submissions via a REST API.
This view processes user submissions for a "Contact Us" form. It validates the received
data using a serializer, applies rate limiting for IP-based requests, and sends emails
asynchronously. The view is prepared to handle multiple response formats using configured
renderers.
Attributes:
serializer_class: The serializer class used to validate incoming data.
renderer_classes: A list of renderers to support multiple response formats.
Methods:
post: Handles POST requests to process form submissions.
"""
serializer_class = ContactUsSerializer
renderer_classes = [
CamelCaseJSONRenderer,
MultiPartRenderer,
XMLRenderer,
YAMLRenderer,
]
@ratelimit(key="ip", rate="2/h")
def post(self, request, *args, **kwargs):
serializer = self.serializer_class(data=request.data)
serializer.is_valid(raise_exception=True)
contact_us_email.delay(serializer.validated_data)
return Response(data=serializer.data, status=status.HTTP_200_OK)
@extend_schema_view(**REQUEST_CURSED_URL_SCHEMA)
class RequestCursedURLView(APIView):
"""
Handles requests for processing and validating URLs from incoming POST requests.
Particularly intended for validating and processing URLs provided by clients. It utilizes rate-limiting, caching, and
various response format renderers to optimize performance and ensure safe handling of external data.
Attributes:
permission_classes (list): Specifies the permissions required to access this view.
renderer_classes (list): Configures the response format renderers available for this view.
Methods:
post: Handles the POST request to validate the URL, fetch its data if valid, and returns the processed response.
"""
permission_classes = [
AllowAny,
]
renderer_classes = [
CamelCaseJSONRenderer,
MultiPartRenderer,
XMLRenderer,
YAMLRenderer,
]
@ratelimit(key="ip", rate="10/h")
def post(self, request, *args, **kwargs):
url = request.data.get("url")
if not is_url_safe(url):
return Response(
data={"error": _("only URLs starting with http(s):// are allowed")},
status=status.HTTP_400_BAD_REQUEST,
)
try:
data = cache.get(url, None)
if not data:
response = requests.get(url, headers={"content-type": "application/json"})
response.raise_for_status()
data = camelize(response.json())
cache.set(url, data, 86400)
return Response(
data=data,
status=status.HTTP_200_OK,
)
except Exception as e:
return Response(
data={"error": str(e)},
status=status.HTTP_400_BAD_REQUEST,
)
@extend_schema_view(**SEARCH_SCHEMA)
class GlobalSearchView(APIView):
"""
Class-based view for handling global search functionality.
This class is designed to process search queries from HTTP GET requests. It is
capable of rendering results in multiple formats including CamelCase JSON,
MultiPart, XML, and YAML. The class uses a custom schema for API documentation
and processes search queries passed as parameters in the request.
Attributes:
renderer_classes (list): List of renderer classes used to serialize responses
into various formats such as CamelCase JSON, MultiPart, XML, and YAML.
Methods:
get: Handles HTTP GET requests by processing the search query and returning
formatted search results.
"""
renderer_classes = [
CamelCaseJSONRenderer,
MultiPartRenderer,
XMLRenderer,
YAMLRenderer,
]
def get(self, request, *args, **kwargs):
return Response(camelize({"results": process_query(query=request.GET.get("q", "").strip(), request=request)}))
@extend_schema_view(**BUY_AS_BUSINESS_SCHEMA)
class BuyAsBusinessView(APIView):
"""
View for buying as a business.
This view handles the logic of creating orders and processing transactions when a
business makes a purchase. It ensures that the request data is properly validated
and processed, handles the creation of the order, and starts the transaction process.
The view also restricts the rate of requests based on the client's IP address
to prevent abuse.
Attributes:
schema (class): Extended schema for API documentation.
Methods:
post(request, *_args, **kwargs):
Handles the post request to process a business purchase.
"""
@ratelimit(key="ip", rate="2/h", block=True)
def post(self, request, *_args, **kwargs):
serializer = BuyAsBusinessOrderSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
order = Order.objects.create(status="MOMENTAL")
products = [product.get("product_uuid") for product in serializer.validated_data.get("products")]
transaction = order.buy_without_registration(
products=products,
promocode_uuid=serializer.validated_data.get("promocode_uuid"),
customer_name=serializer.validated_data.get("customer_name"),
customer_email=serializer.validated_data.get("customer_email"),
customer_phone=serializer.validated_data.get("customer_phone"),
customer_billing_address=serializer.validated_data.get("customer_billing_address_uuid"),
customer_shipping_address=serializer.validated_data.get("customer_shipping_address_uuid"),
payment_method=serializer.validated_data.get("payment_method"),
is_business=True,
)
return Response(
status=status.HTTP_202_ACCEPTED,
data=TransactionProcessSerializer(transaction).data,
)
def download_digital_asset_view(request, *args, **kwargs):
"""
Handles the downloading of a digital asset associated with an order. Ensures that users
are permitted to download the asset only once. Validates the request, retrieves the file,
and serves it as a downloadable response. Returns appropriate error responses for different
failure scenarios.
Args:
request: The HTTP request object containing information about the client request.
*args: Additional positional arguments.
**kwargs: Additional keyword arguments.
Raises:
BadRequest: If the digital asset has already been downloaded.
DigitalAssetDownload.DoesNotExist: If the requested digital asset cannot be found.
Returns:
A FileResponse containing the digital asset file if the request is valid. Returns
a JsonResponse with an error message if an error occurs during the process.
"""
try:
uuid = force_str(urlsafe_base64_decode(kwargs["encoded_uuid"]))
download = DigitalAssetDownload.objects.get(order_product__uuid=uuid)
if download.num_downloads >= 1:
raise BadRequest(_("you can only download the digital asset once"))
download.num_downloads += 1
download.save()
file_path = download.order_product.product.stocks.first().digital_asset.file.path
content_type, encoding = mimetypes.guess_type(file_path)
if not content_type:
content_type = "application/octet-stream"
with open(file_path, "rb") as file:
response = FileResponse(file, content_type=content_type)
filename = os.path.basename(file_path)
response["Content-Disposition"] = f'attachment; filename="{filename}"'
return response
except BadRequest as e:
return JsonResponse({"error": str(e)}, status=400)
except DigitalAssetDownload.DoesNotExist:
return JsonResponse({"error": "Digital asset not found"}, status=404)
except Exception as e:
capture_exception(e)
return JsonResponse(
{"error": "An error occurred while trying to download the digital asset"},
status=500,
)
def favicon_view(request, *args, **kwargs):
"""
Handles requests for the favicon of a website. This function attempts to serve the favicon
file located in the static directory of the project. If the favicon file is not found,
an HTTP 404 error is raised to indicate the resource is unavailable.
Args:
request: The HTTP request object.
*args: Additional positional arguments that are ignored in this function.
**kwargs: Additional keyword arguments that are ignored in this function.
Returns:
FileResponse: A file response containing the favicon image with the content-type
"image/x-icon".
Raises:
Http404: Raised if the favicon file is not found.
"""
try:
favicon_path = os.path.join(settings.BASE_DIR, "static/favicon.png")
return FileResponse(open(favicon_path, "rb"), content_type="image/x-icon")
except FileNotFoundError:
raise Http404(_("favicon not found"))
def index(request, *args, **kwargs):
"""
Redirects the request to the admin index page.
The function handles incoming HTTP requests and redirects them to the Django
admin interface index page. It uses Django's `redirect` function for handling
the HTTP redirection.
Args:
request: The HttpRequest object representing the incoming request.
*args: Additional positional arguments, if any.
**kwargs: Additional keyword arguments, if any.
Returns:
HttpResponseRedirect: An HTTP response that redirects to the admin index.
"""
return redirect("admin:index")