def METHOD_NAME(
    self, user: AbstractUser, job_type_name: str, sync=False, **kwargs
) -> Job:
    """
    Creates a new job and schedule the asynchronous task.
    :param user: The user whom launch the task.
    :param job_type_name: The job type we want to launch.
    :param sync: True if you want to execute the job immediately.
    :return: The newly created job.
    """
    job_type = job_type_registry.get(job_type_name)
    model_class = job_type.model_class
    # Check how many job of same type are running simultaneously. If count > max
    # we don't want to create a new one.
    running_jobs = model_class.objects.filter(
        user_id=user.id
    ).is_pending_or_running()
    if len(running_jobs) >= job_type.max_count:
        raise MaxJobCountExceeded(
            f"You can only launch {job_type.max_count} {job_type_name} job(s) at "
            "the same time."
        )
    job_values = job_type.prepare_values(kwargs, user)
    job = model_class.objects.create(user=user, **job_values)
    job_type.after_job_creation(job, kwargs)
    if sync:
        run_async_job(job.id)
        job.refresh_from_db()
    else:
        # This wrapper ensure the job doesn't stay in pending state if something
        # goes wrong during the delay call. This is related to the redis connection
        # failure that triggers a sys.exit(1) to be called in gunicorn.
        def call_async_job_safe():
            try:
                run_async_job.delay(job.id)
            except BaseException as e:
                job.refresh_from_db()
                if job.state == JOB_PENDING:
                    job.state = JOB_FAILED
                    job.error = str(e)
                    job.human_readable_error = (
                        f"Something went wrong during the job({job.id}) execution."
                    )
                    job.save()
                raise
        transaction.on_commit(call_async_job_safe)
    return job