Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# one-two context entry pattern
async with orchestrator(DATAFLOW) as octx:
# Run all the operations, Each iteration of this loop happens
# when all inputs are exhausted for a context, the output
# operations are then run and their results are yielded
async for package_name, results in octx.run(
{
# For each package add a new input set to the input network
# The context operations execute under is the package name
# to evaluate. Contexts ensure that data pertaining to
# package A doesn't mingle with data pertaining to package B
package_name: [
# The only input to the operations is the package name.
Input(
value=package_name,
definition=pypi_package_json.op.inputs[
"package"
],
)
]
for package_name in self.packages
}
):
# Grab the number of safety issues and the bandit report
# from the results dict
safety_issues = results[
safety_check.op.outputs["issues"].name
]
bandit_report = results[
run_bandit.op.outputs["report"].name
]
# Decide if those numbers mean we should stop ship or not