import functools import html import os import random import re import time from functools import reduce import factory import markdown2 from bs4 import BeautifulSoup from celery.utils.log import get_task_logger from celery.utils.time import get_exponential_backoff_interval from django.utils.html import urlize logger = get_task_logger(__name__) # Faker that always returns unique values class UniqueFaker(factory.Faker): @classmethod def _get_faker(cls, locale=None): return super()._get_faker(locale).unique # Context manager for tasks that are intended to retry # It will rerun the whole task if exception(s) exc has happened class OkToRetry: def __init__(self, task, exc, num_retries=None, compute_countdown=None, allow_jitter=True): self.task = task self.num_retries = num_retries self.compute_countdown = compute_countdown self.allow_jitter = allow_jitter if not isinstance(exc, (list, tuple)): exc = [exc] self.exc = exc def __enter__(self): pass def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is not None and any(issubclass(exc_type, exc) for exc in self.exc): if self.num_retries is None or self.task.request.retries + 1 <= self.num_retries: countdown = self.get_countdown(exc_val) logger.warning( f"Retrying task gracefully in {countdown} seconds due to {exc_type.__name__}. " f"args: {self.task.request.args}, kwargs: {self.task.request.kwargs}" ) self.rerun_task(countdown) return True def get_countdown(self, exc_val): if self.compute_countdown is not None: countdown = self.compute_countdown(exc_val) if self.allow_jitter is True: countdown = countdown + random.uniform(0, 2) else: countdown = get_exponential_backoff_interval( factor=self.task.retry_backoff, retries=self.task.request.retries, maximum=600, full_jitter=True ) return countdown def rerun_task(self, countdown): self.task.apply_async( self.task.request.args, kwargs=self.task.request.kwargs, retries=self.task.request.retries + 1, countdown=countdown, ) # lru cache version with addition of timeout. # Timeout added to not to occupy memory with too old values def timed_lru_cache(timeout: int, maxsize: int = 128, typed: bool = False): def wrapper_cache(func): func = functools.lru_cache(maxsize=maxsize, typed=typed)(func) func.delta = timeout * 10**9 func.expiration = time.monotonic_ns() + func.delta @functools.wraps(func) def wrapped_func(*args, **kwargs): if time.monotonic_ns() >= func.expiration: func.cache_clear() func.expiration = time.monotonic_ns() + func.delta return func(*args, **kwargs) wrapped_func.cache_info = func.cache_info wrapped_func.cache_clear = func.cache_clear return wrapped_func return wrapper_cache def getenv_boolean(variable_name: str, default: bool) -> bool: value = os.environ.get(variable_name) if value is None: return default return value.lower() in ("true", "1") def batch_queryset(qs, batch_size=1000): qs_count = qs.count() for start in range(0, qs_count, batch_size): end = min(start + batch_size, qs_count) yield qs[start:end] def is_regex_valid(regex) -> bool: try: re.compile(regex) return True except re.error: return False def isoformat_with_tz_suffix(value): """ Default python datetime.isoformat() return tz offset like +00:00 instead of military tz suffix (e.g.Z for UTC)". On the other hand DRF returns datetime with military tz suffix. This utility function exists to return consistent datetime string in api. Is is copied from DRF DateTimeField.to_representation """ value = value.isoformat() if value.endswith("+00:00"): value = value[:-6] + "Z" return value def is_string_with_visible_characters(string): return type(string) == str and not string.isspace() and not string == "" def str_or_backup(string, backup): return string if is_string_with_visible_characters(string) else backup def clean_html(text): text = "".join(BeautifulSoup(text, features="html.parser").find_all(text=True)) return text def convert_slack_md_to_html(text): text = re.sub(r"\*", "**", text) return convert_md_to_html(text) def convert_md_to_html(text): text = markdown2.markdown( text, extras=[ "cuddled-lists", "code-friendly", # Disable _ and __ for em and strong. # This gives us 
 and  tags for ```-fenced blocks
            "fenced-code-blocks",
            "pyshell",
        ],
    ).strip()
    # Special handling cases for lists
    text = text.replace("\n\n