Skip to content

Registry

hypha.apply.funds.workflows.registry

This file defines classes which allow you to compose workflows based on the following structure:

Workflow -> Stage -> Phase -> Action

Current limitations: * Changing the name of a phase will mean that any object which references it cannot progress. [will be fixed when streamfield, may require intermediate fix prior to launch] * Do not reorder without looking at workflow automations steps in form_valid() in hypha/apply/funds/views.py and hypha/apply/review/views.py.

Request module-attribute 

Request = Workflow('Request', 'single', **phase_data(SingleStageDefinition))

RequestSameTime module-attribute 

RequestSameTime = Workflow('Request with same time review', 'single_same', **phase_data(SingleStageSameDefinition))

RequestExternal module-attribute 

RequestExternal = Workflow('Request with external review', 'single_ext', **phase_data(SingleStageExternalDefinition))

RequestCommunity module-attribute 

RequestCommunity = Workflow('Request with community review', 'single_com', **phase_data(SingleStageCommunityDefinition))

ConceptProposal module-attribute 

ConceptProposal = Workflow('Concept & Proposal', 'double', **phase_data(DoubleStageDefinition))

WORKFLOWS module-attribute 

WORKFLOWS = {admin_name: Request, admin_name: RequestSameTime, admin_name: RequestExternal, admin_name: RequestCommunity, admin_name: ConceptProposal}

PHASES module-attribute 

PHASES = list(from_iterable(items() for workflow in values()))

STATUSES module-attribute 

STATUSES = defaultdict(set)

active_statuses module-attribute 

active_statuses = [status for (status, _) in PHASES if 'accepted' not in status and 'rejected' not in status and 'invited' not in status]

review_statuses module-attribute 

review_statuses = get_review_statuses()

ext_review_statuses module-attribute 

ext_review_statuses = get_ext_review_statuses()

ext_or_higher_statuses module-attribute 

ext_or_higher_statuses = get_ext_or_higher_statuses()

accepted_statuses module-attribute 

accepted_statuses = get_accepted_statuses()

dismissed_statuses module-attribute 

dismissed_statuses = get_dismissed_statuses()

phase_data 

phase_data(phases)

Transforms a workflow definition into a dictionary of Phase objects.

Parameters:

  • phases (List[dict]) –

    A list of dictionaries defining the workflow phases and their configurations.

Returns:

  • dict[str, Phase] –

    A dictionary where keys are phase names and values are Phase objects, each initialized with: - phase name - step number (order in workflow) - additional configuration data from the phase definition

Example

Input phases = [ {'draft': {'permissions': {...}}}, {'review': {'permissions': {...}}} ]

Returns = { 'draft': Phase('draft', step=0, permissions={...}), 'review': Phase('review', step=1, permissions={...}) }

Source code in hypha/apply/funds/workflows/registry.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def phase_data(phases: List[dict]) -> dict[str, Phase]:
    """
    Transforms a workflow definition into a dictionary of Phase objects.

    Args:
        phases: A list of dictionaries defining the workflow phases and their configurations.

    Returns:
        A dictionary where keys are phase names and values are Phase objects, each initialized with:
            - phase name
            - step number (order in workflow)
            - additional configuration data from the phase definition

    Example:
        Input phases = [
            {'draft': {'permissions': {...}}},
            {'review': {'permissions': {...}}}
        ]

        Returns = {
            'draft': Phase('draft', step=0, permissions={...}),
            'review': Phase('review', step=1, permissions={...})
        }
    """

    def unpack_phases(phases):
        return (
            (step, name, phase_data)
            for step, step_data in enumerate(phases)
            for name, phase_data in step_data.items()
        )

    return {
        phase_name: Phase(phase_name, step=step, **phase_data)
        for step, phase_name, phase_data in unpack_phases(phases)
    }

get_review_active_statuses 

get_review_active_statuses(user=None)
Source code in hypha/apply/funds/workflows/registry.py 
112
113
114
115
116
117
118
119
120
121
def get_review_active_statuses(user=None):
    reviews = set()

    for phase_name, phase in PHASES:
        if phase_name in active_statuses:
            if user is None:
                reviews.add(phase_name)
            elif phase.permissions.can_review(user):
                reviews.add(phase_name)
    return reviews

get_review_statuses 

get_review_statuses(user=None)
Source code in hypha/apply/funds/workflows/registry.py 
124
125
126
127
128
129
130
131
132
133
def get_review_statuses(user=None):
    reviews = set()

    for phase_name, phase in PHASES:
        if "review" in phase_name and "discussion" not in phase_name:
            if user is None:
                reviews.add(phase_name)
            elif phase.permissions.can_review(user):
                reviews.add(phase_name)
    return reviews

get_ext_review_statuses 

get_ext_review_statuses()
Source code in hypha/apply/funds/workflows/registry.py 
136
137
138
139
140
141
142
def get_ext_review_statuses():
    reviews = set()

    for phase_name, _phase in PHASES:
        if phase_name.endswith("external_review"):
            reviews.add(phase_name)
    return reviews

get_ext_or_higher_statuses 

get_ext_or_higher_statuses()

Returns a set of all the statuses for all workflow which are External Review or higher than that.

Source code in hypha/apply/funds/workflows/registry.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def get_ext_or_higher_statuses():
    """
    Returns a set of all the statuses for all workflow which are
    External Review or higher than that.
    """
    reviews = set()

    for workflow in WORKFLOWS.values():
        step = None
        for phase in workflow.values():
            if phase.name.endswith("external_review"):
                # Update the step for this workflow as External review state
                step = phase.step

            # Phase should have step higher or equal than External
            # review state for this workflow
            if step and phase.step >= step:
                reviews.add(phase.name)
    return reviews

get_accepted_statuses 

get_accepted_statuses()
Source code in hypha/apply/funds/workflows/registry.py 
166
167
168
169
170
171
def get_accepted_statuses():
    accepted_statuses = set()
    for phase_name, phase in PHASES:
        if phase.display_name == "Accepted":
            accepted_statuses.add(phase_name)
    return accepted_statuses

get_dismissed_statuses 

get_dismissed_statuses()
Source code in hypha/apply/funds/workflows/registry.py 
174
175
176
177
178
179
def get_dismissed_statuses():
    dismissed_statuses = set()
    for phase_name, phase in PHASES:
        if phase.display_name == "Dismissed":
            dismissed_statuses.add(phase_name)
    return dismissed_statuses