add: custom parameters for Callbacks

rq/job.py

@@ -254,6 +254,7 @@ def create(
  on_success = Callback(on_success) # backward compatibility
  job._success_callback_name = on_success.name
  job._success_callback_timeout = on_success.timeout
+ job._success_callback_params = on_success.params
 
  if on_failure:
  if not isinstance(on_failure, Callback):
@@ -264,6 +265,7 @@ def create(
  on_failure = Callback(on_failure) # backward compatibility
  job._failure_callback_name = on_failure.name
  job._failure_callback_timeout = on_failure.timeout
+ job._success_callback_params = on_failure.params
 
  if on_stopped:
  if not isinstance(on_stopped, Callback):
@@ -274,6 +276,7 @@ def create(
  on_stopped = Callback(on_stopped) # backward compatibility
  job._stopped_callback_name = on_stopped.name
  job._stopped_callback_timeout = on_stopped.timeout
+ job._success_callback_params = on_stopped.params
 
  # Extra meta data
  job.description = description or job.get_call_string()
@@ -637,10 +640,13 @@ def __init__(self, id: Optional[str] = None, connection: Optional['Redis'] = Non
  self._args = UNEVALUATED
  self._kwargs = UNEVALUATED
  self._success_callback_name = None
+ self._success_callback_params = None
  self._success_callback = UNEVALUATED
  self._failure_callback_name = None
+ self._failure_callback_params = None
  self._failure_callback = UNEVALUATED
  self._stopped_callback_name = None
+ self._stopped_callback_params = None
  self._stopped_callback = UNEVALUATED
  self.description: Optional[str] = None
  self.origin: str = ''
@@ -942,18 +948,27 @@ def restore(self, raw_data) -> Any:
  if obj.get('success_callback_name'):
  self._success_callback_name = obj.get('success_callback_name').decode()
 
+ if obj.get('success_callback_params'):
+ self._success_callback_params = json.loads(obj.get('success_callback_params').decode())
+
  if 'success_callback_timeout' in obj:
  self._success_callback_timeout = int(obj.get('success_callback_timeout'))
 
  if obj.get('failure_callback_name'):
  self._failure_callback_name = obj.get('failure_callback_name').decode()
 
+ if obj.get('failure_callback_params'):
+ self._failure_callback_params = json.loads(obj.get('failure_callback_params').decode())
+
  if 'failure_callback_timeout' in obj:
  self._failure_callback_timeout = int(obj.get('failure_callback_timeout'))
 
  if obj.get('stopped_callback_name'):
  self._stopped_callback_name = obj.get('stopped_callback_name').decode()
 
+ if obj.get('stopped_callback_params'):
+ self._stopped_callback_params = json.loads(obj.get('stopped_callback_params').decode())
+
  if 'stopped_callback_timeout' in obj:
  self._stopped_callback_timeout = int(obj.get('stopped_callback_timeout'))
 
@@ -1040,10 +1055,16 @@ def to_dict(self, include_meta: bool = True, include_result: bool = True) -> dic
  obj['timeout'] = self.timeout
  if self._success_callback_timeout is not None:
  obj['success_callback_timeout'] = self._success_callback_timeout
+ if self._success_callback_params is not None:
+ obj['success_callback_params'] = json.dumps(self._success_callback_params)
  if self._failure_callback_timeout is not None:
  obj['failure_callback_timeout'] = self._failure_callback_timeout
+ if self._failure_callback_params is not None:
+ obj['failure_callback_params'] = json.dumps(self._failure_callback_params)
  if self._stopped_callback_timeout is not None:
  obj['stopped_callback_timeout'] = self._stopped_callback_timeout
+ if self._stopped_callback_params is not None:
+ obj['stopped_callback_params'] = json.dumps(self._stopped_callback_params)
  if self.result_ttl is not None:
  obj['result_ttl'] = self.result_ttl
  if self.failure_ttl is not None:
@@ -1418,7 +1439,7 @@ def execute_success_callback(self, death_penalty_class: Type[BaseDeathPenalty],
 
  logger.debug('Running success callbacks for %s', self.id)
  with death_penalty_class(self.success_callback_timeout, JobTimeoutException, job_id=self.id):
- self.success_callback(self, self.connection, result)
+ self.success_callback(self, self.connection, result, self._success_callback_params)
 
  def execute_failure_callback(self, death_penalty_class: Type[BaseDeathPenalty], *exc_info):
  """Executes failure_callback with possible timeout"""
@@ -1428,7 +1449,7 @@ def execute_failure_callback(self, death_penalty_class: Type[BaseDeathPenalty],
  logger.debug('Running failure callbacks for %s', self.id)
  try:
  with death_penalty_class(self.failure_callback_timeout, JobTimeoutException, job_id=self.id):
- self.failure_callback(self, self.connection, *exc_info)
+ self.failure_callback(self, self.connection, self._failure_callback_params, *exc_info)
  except Exception: # noqa
  logger.exception(f'Job {self.id}: error while executing failure callback')
  raise
@@ -1438,7 +1459,7 @@ def execute_stopped_callback(self, death_penalty_class: Type[BaseDeathPenalty]):
  logger.debug('Running stopped callbacks for %s', self.id)
  try:
  with death_penalty_class(self.stopped_callback_timeout, JobTimeoutException, job_id=self.id):
- self.stopped_callback(self, self.connection)
+ self.stopped_callback(self, self.connection, self._stopped_callback_params)
  except Exception: # noqa
  logger.exception(f'Job {self.id}: error while executing stopped callback')
  raise
@@ -1645,12 +1666,18 @@ def __init__(self, max: int, interval: Union[int, List[int]] = 0):
 
 
 class Callback:
- def __init__(self, func: Union[str, Callable[..., Any]], timeout: Optional[Any] = None):
+ def __init__(
+ self,
+ func: Union[str, Callable[..., Any]],
+ timeout: Optional[Any] = None,
+ params: Optional[Dict[str, Any]] = None,
+ ):
  if not isinstance(func, str) and not inspect.isfunction(func) and not inspect.isbuiltin(func):
  raise ValueError('Callback `func` must be a string or function')
 
  self.func = func
  self.timeout = parse_timeout(timeout) if timeout else CALLBACK_TIMEOUT
+ self.params = params
 
  @property
  def name(self) -> str:

rq/worker.py

@@ -1472,6 +1472,7 @@ def perform_job(self, job: 'Job', queue: 'Queue') -> bool:
  timeout = job.timeout or self.queue_class.DEFAULT_TIMEOUT
  with self.death_penalty_class(timeout, JobTimeoutException, job_id=job.id):
  self.log.debug('Performing Job...')
+ self.log.debug(f"custom job params: {job._success_callback_params}")
  rv = job.perform()
  self.log.debug('Finished performing Job ID %s', job.id)