src/sisy/models.py
def task_with_callable(the_callable, label=None, schedule=DEFAULT_SCHEDULE, userdata=None, pk_override=None):
    """Factory function to create a properly initialized task."""
    task = Task()
    if isinstance(the_callable, str):
        if pk_override is not None:
            components = the_callable.split('.')
            info = dict(
                func_type='instancemethod',
                module_name='.'.join(components[:-2]),
                class_name=components[-2],
                class_path='.'.join(components[:-1]),
                model_pk=pk_override,
                func_name=components[-1],
                func_path=the_callable,
            )
            task.funcinfo = info
        else:
            task.funcinfo = get_func_info(func_from_string(the_callable))
    else:
        task.funcinfo = get_func_info(the_callable)

    if label is None:
        task.label = task.funcinfo['func_path']
    else:
        task.label = label

    task.schedule = schedule
    if not croniter.is_valid(task.schedule):
        raise ValueError(f"Cron schedule {task.schedule} is not valid")
    
    if userdata is None:
        task.userdata = dict()
    else:
        if isinstance(userdata, dict):
            task.userdata = userdata
        else:
            raise ValueError("Userdata must be a dictionary of JSON-serializable data")

    return task
def task_with_callable(the_callable, label=None, schedule=DEFAULT_SCHEDULE, userdata=None, pk_override=None):
    task = Task()
    if isinstance(the_callable, str):
        if pk_override is not None:
            components = the_callable.split('.')
            info = dict(
                func_type='instancemethod',
                module_name='.'.join(components[:-2]),
                class_name=components[-2],
                class_path='.'.join(components[:-1]),
                model_pk=pk_override,
                func_name=components[-1],
                func_path=the_callable,
            )
            task.funcinfo = info
        else:
            task.funcinfo = get_func_info(func_from_string(the_callable))
    else:
        task.funcinfo = get_func_info(the_callable)

    if label is None:
        task.label = task.funcinfo['func_path']
    else:
        task.label = label

    task.schedule = schedule
    if not croniter.is_valid(task.schedule):
        raise ValueError(f"Cron schedule {task.schedule} is not valid")
    
    if userdata is None:
        task.userdata = dict()
    else:
        if isinstance(userdata, dict):
            task.userdata = userdata
        else:
            raise ValueError("Userdata must be a dictionary of JSON-serializable data")

    return task
def task_with_callable ( the_callable , label = None , schedule = DEFAULT_SCHEDULE , userdata = None , pk_override = None ) : task = Task ( ) if isinstance ( the_callable , str ) : if pk_override is not None : components = the_callable . split ( '.' ) info = dict ( func_type = 'instancemethod' , module_name = '.' . join ( components [ : - 2 ] ) , class_name = components [ - 2 ] , class_path = '.' . join ( components [ : - 1 ] ) , model_pk = pk_override , func_name = components [ - 1 ] , func_path = the_callable , ) task . funcinfo = info else : task . funcinfo = get_func_info ( func_from_string ( the_callable ) ) else : task . funcinfo = get_func_info ( the_callable ) if label is None : task . label = task . funcinfo [ 'func_path' ] else : task . label = label task . schedule = schedule if not croniter . is_valid ( task . schedule ) : raise ValueError ( f"Cron schedule {task.schedule} is not valid" ) if userdata is None : task . userdata = dict ( ) else : if isinstance ( userdata , dict ) : task . userdata = userdata else : raise ValueError ( "Userdata must be a dictionary of JSON-serializable data" ) return task
Factory function to create a properly initialized task.
Factory function to create a properly initialized task .