import logging import re import secrets from contextlib import contextmanager from typing import Any, Generator from constance import config from django.conf import settings from django.core import mail from django.core.cache import cache from django.core.mail.backends.smtp import EmailBackend from django.db import transaction from django.utils.crypto import get_random_string from django.utils.translation import get_language from graphene import Context from rest_framework.request import Request logger = logging.getLogger(__name__) def graphene_current_lang() -> str: """ Determines the currently active language code. This function retrieves the current language from the available language settings. If no language is set, it defaults to the application's default language code. The language code is returned in lowercase. Returns: str: The currently active language code in lowercase. """ return (get_language() or settings.LANGUAGE_CODE).lower() def graphene_abs(request: Request | Context, path_or_url: str) -> str: """ Builds and returns an absolute URI for a given path or URL. Summary: This function takes a relative path or URL and constructs a fully qualified absolute URI using the request object. Args: request: The request object used to build the absolute URI. path_or_url: str The relative path or URL to be converted to an absolute URI. Returns: str: The absolute URI corresponding to the provided path or URL. """ return str(request.build_absolute_uri(path_or_url)) # ty: ignore[possibly-missing-attribute] def get_random_code() -> str: """ Generates a random alphanumeric string of a fixed length. This function uses a utility function to generate a random string consisting of letters and digits. The length of the string is fixed to 20 characters. The generated string can be used for purposes such as unique identifiers or tokens. Returns: str: Randomly generated alphanumeric string of length 20. """ return get_random_string(20) def get_product_uuid_as_path(instance, filename: str = "") -> str: return "products" + "/" + str(instance.product.uuid) + "/" + filename def get_vendor_name_as_path(instance, filename: str = "") -> str: return "vendors_responses/" + str(instance.name) + "/" + filename def get_brand_name_as_path(instance, filename: str = "") -> str: return "brands/" + str(instance.name) + "/" + filename @contextmanager def atomic_if_not_debug() -> Generator[None, None, None]: """ Context manager to wrap a block of code in an atomic transaction if the DEBUG setting is not enabled. This context manager ensures that the code block executes within an atomic database transaction, preventing partial updates to the database in case of an exception. If the DEBUG setting is enabled, no transaction is enforced, allowing for easier debugging. """ if not settings.DEBUG: with transaction.atomic(): yield else: yield def is_url_safe(url: str) -> bool: return bool(re.match(r"^https://", url, re.IGNORECASE)) def format_attributes(attributes: str | None = None) -> dict[str, Any]: """ Parses a string of attributes into a dictionary. This function takes a string input representing attributes and their values, formatted as `key=value` pairs separated by commas, and converts it into a dictionary. It returns an empty dictionary if the input is `None` or invalid. Invalid key-value pairs within the input string are skipped. """ if not attributes: return {} try: attribute_pairs = attributes.split(",") except AttributeError: return {} result = {} for attr_pair in attribute_pairs: try: key, value = attr_pair.split("=") result[key] = value except ValueError: continue return result def get_project_parameters() -> Any: """ Fetches project parameters from cache or configuration. This function retrieves project parameters from a cache if available. If they are not cached, it collects the parameters from a designated configuration source, formats their keys to lowercase, and then stores them in the cache for a limited period. """ parameters = cache.get("parameters", {}) if not parameters: for key in settings.EXPOSABLE_KEYS: if not getattr(config, key, ""): continue parameters[key.lower()] = getattr(config, key) cache.set("parameters", parameters, 60 * 60) return parameters def resolve_translations_for_elasticsearch(instance, field_name: str) -> None: """ Resolves translations for a given field in an Elasticsearch-compatible format. It checks if the localized version of the field contains data, and if not, sets it to the value of the default field. """ field = getattr(instance, f"{field_name}_{settings.LANGUAGE_CODE}", "") filled_field = getattr(instance, field_name, "") if not field: setattr(instance, f"{field_name}_{settings.LANGUAGE_CODE}", filled_field) CROCKFORD = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" def generate_human_readable_id(length: int = 6) -> str: """ Generates a human-readable identifier using Crockford's Base32 characters. This function creates a string of a specified length composed of randomly selected characters from the Crockford Base32 alphabet. A dash is inserted at a random or mid-point position in the identifier for better readability. Parameters: length (int): The length of the identifier excluding the dash. Must be greater than 0. Default is 6. Returns: str: A human-readable identifier with the specified length plus a dash. """ chars = [secrets.choice(CROCKFORD) for _ in range(length)] pos = ( (secrets.randbelow(length - 1) + 1) if secrets.choice([True, False]) else (length // 2) ) chars.insert(pos, "-") return "".join(chars) def generate_human_readable_token() -> str: """ Generates a human-readable token. This function creates a random token using characters from the CROCKFORD base32 set. The generated token is 20 characters long and is designed to be human-readable. Returns: str: A 20-character random token. """ return "".join([secrets.choice(CROCKFORD) for _ in range(20)]) def is_status_code_success(status_code: int) -> bool: return 200 <= status_code < 300 def get_dynamic_email_connection() -> EmailBackend: return mail.get_connection( host=config.EMAIL_HOST, port=config.EMAIL_PORT, username=config.EMAIL_HOST_USER, password=config.EMAIL_HOST_PASSWORD, use_tls=config.EMAIL_USE_TLS, use_ssl=config.EMAIL_USE_SSL, timeout=30, fail_silently=False, backend=config.EMAIL_BACKEND, )