schon/core/utils/__init__.py

import logging
import re
import secrets
from contextlib import contextmanager

from constance import config
from django.conf import settings
from django.core.cache import cache
from django.db import transaction
from django.utils.crypto import get_random_string
from django.utils.translation import get_language

from evibes.settings import DEBUG, EXPOSABLE_KEYS, LANGUAGE_CODE

logger = logging.getLogger("django")


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, 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))


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():
    """
    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 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:
    """
    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() -> dict:
    """
    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 EXPOSABLE_KEYS:
            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}_{LANGUAGE_CODE}", "")
    filled_field = getattr(instance, field_name, "")
    if not field:
        setattr(instance, f"{field_name}_{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