(self, envid, dest, save_tar=False)
| 136 | return response |
| 137 | |
| 138 | def get_template(self, envid, dest, save_tar=False): |
| 139 | url = "/v1/algorithm-environments/edge/environment-specifications/" + envid + "/template" |
| 140 | filename = "template.tar.gz" |
| 141 | |
| 142 | if not os.path.exists(dest): |
| 143 | os.makedirs(dest) |
| 144 | |
| 145 | filepath = os.path.join(dest, filename) |
| 146 | response = self.getStreamHelper(url) |
| 147 | |
| 148 | if response.ok: |
| 149 | with open(filepath, 'wb') as f: |
| 150 | for chunk in response.iter_content(chunk_size=1024 * 8): |
| 151 | if chunk: |
| 152 | f.write(chunk) |
| 153 | f.flush() |
| 154 | os.fsync(f.fileno()) |
| 155 | |
| 156 | tar = tarfile.open(filepath, "r:gz") |
| 157 | tar.extractall(dest) |
| 158 | tar.close() |
| 159 | |
| 160 | if not save_tar: |
| 161 | try: |
| 162 | os.remove(filepath) |
| 163 | except OSError as e: |
| 164 | print(e) |
| 165 | return response |
| 166 | else: |
| 167 | return json.loads(response.content.decode("utf-8")) |
| 168 | |
| 169 | def get_environment(self, language): |
| 170 | url = "/v1/algorithm-environments/edge/languages/" + language + "/environments" |
nothing calls this directly
no test coverage detected