Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
pypi_package_url,
pypi_package_contents,
cleanup_pypi_package,
safety_check,
run_bandit,
GetSingle,
)
# Seed inputs are added to each executing context. The following Input tells the
# GetSingle output operation that we want the output of the network to include
# data matching the "issues" output of the safety_check operation, and the
# "report" output of the run_bandit operation, for each context.
DATAFLOW.seed.append(
Input(
value=[
safety_check.op.outputs["issues"].name,
run_bandit.op.outputs["report"].name,
],
definition=GetSingle.op.inputs["spec"],
)
)
class Install(CMD):
arg_packages = Arg(
"packages", nargs="+", help="Package to check if we should install"
)
async def run(self):
# Create an Orchestrator which will manage the running of our operations
async with MemoryOrchestrator.withconfig({}) as orchestrator:
# Create a orchestrator context, everything in DFFML follows this