(self, pkg: str, version: str= "", no_cache: bool=False, call_log:Optional[Callable[[str], None]] = None)
| 326 | return _run_command_with_call_log(sh, call_log) |
| 327 | |
| 328 | def pip_install(self, pkg: str, version: str= "", no_cache: bool=False, call_log:Optional[Callable[[str], None]] = None) -> Optional[str]: |
| 329 | if not self.pip_bin(): |
| 330 | self._install_pip(call_log) |
| 331 | if not callable(call_log): |
| 332 | call_log = lambda x: print(x) |
| 333 | cmd = [self.pip_bin(), "install"] |
| 334 | if version: |
| 335 | cmd.append("{}=={}".format(pkg, version)) |
| 336 | else: |
| 337 | cmd.append(pkg) |
| 338 | |
| 339 | cmd.append("-i") |
| 340 | if self._pip_source: |
| 341 | cmd.append(self._pip_source) |
| 342 | else: |
| 343 | cmd.append(self.default_pip_source) |
| 344 | if no_cache: |
| 345 | cmd.append("--no-cache-dir") |
| 346 | |
| 347 | cmd_pip = " ".join(cmd) |
| 348 | if pkg.lower().find("mysqlclient") != -1: |
| 349 | mysql_flag = self.build_mysql_env() |
| 350 | cmd_pip = mysql_flag + cmd_pip |
| 351 | |
| 352 | cmd = self.activate_shell() + "\n" + cmd_pip |
| 353 | error_list = set() |
| 354 | err_msg = "ERROR: Failed building wheel for" |
| 355 | def check_error_with_log(output: str) -> None: |
| 356 | idx = output.find(err_msg) |
| 357 | if idx != -1: |
| 358 | pkg_name = output[idx + len(err_msg):].strip().split()[0].strip() |
| 359 | if pkg_name in self._ONLY_BINARY_NAMES: |
| 360 | error_list.add(pkg_name) |
| 361 | return call_log(output) |
| 362 | |
| 363 | res = _run_command_with_call_log(cmd, check_error_with_log) |
| 364 | if error_list: |
| 365 | call_log("发现编译错误的包:【{}】,尝试以二进制方式安装".format(", ".join(error_list))) |
| 366 | cmd = cmd + " --only-binary=" + ",".join(error_list) |
| 367 | return _run_command_with_call_log(cmd, call_log) |
| 368 | return res |
| 369 | |
| 370 | @staticmethod |
| 371 | def build_mysql_env() -> str: |
no test coverage detected