Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
url = "{0}/{1}/archive/{2}.zip".format(github_prefix, git_repo, branch)
response = urlopen(url)
temp_dir = os.path.join(os.path.expanduser("~"), '.iotedgedev')
self.utility.ensure_dir(temp_dir)
zip_file_prefix = "{0}-{1}".format(git_repo, branch)
temp_template_dir = os.path.join(temp_dir, zip_file_prefix)
if os.path.exists(temp_template_dir):
shutil.rmtree(temp_template_dir)
with ZipFile(BytesIO(response.read())) as zip_f:
zip_f.extractall(temp_dir)
shutil.move(temp_template_dir, os.path.join(cwd, name))
module = Module(self.envvars, self.utility, os.path.join(cwd, name))
module.repository = repo
module.dump()
elif template == "csharp":
dotnet = DotNet(self.output, self.utility)
dotnet.install_module_template()
dotnet.create_custom_module(name, repo, cwd)
elif template == "java":
self.utility.check_dependency("mvn --help".split(), "To add new Java modules, the Maven tool")
cmd = ["mvn",
"archetype:generate",
"-DarchetypeGroupId=com.microsoft.azure",
"-DarchetypeArtifactId=azure-iot-edge-archetype",
"-DgroupId={0}".format(group_id),
"-DartifactId={0}".format(name),
"-Dversion=1.0.0-SNAPSHOT",
"-Dpackage={0}".format(group_id),
modules_path = os.path.join(template_file_folder, self.envvars.MODULES_PATH)
if os.path.isdir(modules_path):
for folder_name in os.listdir(modules_path):
project_folder = os.path.join(modules_path, folder_name)
if os.path.exists(os.path.join(project_folder, "module.json")):
module = Module(self.envvars, self.utility, project_folder)
self._update_module_maps("MODULES.{0}".format(folder_name), module, placeholder_tag_map, tag_build_profile_map, default_platform)
# get image tags for ${MODULEDIR.xxx} placeholder
user_modules = deployment_manifest.get_user_modules()
for module_name, module_info in user_modules.items():
image = module_info["settings"]["image"]
match_result = re.search(Constants.moduledir_placeholder_pattern, image)
if (match_result is not None):
module_dir = match_result.group(1)
module = Module(self.envvars, self.utility, os.path.join(template_file_folder, module_dir))
self._update_module_maps("MODULEDIR<{0}>".format(module_dir), module, placeholder_tag_map, tag_build_profile_map, default_platform)
replacements = {}
for module_name, module_info in user_modules.items():
image = module_info["settings"]["image"]
if image in placeholder_tag_map:
tag = placeholder_tag_map[image]
replacements[module_name] = tag
if not self.utility.in_asterisk_list(module_name, bypass_modules):
tags_to_build.add(tag)
if not no_build or not no_push:
docker = Docker(self.envvars, self.utility, self.output)
for tag in tags_to_build:
if tag in tag_build_profile_map:
# BUILD DOCKER IMAGE
placeholder_tag_map = {}
# map image tag to BuildProfile object
tag_build_profile_map = {}
# image tags to build
# sample: 'localhost:5000/filtermodule:0.0.1-amd64'
tags_to_build = set()
deployment_manifest = DeploymentManifest(self.envvars, self.output, self.utility, template_file, True, True)
# get image tags for ${MODULES.modulename.xxx} placeholder
modules_path = os.path.join(template_file_folder, self.envvars.MODULES_PATH)
if os.path.isdir(modules_path):
for folder_name in os.listdir(modules_path):
project_folder = os.path.join(modules_path, folder_name)
if os.path.exists(os.path.join(project_folder, "module.json")):
module = Module(self.envvars, self.utility, project_folder)
self._update_module_maps("MODULES.{0}".format(folder_name), module, placeholder_tag_map, tag_build_profile_map, default_platform)
# get image tags for ${MODULEDIR.xxx} placeholder
user_modules = deployment_manifest.get_user_modules()
for module_name, module_info in user_modules.items():
image = module_info["settings"]["image"]
match_result = re.search(Constants.moduledir_placeholder_pattern, image)
if (match_result is not None):
module_dir = match_result.group(1)
module = Module(self.envvars, self.utility, os.path.join(template_file_folder, module_dir))
self._update_module_maps("MODULEDIR<{0}>".format(module_dir), module, placeholder_tag_map, tag_build_profile_map, default_platform)
replacements = {}
for module_name, module_info in user_modules.items():
image = module_info["settings"]["image"]
if image in placeholder_tag_map: