MCPcopy Create free account
hub / github.com/celery/celery / _firstpass

Method _firstpass

celery/bootsteps.py:224–233  ·  view source on GitHub ↗
(self, steps)

Source from the content-addressed store, hash-verified

222 return next((C for C in self.steps.values() if C.last), None)
223
224 def _firstpass(self, steps):
225 for step in steps.values():
226 step.requires = [symbol_by_name(dep) for dep in step.requires]
227 stream = deque(step.requires for step in steps.values())
228 while stream:
229 for node in stream.popleft():
230 node = symbol_by_name(node)
231 if node.name not in self.steps:
232 steps[node.name] = node
233 stream.append(node.requires)
234
235 def _finalize_steps(self, steps):
236 last = self._find_last()

Callers 1

_finalize_stepsMethod · 0.95

Calls

no outgoing calls

Tested by

no test coverage detected