bika.lims.workflow.analysisrequest.events.after_restore() - Code Metrics - Inspection of "Sample dispatch workflow" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — 2.x (#1772)

by Ramon

created 2021-02-08 17:16 UTC

after_restore() B

↳ Parent: bika.lims.workflow.analysisrequest.events

Complexity

Conditions

Size

Total Lines	34
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	19
dl	0
loc	34
rs	8.9833
c	0
b	0
f	0
cc	5
nop	1

# -*- coding: utf-8 -*-
#
# This file is part of SENAITE.CORE.
#
# SENAITE.CORE is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright 2018-2021 by it's authors.
# Some rights reserved, see README and LICENSE.

from bika.lims import api
from bika.lims.api.snapshot import pause_snapshots_for
from bika.lims.api.snapshot import resume_snapshots_for
from bika.lims.interfaces import IAnalysisRequestPartition
from bika.lims.interfaces import IDetachedPartition
from bika.lims.interfaces import IReceived
from bika.lims.interfaces import IVerified
from bika.lims.utils import changeWorkflowState
from bika.lims.utils.analysisrequest import create_retest
from bika.lims.workflow import doActionFor as do_action_for
from bika.lims.workflow import get_prev_status_from_history
from bika.lims.workflow.analysisrequest import do_action_to_analyses
from bika.lims.workflow.analysisrequest import do_action_to_ancestors
from bika.lims.workflow.analysisrequest import do_action_to_descendants
from DateTime import DateTime
from Products.CMFCore.WorkflowCore import WorkflowException
from senaite.core.workflow import SAMPLE_WORKFLOW
from zope.interface import alsoProvides
from zope.interface import noLongerProvides


def before_sample(analysis_request):
    """Method triggered before "sample" transition for the Analysis Request
    passed in is performed
    """
    if not analysis_request.getDateSampled():
        analysis_request.setDateSampled(DateTime())
    if not analysis_request.getSampler():
        analysis_request.setSampler(api.get_current_user().id)


def after_no_sampling_workflow(analysis_request):
    """Function triggered after "no_sampling_workflow transition for the
    Analysis Request passed in is performed
    """
    setup = api.get_setup()
    if setup.getAutoreceiveSamples():
        # Auto-receive samples is enabled. Note transition to "received" state
        # will only take place if the current user has enough privileges (this
        # is handled by do_action_for already).
        do_action_for(analysis_request, "receive")


def after_reject(analysis_request):
    """Method triggered after a 'reject' transition for the Analysis Request
    passed in is performed. Cascades the transition to descendants (partitions)
    and analyses as well. If "notify on rejection" setting is enabled, sends a
    notification to the client contact.
    """
    do_action_to_descendants(analysis_request, "reject")
    do_action_to_analyses(analysis_request, "reject")


def after_retract(analysis_request):
    """Method triggered after a 'retract' transition for the Analysis Request
    passed in is performed. Cascades the transition to descendants (partitions)
    and analyses as well.
    """
    do_action_to_descendants(analysis_request, "retract")
    do_action_to_analyses(analysis_request, "retract")


def after_invalidate(obj):
    """Function triggered after 'invalidate' transition for the Analysis
    Request passed in is performed. Creates a retest
    """
    create_retest(obj)


def after_submit(analysis_request):
    """Function called after a 'submit' transition is triggered. Promotes the
    submission of ancestors and descendants (partitions).
    """
    # This transition is not meant to be triggered manually by the user, rather
    # promoted by analyses. Hence, there is no need to cascade the transition
    # to the analyses the analysis request contains.
    do_action_to_ancestors(analysis_request, "submit")
    do_action_to_descendants(analysis_request, "submit")


def after_verify(analysis_request):
    """Method triggered after a 'verify' transition for the Analysis Request
    passed in is performed. Promotes the 'verify' transition to ancestors,
    descendants and to the analyses that belong to the analysis request.
    """
    # Mark this analysis request as IReceived
    alsoProvides(analysis_request, IVerified)

    # This transition is not meant to be triggered manually by the user, rather
    # promoted by analyses. Hence, there is no need to cascade the transition
    # to the analyses the analysis request contains.
    do_action_to_ancestors(analysis_request, "verify")
    do_action_to_descendants(analysis_request, "verify")


def after_prepublish(analysis_request):
    """Method triggered after a 'prepublish' transition for the Analysis
    Request passed in is performed. Performs the 'publish' transition to the
    descendant partitions.

    Also see: https://github.com/senaite/senaite.core/pull/1428
    """
    do_action_to_descendants(analysis_request, "publish")


def after_publish(analysis_request):
    """Method triggered after an 'publish' transition for the Analysis Request
    passed in is performed. Performs the 'publish' transition Publishes the
    descendant partitions and all analyses associated to the analysis request
    as well.
    """
    do_action_to_descendants(analysis_request, "publish")
    do_action_to_analyses(analysis_request, "publish")


def after_reinstate(analysis_request):
    """Method triggered after a 'reinstate' transition for the Analysis Request
    passed in is performed. Sets its status to the last status before it was
    cancelled. Reinstates the descendant partitions and all the analyses
    associated to the analysis request as well.
    """
    do_action_to_descendants(analysis_request, "reinstate")
    do_action_to_analyses(analysis_request, "reinstate")

    # Force the transition to previous state before the request was cancelled
    prev_status = get_prev_status_from_history(analysis_request, "cancelled")
    changeWorkflowState(analysis_request, SAMPLE_WORKFLOW, prev_status,
                        action="reinstate")
    analysis_request.reindexObject()


def after_cancel(analysis_request):
    """Method triggered after a 'cancel' transition for the Analysis Request
    passed in is performed. Cascades this transition to its analyses and
    partitions.
    """
    do_action_to_descendants(analysis_request, "cancel")
    do_action_to_analyses(analysis_request, "cancel")


def after_receive(analysis_request):
    """Method triggered after "receive" transition for the Analysis Request
    passed in is performed
    """
    # Mark this analysis request as IReceived
    alsoProvides(analysis_request, IReceived)

    analysis_request.setDateReceived(DateTime())
    do_action_to_analyses(analysis_request, "initialize")


def after_sample(analysis_request):
    """Method triggered after "sample" transition for the Analysis Request
    passed in is performed
    """
    analysis_request.setDateSampled(DateTime())


def after_rollback_to_receive(analysis_request):
    """Function triggered after "rollback to receive" transition is performed
    """
    if IVerified.providedBy(analysis_request):
        noLongerProvides(analysis_request, IVerified)


def after_detach(analysis_request):
    """Function triggered after "detach" transition is performed
    """
    # Unbind the sample from its parent (the primary)
    parent = analysis_request.getParentAnalysisRequest()
    analysis_request.setParentAnalysisRequest(None)

    # Assign the primary from which the sample has been detached
    analysis_request.setDetachedFrom(parent)

    # This sample is no longer a partition
    noLongerProvides(analysis_request, IAnalysisRequestPartition)

    # And we mark the sample with IDetachedPartition
    alsoProvides(analysis_request, IDetachedPartition)

    # Reindex both the parent and the detached one
    analysis_request.reindexObject()
    parent.reindexObject()

    # And the analyses too. aranalysesfield relies on a search against the
    # catalog to return the analyses: calling `getAnalyses` to the parent
    # will return all them, so no need to do the same with the detached
    analyses = parent.getAnalyses(full_objects=True)
    map(lambda an: an.reindexObject(), analyses)


def after_dispatch(sample):
    """Event triggered after "dispatch" transition takes place for a given sample
    """
    primary = sample.getParentAnalysisRequest()

    def get_last_wf_comment(obj):
        entry = api.get_review_history(obj)[0]
        return entry.get("comments", "")

    def dispatch(obj, comment=""):
        wf = api.get_tool("portal_workflow")
        try:
            wf.doActionFor(obj, "dispatch", comment=comment)
            return True
        except WorkflowException:
            return False

    if not primary:
        # propagate to transitions
        partitions = sample.getDescendants(all_descendants=False)
        for partition in partitions:
            comment = get_last_wf_comment(sample)
            dispatch(partition, comment)
        return

    # Return when primary sample is already dispatched
    if api.get_workflow_status_of(primary) == "dispatched":
        return

    # Dipsatch primary sample when all partitions are dispatched
    parts = primary.getDescendants()
    # Partitions in some statuses won't be considered
    skip = ["dispatched", "cancelled", "retracted", "rejected"]
    parts = filter(lambda part: api.get_review_status(part) not in skip, parts)
    if len(parts) == 0:
        # There are no partitions left, transition the primary
        comment = get_last_wf_comment(sample)
        dispatch(primary, comment)


def after_restore(sample):
    """Event triggered after "restore" transition takes place for a sample
    """

    # Transition the sample to the state before it was stored
    previous_state = api.get_previous_worfklow_status_of(
        sample, skip=["dispatched"], default="sample_due")

    # Note: we pause the snapshots here because events are fired next
    pause_snapshots_for(sample)
    changeWorkflowState(sample, SAMPLE_WORKFLOW, previous_state)
    resume_snapshots_for(sample)

    # Reindex the sample
    sample.reindexObject()

    # If the sample is a partition, try to promote to the primary
    primary = sample.getParentAnalysisRequest()
    if not primary:
        # propagate to transitions
        partitions = sample.getDescendants(all_descendants=False)
        for partition in partitions:
            do_action_for(partition, "restore")
        return

    # Return when primary sample is not dispatched
    if api.get_workflow_status_of(primary) != "dispatched":
        return

    # Restore primary sample if all its partitions have been restored
    parts = primary.getDescendants()
    states = map(api.get_workflow_status_of, parts)
    if "dispatched" not in states:
        do_action_for(primary, "restore")


1			# -- coding: utf-8 --
2			#
3			# This file is part of SENAITE.CORE.
4			#
5			# SENAITE.CORE is free software: you can redistribute it and/or modify it under
6			# the terms of the GNU General Public License as published by the Free Software
7			# Foundation, version 2.
8			#
9			# This program is distributed in the hope that it will be useful, but WITHOUT
10			# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11			# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
12			# details.
13			#
14			# You should have received a copy of the GNU General Public License along with
15			# this program; if not, write to the Free Software Foundation, Inc., 51
16			# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17			#
18			# Copyright 2018-2021 by it's authors.
19			# Some rights reserved, see README and LICENSE.
20
21			from bika.lims import api
22			from bika.lims.api.snapshot import pause_snapshots_for
23			from bika.lims.api.snapshot import resume_snapshots_for
24			from bika.lims.interfaces import IAnalysisRequestPartition
25			from bika.lims.interfaces import IDetachedPartition
26			from bika.lims.interfaces import IReceived
27			from bika.lims.interfaces import IVerified
28			from bika.lims.utils import changeWorkflowState
29			from bika.lims.utils.analysisrequest import create_retest
30			from bika.lims.workflow import doActionFor as do_action_for
31			from bika.lims.workflow import get_prev_status_from_history
32			from bika.lims.workflow.analysisrequest import do_action_to_analyses
33			from bika.lims.workflow.analysisrequest import do_action_to_ancestors
34			from bika.lims.workflow.analysisrequest import do_action_to_descendants
35			from DateTime import DateTime
36			from Products.CMFCore.WorkflowCore import WorkflowException
37			from senaite.core.workflow import SAMPLE_WORKFLOW
38			from zope.interface import alsoProvides
39			from zope.interface import noLongerProvides
40
41
42			def before_sample(analysis_request):
43			"""Method triggered before "sample" transition for the Analysis Request
44			passed in is performed
45			"""
46			if not analysis_request.getDateSampled():
47			analysis_request.setDateSampled(DateTime())
48			if not analysis_request.getSampler():
49			analysis_request.setSampler(api.get_current_user().id)
50
51
52			def after_no_sampling_workflow(analysis_request):
53			"""Function triggered after "no_sampling_workflow transition for the
54			Analysis Request passed in is performed
55			"""
56			setup = api.get_setup()
57			if setup.getAutoreceiveSamples():
58			# Auto-receive samples is enabled. Note transition to "received" state
59			# will only take place if the current user has enough privileges (this
60			# is handled by do_action_for already).
61			do_action_for(analysis_request, "receive")
62
63
64			def after_reject(analysis_request):
65			"""Method triggered after a 'reject' transition for the Analysis Request
66			passed in is performed. Cascades the transition to descendants (partitions)
67			and analyses as well. If "notify on rejection" setting is enabled, sends a
68			notification to the client contact.
69			"""
70			do_action_to_descendants(analysis_request, "reject")
71			do_action_to_analyses(analysis_request, "reject")
72
73
74			def after_retract(analysis_request):
75			"""Method triggered after a 'retract' transition for the Analysis Request
76			passed in is performed. Cascades the transition to descendants (partitions)
77			and analyses as well.
78			"""
79			do_action_to_descendants(analysis_request, "retract")
80			do_action_to_analyses(analysis_request, "retract")
81
82
83			def after_invalidate(obj):
84			"""Function triggered after 'invalidate' transition for the Analysis
85			Request passed in is performed. Creates a retest
86			"""
87			create_retest(obj)
88
89
90			def after_submit(analysis_request):
91			"""Function called after a 'submit' transition is triggered. Promotes the
92			submission of ancestors and descendants (partitions).
93			"""
94			# This transition is not meant to be triggered manually by the user, rather
95			# promoted by analyses. Hence, there is no need to cascade the transition
96			# to the analyses the analysis request contains.
97			do_action_to_ancestors(analysis_request, "submit")
98			do_action_to_descendants(analysis_request, "submit")
99
100
101			def after_verify(analysis_request):
102			"""Method triggered after a 'verify' transition for the Analysis Request
103			passed in is performed. Promotes the 'verify' transition to ancestors,
104			descendants and to the analyses that belong to the analysis request.
105			"""
106			# Mark this analysis request as IReceived
107			alsoProvides(analysis_request, IVerified)
108
109			# This transition is not meant to be triggered manually by the user, rather
110			# promoted by analyses. Hence, there is no need to cascade the transition
111			# to the analyses the analysis request contains.
112			do_action_to_ancestors(analysis_request, "verify")
113			do_action_to_descendants(analysis_request, "verify")
114
115
116			def after_prepublish(analysis_request):
117			"""Method triggered after a 'prepublish' transition for the Analysis
118			Request passed in is performed. Performs the 'publish' transition to the
119			descendant partitions.
120
121			Also see: https://github.com/senaite/senaite.core/pull/1428
122			"""
123			do_action_to_descendants(analysis_request, "publish")
124
125
126			def after_publish(analysis_request):
127			"""Method triggered after an 'publish' transition for the Analysis Request
128			passed in is performed. Performs the 'publish' transition Publishes the
129			descendant partitions and all analyses associated to the analysis request
130			as well.
131			"""
132			do_action_to_descendants(analysis_request, "publish")
133			do_action_to_analyses(analysis_request, "publish")
134
135
136			def after_reinstate(analysis_request):
137			"""Method triggered after a 'reinstate' transition for the Analysis Request
138			passed in is performed. Sets its status to the last status before it was
139			cancelled. Reinstates the descendant partitions and all the analyses
140			associated to the analysis request as well.
141			"""
142			do_action_to_descendants(analysis_request, "reinstate")
143			do_action_to_analyses(analysis_request, "reinstate")
144
145			# Force the transition to previous state before the request was cancelled
146			prev_status = get_prev_status_from_history(analysis_request, "cancelled")
147			changeWorkflowState(analysis_request, SAMPLE_WORKFLOW, prev_status,
148			action="reinstate")
149			analysis_request.reindexObject()
150
151
152			def after_cancel(analysis_request):
153			"""Method triggered after a 'cancel' transition for the Analysis Request
154			passed in is performed. Cascades this transition to its analyses and
155			partitions.
156			"""
157			do_action_to_descendants(analysis_request, "cancel")
158			do_action_to_analyses(analysis_request, "cancel")
159
160
161			def after_receive(analysis_request):
162			"""Method triggered after "receive" transition for the Analysis Request
163			passed in is performed
164			"""
165			# Mark this analysis request as IReceived
166			alsoProvides(analysis_request, IReceived)
167
168			analysis_request.setDateReceived(DateTime())
169			do_action_to_analyses(analysis_request, "initialize")
170
171
172			def after_sample(analysis_request):
173			"""Method triggered after "sample" transition for the Analysis Request
174			passed in is performed
175			"""
176			analysis_request.setDateSampled(DateTime())
177
178
179			def after_rollback_to_receive(analysis_request):
180			"""Function triggered after "rollback to receive" transition is performed
181			"""
182			if IVerified.providedBy(analysis_request):
183			noLongerProvides(analysis_request, IVerified)
184
185
186			def after_detach(analysis_request):
187			"""Function triggered after "detach" transition is performed
188			"""
189			# Unbind the sample from its parent (the primary)
190			parent = analysis_request.getParentAnalysisRequest()
191			analysis_request.setParentAnalysisRequest(None)
192
193			# Assign the primary from which the sample has been detached
194			analysis_request.setDetachedFrom(parent)
195
196			# This sample is no longer a partition
197			noLongerProvides(analysis_request, IAnalysisRequestPartition)
198
199			# And we mark the sample with IDetachedPartition
200			alsoProvides(analysis_request, IDetachedPartition)
201
202			# Reindex both the parent and the detached one
203			analysis_request.reindexObject()
204			parent.reindexObject()
205
206			# And the analyses too. aranalysesfield relies on a search against the
207			# catalog to return the analyses: calling `getAnalyses` to the parent
208			# will return all them, so no need to do the same with the detached
209			analyses = parent.getAnalyses(full_objects=True)
210			map(lambda an: an.reindexObject(), analyses)
211
212
213			def after_dispatch(sample):
214			"""Event triggered after "dispatch" transition takes place for a given sample
215			"""
216			primary = sample.getParentAnalysisRequest()
217
218			def get_last_wf_comment(obj):
219			entry = api.get_review_history(obj)[0]
220			return entry.get("comments", "")
221
222			def dispatch(obj, comment=""):
223			wf = api.get_tool("portal_workflow")
224			try:
225			wf.doActionFor(obj, "dispatch", comment=comment)
226			return True
227			except WorkflowException:
228			return False
229
230			if not primary:
231			# propagate to transitions
232			partitions = sample.getDescendants(all_descendants=False)
233			for partition in partitions:
234			comment = get_last_wf_comment(sample)
235			dispatch(partition, comment)
236			return
237
238			# Return when primary sample is already dispatched
239			if api.get_workflow_status_of(primary) == "dispatched":
240			return
241
242			# Dipsatch primary sample when all partitions are dispatched
243			parts = primary.getDescendants()
244			# Partitions in some statuses won't be considered
245			skip = ["dispatched", "cancelled", "retracted", "rejected"]
246			parts = filter(lambda part: api.get_review_status(part) not in skip, parts)
247			if len(parts) == 0:
248			# There are no partitions left, transition the primary
249			comment = get_last_wf_comment(sample)
250			dispatch(primary, comment)
251
252
253			def after_restore(sample):
254			"""Event triggered after "restore" transition takes place for a sample
255			"""
256
257			# Transition the sample to the state before it was stored
258			previous_state = api.get_previous_worfklow_status_of(
259			sample, skip=["dispatched"], default="sample_due")
260
261			# Note: we pause the snapshots here because events are fired next
262			pause_snapshots_for(sample)
263			changeWorkflowState(sample, SAMPLE_WORKFLOW, previous_state)
264			resume_snapshots_for(sample)
265
266			# Reindex the sample
267			sample.reindexObject()
268
269			# If the sample is a partition, try to promote to the primary
270			primary = sample.getParentAnalysisRequest()
271			if not primary:
272			# propagate to transitions
273			partitions = sample.getDescendants(all_descendants=False)
274			for partition in partitions:
275			do_action_for(partition, "restore")
276			return
277
278			# Return when primary sample is not dispatched
279			if api.get_workflow_status_of(primary) != "dispatched":
280			return
281
282			# Restore primary sample if all its partitions have been restored
283			parts = primary.getDescendants()
284			states = map(api.get_workflow_status_of, parts)
285			if "dispatched" not in states:
286			do_action_for(primary, "restore")
287

senaite / senaite.core

Pull Request — 2.x (#1772)

after_restore() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like