Refactor logging setup and enhance user site directory management in PackageInstaller and Upgrader
This commit is contained in:
parent
c9af30edda
commit
fa15576608
3 changed files with 45 additions and 9 deletions
|
|
@ -8,7 +8,17 @@ from pathlib import Path
|
|||
|
||||
import httpx
|
||||
|
||||
LOG = logging.getLogger("package_installer")
|
||||
LOG: logging.Logger = logging.getLogger("package_installer")
|
||||
|
||||
if base_dir := os.environ.get("YTP_CONFIG_PATH"):
|
||||
base_dir = Path(base_dir)
|
||||
user_site: Path = base_dir / f"python{sys.version_info.major}.{sys.version_info.minor}-packages"
|
||||
|
||||
if not user_site.exists():
|
||||
user_site.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if user_site.is_dir() and str(user_site) not in sys.path:
|
||||
sys.path.insert(0, str(user_site))
|
||||
|
||||
|
||||
def parse_version(v: str) -> tuple[int, ...]:
|
||||
|
|
@ -113,13 +123,16 @@ class PackageInstaller:
|
|||
cmd.extend(["--disable-pip-version-check", pkg])
|
||||
|
||||
try:
|
||||
subprocess.run(
|
||||
proc: subprocess.CompletedProcess[bytes] = subprocess.run(
|
||||
cmd,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
|
||||
)
|
||||
self.out(out=proc.stdout, err=proc.stderr)
|
||||
except subprocess.CalledProcessError as e:
|
||||
LOG.error(f"Failed to install package '{pkg}'. Error: {e}")
|
||||
LOG.error(f"Failed to install '{pkg}' (exit {e.returncode}). {e!s}")
|
||||
self.out(out=e.stdout, err=e.stderr)
|
||||
raise
|
||||
|
||||
def check(self, pkgs: Packages):
|
||||
|
|
@ -165,3 +178,12 @@ class PackageInstaller:
|
|||
return True
|
||||
|
||||
return False
|
||||
|
||||
def out(self, out: bytes | str | None = None, err=bytes | str | None) -> None:
|
||||
if out:
|
||||
for line in (line.strip() for line in out.strip().splitlines() if line.strip()):
|
||||
LOG.info(line.decode() if isinstance(line, bytes) else line)
|
||||
|
||||
if err:
|
||||
for line in (line.strip() for line in err.strip().splitlines() if line.strip()):
|
||||
LOG.warning(line.decode() if isinstance(line, bytes) else line)
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import time
|
|||
from logging.handlers import TimedRotatingFileHandler
|
||||
from multiprocessing.managers import SyncManager
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import coloredlogs
|
||||
from dotenv import load_dotenv
|
||||
|
|
@ -14,6 +15,9 @@ from dotenv import load_dotenv
|
|||
from .Utils import FileLogFormatter, arg_converter
|
||||
from .version import APP_BRANCH, APP_BUILD_DATE, APP_COMMIT_SHA, APP_VERSION
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from subprocess import CompletedProcess
|
||||
|
||||
|
||||
class Config:
|
||||
app_env: str = "production"
|
||||
|
|
@ -558,8 +562,8 @@ class Config:
|
|||
try:
|
||||
import subprocess
|
||||
|
||||
branch_result = subprocess.run(
|
||||
["git", "rev-parse", "--abbrev-ref", "HEAD"], # noqa: S607
|
||||
branch_result: CompletedProcess[str] = subprocess.run(
|
||||
["git", "-c", f"safe.directory={git_path.parent!s}", "rev-parse", "--abbrev-ref", "HEAD"], # noqa: S607
|
||||
cwd=os.path.dirname(git_path),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
|
|
@ -576,8 +580,8 @@ class Config:
|
|||
logging.warning("Git branch name is empty.")
|
||||
return
|
||||
|
||||
commit_result = subprocess.run(
|
||||
["git", "log", "-1", "--format=%ct_%H"], # noqa: S607
|
||||
commit_result: CompletedProcess[str] = subprocess.run(
|
||||
["git", "-c", f"safe.directory={git_path.parent!s}", "log", "-1", "--format=%ct_%H"], # noqa: S607
|
||||
cwd=os.path.dirname(git_path),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
|
|
|
|||
|
|
@ -13,7 +13,17 @@ from pathlib import Path
|
|||
from dotenv import load_dotenv
|
||||
from library.PackageInstaller import PackageInstaller, Packages
|
||||
|
||||
LOG = logging.getLogger("upgrader")
|
||||
LOG: logging.Logger = logging.getLogger("upgrader")
|
||||
|
||||
if base_dir := os.environ.get("YTP_CONFIG_PATH"):
|
||||
base_dir = Path(base_dir)
|
||||
user_site: Path = base_dir / f"python{sys.version_info.major}.{sys.version_info.minor}-packages"
|
||||
|
||||
if not user_site.exists():
|
||||
user_site.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if user_site.is_dir() and str(user_site) not in sys.path:
|
||||
sys.path.insert(0, str(user_site))
|
||||
|
||||
|
||||
class Upgrader:
|
||||
|
|
@ -59,7 +69,7 @@ class Upgrader:
|
|||
Packages(
|
||||
env=os.environ.get("YTP_PIP_PACKAGES", None),
|
||||
file=str(Path(config_path, "pip.txt")),
|
||||
upgrade=not bool(os.environ.get("YTP_PIP_IGNORE_UPDATES", False)),
|
||||
upgrade=not bool(os.environ.get("YTP_PIP_IGNORE_UPDATES", None)),
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
|
|
|
|||
Loading…
Reference in a new issue