Extend and Customize¶
This package is built to be extended. You can use the Zope Component Architecture to provide specific Adapters to both control how a task is processed and to indicate which processes/logic needs to be executed asynchronously by senaite.queue. The process or logic to be handled by senaite.queue can be from either SENAITE LIMS or from any other SENAITE-specific add-on.
Queued task for a workflow action¶
Let’s imagine you have your own add-on with a custom transition/action (e.g. dispatch) in sample’s workflow, that transitions the sample to a dispatched status. The user can choose multiple samples at once from the listing and transition all them at once. This functionality might entail an undesired impact on performance, specially if hundreds of samples are selected at once.
To address this functionality, we can extend senaite.queue in our own add-on. We are not interested in replacing the logic behind such transition, but feed the queue for this action. Therefore, we can make use of the generic adapter WorkflowGenericQueueAdapter that comes by default with senaite.queue and only do the registration in configure.zcml:
<adapter
name="workflow_action_dispatch"
for="*
zope.publisher.interfaces.browser.IBrowserRequest"
factory="senaite.queue.actions.WorkflowActionGenericQueueAdapter"
provides="bika.lims.interfaces.IWorkflowActionAdapter"
permission="zope.Public" />
This is a named adapter, and the name must be the action id with workflow_action prepended. When the workflow action dispatch is triggered, the system looks for registered adapters and if a match is found, the adapter is called. Note that for field is neither context-specific nor layer specific, so this adapter will always be called when the action dispatch is triggered, regardless of context and layer.
Alternatively, you can directly feed the queue programmatically:
from senaite.queue import api
api.add_action_task(objects, action)
Parameter objects can be either a brain, an object, a uid or a list/tuple of any of them.
Queued task for custom logic¶
Imagine that instead of having a workflow action “dispatch” in place, you rather have a simple view from which the user can choose samples and generate a dispatch pdf from all them at once. Basically you want to feed the queue directly by your own:
class DispatchSamplesView(BrowserView):
def __call__(self):
...
# Get the selected samples from the form
uids = self.request.form.get("selected_uids", [])
# Queue the task
params = {"uids": uids}
api.add_task("my.addon.task_dispatch", self.context, **params)
Note the following:
- We use a “uids” field to store the list of objects to be processed
- We’ve set a custom task id my.addon.task_dispatch. This task id will be used by senaite.queue to look for a suitable adapter able to handle tasks with this id.
Create an adapter in charge of handling the task:
from bika.lims import api as _api
from Products.Archetypes.interfaces.base import IBaseObject
from senaite.queue import api
from senaite.queue.queue import get_chunks_for
from senaite.queue.interfaces import IQueuedTaskAdapter
DISPATCH_TASK_ID = "my.addon.task_dispatch"
class DispatchQueuedTaskAdapter(object):
"""Adapter for dispatch transition
"""
implements(IQueuedTaskAdapter)
adapts(IBaseObject)
def __init__(self, context):
self.context = context
def process(self, task):
"""Process the objects from the task
"""
# If there are too many objects to process, split them in chunks to
# prevent the task to take too much time to complete
chunks = get_chunks_for(task)
# Process the first chunk
objects = map(_api.get_object_by_uid, chunks[0])
map(dispatch_sample, objects)
# Add remaining objects to the queue
params = {"uids": chunks[1]}
api.add_task(DISPATCH_TASK_ID, self.context, **params)
def dispatch_sample(self, sample):
"""Generates a dispatch report for this sample
"""
# Generate the pdf here
pdf = generate_dispatch_pdf(sample)
# Store the pdf as an attachment to the sample
att = _api.create(sample.aq_parent, "Attachment")
att.setAttachmentFile(open(pdf))
sample.setAttachment(att)
Register this adapter in configure.zcml:
<adapter
name="my.addon.task_dispatch"
factory="my.addon.adapters.DispatchQueuedTaskAdapter"
provides="senaite.queue.interfaces.IQueuedTaskAdapter"
for="*" />
Note that this adapter is not only in charge of generating the dispatch pdfs, but also splits the tasks into separate chunks preventing overload.