Refactor queue_name and request_id into properties

This reduces the complexity of the original functions, which will
go up a bit in the next commit.
This commit is contained in:
Ben Thorner
2021-11-10 17:38:31 +00:00
parent 89a8dd1a03
commit 369a9f7521

View File

@@ -24,15 +24,24 @@ def make_task(app):
start = None start = None
typing = False typing = False
@property
def queue_name(self):
delivery_info = self.request.delivery_info or {}
return delivery_info.get('routing_key', 'none')
@property
def request_id(self):
# Note that each header is a direct attribute of the
# task context (aka "request").
return self.request.get('notify_request_id')
def on_success(self, retval, task_id, args, kwargs): def on_success(self, retval, task_id, args, kwargs):
elapsed_time = time.monotonic() - self.start elapsed_time = time.monotonic() - self.start
delivery_info = self.request.delivery_info or {}
queue_name = delivery_info.get('routing_key', 'none')
app.logger.info( app.logger.info(
"Celery task {task_name} (queue: {queue_name}) took {time}".format( "Celery task {task_name} (queue: {queue_name}) took {time}".format(
task_name=self.name, task_name=self.name,
queue_name=queue_name, queue_name=self.queue_name,
time="{0:.4f}".format(elapsed_time) time="{0:.4f}".format(elapsed_time)
) )
) )
@@ -40,25 +49,22 @@ def make_task(app):
app.statsd_client.timing( app.statsd_client.timing(
"celery.{queue_name}.{task_name}.success".format( "celery.{queue_name}.{task_name}.success".format(
task_name=self.name, task_name=self.name,
queue_name=queue_name queue_name=self.queue_name
), elapsed_time ), elapsed_time
) )
def on_failure(self, exc, task_id, args, kwargs, einfo): def on_failure(self, exc, task_id, args, kwargs, einfo):
delivery_info = self.request.delivery_info or {}
queue_name = delivery_info.get('routing_key', 'none')
app.logger.exception( app.logger.exception(
"Celery task {task_name} (queue: {queue_name}) failed".format( "Celery task {task_name} (queue: {queue_name}) failed".format(
task_name=self.name, task_name=self.name,
queue_name=queue_name, queue_name=self.queue_name,
) )
) )
app.statsd_client.incr( app.statsd_client.incr(
"celery.{queue_name}.{task_name}.failure".format( "celery.{queue_name}.{task_name}.failure".format(
task_name=self.name, task_name=self.name,
queue_name=queue_name queue_name=self.queue_name
) )
) )
@@ -71,9 +77,7 @@ def make_task(app):
# TEMPORARY: remove old piggyback values from kwargs # TEMPORARY: remove old piggyback values from kwargs
kwargs.pop('request_id', None) kwargs.pop('request_id', None)
# Add 'request_id' to 'g' so that it gets logged. Note # Add 'request_id' to 'g' so that it gets logged. Note
# that each header is a direct attribute of the task g.request_id = self.request_id
# context (aka "request").
g.request_id = self.request.get('notify_request_id')
return super().__call__(*args, **kwargs) return super().__call__(*args, **kwargs)