## ## decorators / functions to make the task calls easier ## from django.utils.dateparse import parse_duration from django.db import models from django.utils import timezone from django.apps import apps import re # Regular expression pattern with named groups for "1w2d5h30m10s500ms1000us" without spaces pattern = re.compile( r'(\+|-)?' r'(?:(?P\d+)w)?' r'(?:(?P\d+)d)?' r'(?:(?P\d+)h)?' r'(?:(?P\d+)m)?' r'(?:(?P\d+)s)?' r'(?:(?P\d+)ms)?' r'(?:(?P\d+)us)?' ) def task( *args, **kwargs ): from .models import Task jitter = kwargs.pop('jitter', "") match = pattern.match( jitter ) if jitter and match: kwargs['jitter_length'] = timezone.timedelta( **{ k: int(v) for k, v in match.groupdict().items() if v is not None } ) kwargs['jitter_pivot'] = { "-": "S", None: "M", "+": "E", }[match.group(1)] for f in Task._meta.fields: if not isinstance(f, models.DurationField): continue if f not in kwargs: continue if not kwargs[f]: continue kwargs[f] = parse_duration( kwargs[f] ) return Task( *args, **kwargs ).register def run_on_model_change( *models ): models = [ apps.get_model(m) if isinstance(m, str) else m for m in models ] def decorator( f ): f.watching_models = models return f return decorator from .extender import Extender def extension( *args, **kwargs ): assert Extender.capturing_instance, "Cannot only extend a model during inital import and after apps are ready!" return Extender.capturing_instance( *args, **kwargs ) #