feat: use subprocess.run() to replace os.system();

pypots/utils/commands/__init__.py

@@ -6,6 +6,7 @@
 # License: GLP-v3
 
 
+import subprocess
 from abc import ABC, abstractmethod
 from argparse import ArgumentParser
 
@@ -16,6 +17,21 @@ class BaseCommand(ABC):
  def register_subcommand(parser: ArgumentParser):
  raise NotImplementedError()
 
+ @staticmethod
+ def execute_command(command):
+ exec_result = subprocess.run(
+ command,
+ text=True,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ shell=True,
+ )
+ if exec_result.returncode != 0:
+ if len(exec_result.stderr) > 0:
+ raise RuntimeError(exec_result.stderr)
+ if len(exec_result.stdout) > 0:
+ raise RuntimeError(exec_result.stdout)
+
  @abstractmethod
  def run(self):
  raise NotImplementedError()

pypots/utils/commands/dev.py

@@ -152,7 +152,7 @@ def run(self):
  shutil.rmtree("dist", ignore_errors=True)
  shutil.rmtree("pypots.egg-info", ignore_errors=True)
  elif self._build:
- os.system("python setup.py sdist bdist bdist_wheel")
+ self.execute_command("python setup.py sdist bdist bdist_wheel")
  elif self._run_tests:
  pytest_command = f"pytest -k {self._k}" if self._k else "pytest"
  command_to_run_test = (
@@ -162,21 +162,18 @@ def run(self):
  )
  logger.info(f"Executing '{command_to_run_test}'...")
 
- os.system(command_to_run_test)
+ self.execute_command(command_to_run_test)
  if self._show_coverage:
- os.system("coverage report -m")
- os.system("rm -rf .coverage")
- else:
- logger.info(
- "Omit the code coverage report. Enable it by using --show_coverage if in need."
- )
- os.system("rm -rf .pytest_cache")
+ self.execute_command("coverage report -m")
+ self.execute_command("rm -rf .coverage")
+
+ self.execute_command("rm -rf .pytest_cache")
 
  elif self._lint_code:
  logger.info("Reformatting with Black...")
- os.system("black .")
+ self.execute_command("black .")
  logger.info("Linting with Flake8...")
- os.system("flake8 .")
+ self.execute_command("flake8 .")
 
  except ImportError:
  raise ImportError(IMPORT_ERROR_MESSAGE)

pypots/utils/commands/doc.py

@@ -150,18 +150,6 @@ def check_arguments(self):
  f"but you're running it under the path {os.getcwd()}. Please make a check."
  )
 
- if self._port is not None:
- assert self._view_doc, (
- "Argument `--port` should combine the use of `--view_doc`. "
- "Try `pypots-cli doc --view_doc --port your_port`"
- )
-
- if self._branch is not None:
- assert self._gene_rst, (
- "Argument `--branch` should combine the use of `--gene_rst`. "
- "Try `pypots-cli doc --gene_rst --branch your_branch`"
- )
-
  if self._cleanup:
  assert not self._gene_rst and not self._gene_html and not self._view_doc, (
  "Argument `--cleanup` should be used alone. "
@@ -207,7 +195,7 @@ def run(self):
 
  # Generate the docs according to the cloned code
  logger.info("Generating rst files...")
- os.system(
+ self.execute_command(
  "SPHINX_APIDOC_OPTIONS=members,undoc-members,show-inheritance,inherited-members "
  f"sphinx-apidoc {CLONED_LATEST_PYPOTS} -o {CLONED_LATEST_PYPOTS}/rst"
  )
@@ -224,14 +212,14 @@ def run(self):
  if self._gene_html:
  logger.info("Generating static HTML files...")
  purge_statics()
- os.system("make html")
+ self.execute_command("make html")
 
  if self._view_doc:
  assert os.path.exists(
  "_build/html"
  ), "_build/html does not exists, please run `pypots-cli doc --gene_html` first"
- logger.info("Deploying HTML...")
- os.system(
+ logger.info(f"Deploying HTML to https://127.0.0.1:{self._port}...")
+ self.execute_command(
  f"python -m http.server {self._port} -d _build/html -b 127.0.0.1"
  )