Schedules a job.

    :param job_id: unique identifier for this job
    :param schedule_in: number of seconds from now in which to schedule the
    job or timedelta object.

    :param **kwargs: parameters to attach to the job, key-value structure.

    >>> schedule_job('http://example.com/test', schedule_in=10, num_retries=10)

def schedule_job(job_id, schedule_in, connection=None, **kwargs):
    """Schedules a job.

    :param job_id: unique identifier for this job
    :param schedule_in: number of seconds from now in which to schedule the
    job or timedelta object.

    :param **kwargs: parameters to attach to the job, key-value structure.

    >>> schedule_job('http://example.com/test', schedule_in=10, num_retries=10)
    """
    if not isinstance(schedule_in, int):  # assumed to be a timedelta
        schedule_in = schedule_in.days * 3600 * 24 + schedule_in.seconds
    schedule_at = int(time.time()) + schedule_in

    if connection is None:
        connection = r

    if 'id' in kwargs:
        raise RuntimeError("'id' is a reserved key for the job ID")

    with connection.pipeline() as pipe:
        if schedule_at is not None:
            args = (schedule_at, job_id)
            if isinstance(connection, redis.Redis):
                # StrictRedis or Redis don't have the same argument order
                args = (job_id, schedule_at)
            pipe.zadd(REDIS_KEY, *args)
        delete = []
        hmset = {}
        for key, value in kwargs.items():
            if value is None:
                delete.append(key)
            else:
                hmset[key] = value
        if hmset:
            pipe.hmset(job_key(job_id), hmset)
        if len(delete) > 0:
            pipe.hdel(job_key(job_id), *delete)
        pipe.execute()