Coverage for texgit / repository / process_manager.py: 89%
90 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-22 02:50 +0000
1"""
2A class for managing files and directories.
4A file manager provides a two-level abstraction for assigning names to paths.
5It exists within a certain base directory.
6Inside the base directory, it provides so-called "realms".
7Each realm is a separate namespace.
8With a realm, "names" are mapped to paths.
9The file manager ensures that the same realm-name combination is always
10assigned to the same path.
11The first time it is queried, the path is created.
12This path can be a file or a directory, depending on what was queried.
13Every realm-name combination always uniquely identifies a path and there can
14never be another realm-name combination pointing to the same path.
15The paths are randomized to avoid potential clashes.
17Once the file manager is closed, the realm-name to path associations are
18stored.
19When a new file manager instance is created for the same base directory, the
20associations of realms-names to paths are restored.
21This means that a program that creates output files for certain commands can
22then find these files again later.
23"""
24from os import environ
25from os.path import getsize
26from typing import Final, Iterable, Mapping
28from pycommons.ds.immutable_map import immutable_mapping
29from pycommons.io.console import logger
30from pycommons.io.path import Path, write_lines
31from pycommons.processes.python import PYTHON_ENV, PYTHON_INTERPRETER
32from pycommons.processes.shell import STREAM_CAPTURE, Command
33from pycommons.types import type_error
35from texgit.repository.fix_path import replace_base_path
36from texgit.repository.git_manager import GitManager, GitPath
39def _write(orig: str, dest: Path) -> None:
40 """
41 Write the string to the destination.
43 :param orig: the original string
44 :param dest: the destination
45 """
46 orig = str.rstrip(orig)
47 with dest.open_for_write() as output:
48 write_lines(map(str.rstrip, str.rstrip(orig).splitlines()), output)
49 logger("Wrote r-stripped string of originally "
50 f"{str.__len__(orig)} characters to {dest!r}, "
51 f"produced file of size {getsize(dest)} bytes.")
54def __get_sys_env() -> Mapping[str, str]:
55 """
56 Get the system environment variables in the current environment.
58 :return: A mapping of variable names to values.
59 """
60 base: dict[str, str] = dict(environ)
61 base.update(PYTHON_ENV)
62 return immutable_mapping(base)
65#: the environment that we will pass on
66SYS_ENV: Final[Mapping[str, str]] = __get_sys_env()
67del __get_sys_env
70class ProcessManager(GitManager):
71 """A manager for processes."""
73 def get_argument_file(self, name: str, prefix: str | None = None,
74 suffix: str | None = None) -> tuple[Path, bool]:
75 """
76 Create a file in the argument realm.
78 :param name: the ID for the file
79 :param prefix: the optional prefix
80 :param suffix: the optional suffix
81 :return: the file, plus a `bool` indicating whether it was just
82 created (`True`) or already existed (`False`)
83 """
84 return self.get_file(
85 "args", name, (str.strip(prefix) or None) if prefix else None,
86 (str.strip(suffix) or None) if suffix else None)
88 def filter_argument(self, arg: str) -> str | None:
89 """
90 Filter an argument to be passed to any given file.
92 This function can be used to rewire arguments of certain programs that
93 we want to invoke to specific files.
95 :param arg: the argument
96 :return: the filtered argument
97 """
98 arg = str.strip(arg)
99 if arg:
100 if arg.startswith("(?") and arg.endswith("?)"):
101 name: Final[str] = str.strip(arg[2:-2])
102 if str.__len__(name) <= 0:
103 raise ValueError(f"Invalid ID in {arg!r}.")
104 return self.get_argument_file(name)[0]
105 return arg
106 return None
108 def __execute(self, dest: Path,
109 command: str | Iterable[str],
110 working_dir: Path | None = None,
111 stdin: str | None = None) -> None:
112 """
113 Make a command and environment.
115 :param dest: the destination path
116 :param command: the command
117 :param working_dir: an optional working directory
118 :param stdin: the standard input for the program, or `None`
119 """
120 # process the command
121 cmd_lst: Final[list[str]] = [command] if isinstance(command, str)\
122 else list(command)
123 for i in range(list.__len__(cmd_lst) - 1, -1, -1):
124 cmd: str = str.strip(cmd_lst[i])
125 if str.__len__(cmd) <= 0:
126 del cmd_lst[i]
127 continue
129 # process the arguments
130 for i in range(list.__len__(cmd_lst) - 1, 0, -1):
131 cmd = self.filter_argument(cmd_lst[i])
132 if cmd is None:
133 del cmd_lst[i]
134 continue
135 cmd_lst[i] = cmd
137 if list.__len__(cmd_lst) <= 0:
138 raise ValueError(f"Invalid command {command!r}.")
140 # Now we need to fix the command if we are running inside a virtual
141 # environment. If we are running inside a virtual environment, it is
142 # necessary to use the same Python interpreter that was used to run
143 # texgit. We should also pass along all the Python-related
144 # environment parameters.
145 env: Mapping[str, str] = SYS_ENV
146 if str.lower(cmd_lst[0]).startswith("python3"):
147 cmd_lst[0] = PYTHON_INTERPRETER
148 env = PYTHON_ENV
150 # execute the command and capture the output
151 output: str = Command(
152 command=cmd_lst, working_dir=working_dir, env=env,
153 stdout=STREAM_CAPTURE, stdin=stdin).execute(True)[0]
155 replace: list[Path] = self._get_sensitive_paths()
156 replace.append(dest)
157 replace.sort(key=str.__len__, reverse=True)
158 for base_dir in replace: # fix the base path
159 output = replace_base_path(output, base_dir)
160 _write(output, dest)
162 def get_output(
163 self, name: str, command: str | Iterable[str],
164 repo_url: str | None = None,
165 relative_dir: str | None = None) -> Path:
166 """
167 Get the output of a certain command.
169 :param name: the name for the output
170 :param command: the command itself
171 :param repo_url: the optional repository URL
172 :param relative_dir: the optional directory inside the repository
173 where the command should be executed
174 :return: the path to the output and, if `repo_url` and `relative_dir`
175 were not `None`, then a URL pointing to the directory in the
176 repository, else `None`
177 """
178 self._check_open()
179 path, is_new = self.get_file("output", name)
180 if not is_new:
181 return path
183 if isinstance(repo_url, str):
184 repo_url = str.strip(repo_url) or None
185 elif repo_url is not None:
186 raise type_error(repo_url, "repo_url", (str, None))
188 if isinstance(relative_dir, str):
189 relative_dir = str.strip(relative_dir) or None
190 elif relative_dir is not None:
191 raise type_error(relative_dir, "relative_dir", (str, None))
192 if (repo_url is None) != (relative_dir is None):
193 raise ValueError(f"repo_url and relative_dir must either both be "
194 f"None or neither, but they are {repo_url!r} "
195 f"and {relative_dir!r}.")
196 working_dir: Path | None = None
197 if repo_url is not None:
198 working_dir = self.get_git_dir(repo_url, relative_dir).path
200 self.__execute(dest=path, command=command, working_dir=working_dir)
201 return path
203 def get_git_file(
204 self, repo_url: str, relative_file: str,
205 name: str | None = None,
206 command: str | Iterable[str] | None = None) -> GitPath:
207 """
208 Get a path to a postprocessed file from the given git repository.
210 :param repo_url: the repository url.
211 :param relative_file: the relative path to the file
212 :param name: the name for the output
213 :param command: the command itself
214 :return: a tuple of file and URL
215 """
216 gf: Final[GitPath] = super().get_git_file(repo_url, relative_file)
217 if command:
218 name = str.strip(name)
219 path, is_new = self.get_file("postprocessed", name)
220 if is_new:
221 self.__execute(dest=path, command=command,
222 stdin=gf.path.read_all_str())
223 else:
224 path = gf.path
225 return GitPath(path, gf.repo, gf.repo.make_url(gf.path), gf.basename)