PyXMake Developer Guide 1.0
PyXMake
Loading...
Searching...
No Matches
__poetry.py
1# -*- coding: utf-8 -*-
2# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
3# % Poetry wrapper module - Classes and functions %
4# %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
5"""
6Poetry configuration and management assistance wrapper.
7
8@note:
9Created on 09.08.2022
10
11@version: 1.0
12----------------------------------------------------------------------------------------------
13@requires:
14 -
15
16@change:
17 -
18
19@author: garb_ma [DLR-FA,STM Braunschweig]
20----------------------------------------------------------------------------------------------
21"""
22import os
23import copy
24import ast
25import sys
26import site
27import shlex
28import tomlkit
29import argparse
30import importlib
31import subprocess
32import logging
33import platform
34import posixpath
35
36from packaging.version import parse
37
38# Legacy support to build platform packages for Python 2.7 and below
39try:
40 from cleo.io.io import IO #@UnusedImport
41 from cleo.commands.command import Command as _Command #@UnusedImport
42except:
43 from builtins import object as IO #@UnusedImport @Reimport
44 from builtins import object as _Command #@UnusedImport @Reimport
45
46try:
47 from poetry.plugins.plugin import Plugin as _Plugin #@UnresolvedImport @UnusedImport
48 from poetry.plugins.application_plugin import ApplicationPlugin as _ApplicationPlugin #@UnresolvedImport #@UnusedImport
49except:
50 from builtins import object as _Plugin #@Reimport
51 from builtins import object as _ApplicationPlugin #@Reimport
52
53try:
54 # Import core version of poetry. Deprecated
55 from poetry.core.semver.version import Version as PoetryCoreVersion #@UnresolvedImport #@UnusedImport
56except:
57 # Import core version of poetry in version above 1.6+
58 from poetry.core.constraints.version import Version as PoetryCoreVersion #@UnresolvedImport @Reimport
59
60from packaging.utils import canonicalize_name
61
62from poetry.poetry import Poetry #@UnresolvedImport
63from poetry.utils._compat import metadata #@UnresolvedImport
64from poetry.repositories.installed_repository import InstalledRepository #@UnresolvedImport
65
66from .__git import main as _cli_git
67from .__gitlab import main as _cli_gitlab
68
69logger = logging.getLogger(__name__)
70
71class Plugin(_Plugin):
72 """
73 Poetry plugin interface
74 """
75 def activate(self, poetry, io):
76 # type: (Poetry, IO) -> None
77 """
78 Poetry plugin execution interface.
79 """
80 # Get current work directory.
81 cwd = os.getcwd()
82 ## Only apply this method when installing or building locally.
83 # Also skip rest of routine when root is explicitly excluded.
84 if not sys.argv[1] in ["build","install","version"] or "--no-root" in sys.argv: return
85 # Jump into directory of toml file.
86 os.chdir(os.path.dirname(str(poetry.file)))
87 # Create base command for subshell execution
88 command = " ".join([sys.executable,"-c",'"from PyXMake.Plugin.__poetry import setup; setup(silent_install=False)"'])
89 # Only during install command.
90 if sys.argv[1] in ["install"]:
91 # Use direct call on POSIX systems only.
92 if not os.name.lower() in ["nt"]: setup(silent_install=False)
93 else: subprocess.check_call(command, shell=True)
94 # Obtain all relevant information about this project.
95 project = poetry.local_config.get("name")
96 packages = poetry.local_config.get("packages")
97 # Auto-detect project name and package location, adding support for all POSIX systems.
98 if not packages:
99 if not os.path.exists("src"): poetry._package.packages = [{"include":str(project)}]
100 else: poetry._package.packages = [{"include":str(project), "from":"src"}]
101 # Check for dynamic versioning support during development
102 if parse(poetry.local_config.get("version")) == parse("0.0.0dev") and len(poetry._package.packages) == 1:
103 # Get the correct internal project name
104 project = poetry._package.packages[0].get("include")
105 # Deal with nested include statements in the main package definition
106 project = str(project).split(posixpath.sep)[0]
107 # Add the current project temporarily to the PYTHONPATH
108 try: sys.modules.pop(str(project))
109 except: pass
110 # Current project path takes precedence above all other paths
111 sys.path.insert(0,
112 os.path.normpath(
113 os.path.join(os.path.abspath(os.getcwd()),
114 poetry._package.packages[0].get("from",""))))
115 # Import programmatically
116 try:
117 importlib.import_module(str(poetry.local_config.get("name")))
118 project = poetry.local_config.get("name")
119 except: importlib.import_module(str(project))
120 finally: handle = sys.modules[str(project)]
121 # Remove the path from the overall system path
122 sys.path.pop(0); sys.modules.pop(str(project))
123 version = getattr(handle,"__version__","")
124 version = version or poetry.local_config.get("version")
125 # Do not add dev suffix when building the package.
126 if not sys.argv[1] in ["build","version"]: version += "dev"
127 poetry._package._version = PoetryCoreVersion.parse(version)
128 poetry._package._pretty_version = version
129 # Only during install and build command.
130 settings = {"packages":copy.deepcopy(poetry._package.packages)}
131 # Has no meanningful effect when no build script is given.
132 if sys.argv[1] in ["install","build"]: build(**settings)
133 # Jump back to initial working directory
134 os.chdir(cwd)
135 pass
136
137class ApplicationPlugin(_ApplicationPlugin):
138 """
139 Poetry application plugin interface
140 """
141 class Command(_Command):
142 """
143 Poetry application command interface
144 """
145 # Fully qualified names of the command (default)
146 name = "housekeeping"
147 # Additional supported commands
148 supported = ["housekeeping", "release"]
149 def handle(self): _cli_gitlab(method=self.name)
150 @classmethod
151 def factory(cls): return cls()
152
153 def run(self, application):
154 """
155 Poetry application command line interface
156 """
157 try:
158 from PyXMake.Tools import Utility #@UnusedImport
159 scripts = {"poetry-auto-cli":"default"};
160 except ImportError: scripts = {}
161 # Default variables
162 result = -1
163 scripts.update(application.poetry.local_config.get("scripts",{}))
164 scripts = scripts.keys();
165 # Check if a valid configuration can be fetched from the supplied TOML file. Defaults to False.
166 toml = str(application.poetry.file)
167 # Scan the TOML file for potential supported candidates. False positives are handled later.
168 configuration = list(tomlkit.parse(open(toml).read())["tool"].keys())
169 configuration += list(tomlkit.parse(open(toml).read())["tool"].get("pyxmake",{}).keys())
170 # Remove duplicate entries from the list
171 configuration = list(set(configuration))
172 # Only execute this command when first trailing parameter does not start with a dash
173 if len(sys.argv) >= 3 and not sys.argv[2].startswith("-") and not any([x in sys.argv for x in scripts]):
174 # Fetch all possible shims
175 commands = [" ".join([x] + sys.argv[2:]) for x in scripts]
176 for command in commands:
177 try:
178 ## Check if a valid configuration is present in the local toml file.
179 # Parse its path if true.
180 if sys.argv[2] in configuration: os.environ["pyx_poetry_config"] = toml
181 # Try if a shim works. Fail gracefully.
182 result = subprocess.check_call(command, stderr=sys.stderr, stdout=sys.stdout, shell=True);
183 if result == 0: break
184 # Ignore all erros here.
185 except Exception: pass
186 # The command was executed sucessfully. Terminate the process.
187 if result == 0: sys.exit()
188 pass
189
190 def activate(self, application):
191 """
192 Poetry application registration interface
193 """
194 try:
195 # Verify that the current virtual environment can be accessed.
196 output, _ = subprocess.Popen(["poetry","env","info","-p","--no-plugins"], stdout=subprocess.PIPE).communicate(); path = output.decode().strip();
197 os.environ["PATH"] = os.pathsep.join([os.path.join(path,x) for x in next(os.walk(path))[1] if x not in os.getenv("PATH","")] + [os.getenv("PATH","")])
198 except: pass
199 # Check if the command can be shimmed
200 if sys.argv[1] in ["run"]: self.run(application)
201 # Check if the given CLI options is supported.
202 if sys.argv[1] in self.CommandCommand.supported: setattr(self.CommandCommand,"name",str(sys.argv[1]))
203 # Register CLI command.
204 if hasattr(self.CommandCommand,"name"): application.command_loader.register_factory(self.CommandCommand.name, self.CommandCommand.factory)
205 pass
206
207@classmethod
208def load(cls, env=None, with_dependencies=False):
209 """
210 Load installed packages.
211 """
212 from pathlib import Path
213
214 from poetry.utils.env import Env as sys_env #@UnresolvedImport
215 from poetry.core.packages.dependency import Dependency #@UnresolvedImport
216
217 from dulwich.errors import NotGitRepository
218
219 repo = cls()
220 seen = set()
221 skipped = set()
222
223 if not env: env = sys_env
224 for entry in reversed(env.sys_path):
225 if not entry.strip():
226 logger.debug(
227 "Project environment contains an empty path in <c1>sys_path</>,"
228 " ignoring."
229 )
230 continue
231
232 for distribution in sorted(
233 metadata.distributions(path=[entry]),
234 key=lambda d: str(d._path), # type: ignore[attr-defined]
235 ):
236 path = Path(str(distribution._path)) # type: ignore[attr-defined]
237
238 if path in skipped:
239 continue
240
241 name = distribution.metadata.get("name") # type: ignore[attr-defined]
242 if name is None:
243 logger.warning(
244 (
245 "Project environment contains an invalid distribution"
246 " (<c1>%s</>). Consider removing it manually or recreate"
247 " the environment."
248 ),
249 path,
250 )
251 skipped.add(path)
252 continue
253
254 name = canonicalize_name(name)
255
256 if name in seen:
257 continue
258
259 try: package = cls.create_package_from_distribution(distribution, env)
260 except NotGitRepository: continue
261
262 if with_dependencies:
263 for require in distribution.metadata.get_all("requires-dist", []):
264 dep = Dependency.create_from_pep_508(require)
265 package.add_dependency(dep)
266
267 seen.add(package.name)
268 repo.add_package(package)
269
270 return repo
271
272@classmethod
273def install(cls, path, hash_, size):
274 # type: (str, str, str) -> "RecordEntry"
275 r"""
276 Build a RecordEntry object, from values of the elements.
277
278 Typical usage::
279
280 for row in parse_record_file(f):
281 record = RecordEntry.from_elements(row[0], row[1], row[2])
282
283 Meaning of each element is specified in :pep:`376`.
284
285 :param path: first element (file's path)
286 :param hash\_: second element (hash of the file's contents)
287 :param size: third element (file's size in bytes)
288 :raises InvalidRecordEntry: if any element is invalid
289 """
290 from typing import Optional #@UnusedImport
291 from installer.records import Hash, InvalidRecordEntry
292 # Validate the passed values.
293 issues = []
294 # Path can be empty
295 if not path.strip():
296 issues.append("`path` cannot be empty")
297 # Hash can be empty
298 if hash_.strip():
299 try:
300 hash_value = Hash.parse(hash_) # type: Optional[Hash]
301 except ValueError:
302 issues.append("`hash` does not follow the required format")
303 else:
304 hash_value = None
305 # Size can be empty
306 if size.strip():
307 try:
308 size_value = int(size) # type: Optional[int]
309 except ValueError:
310 issues.append("`size` cannot be non-integer")
311 else:
312 size_value = None
313 # Issues must be empty. Otherwise the installation process using poetry fails.
314 if issues:
315 raise InvalidRecordEntry(elements=(path, hash_, size), issues=issues)
316 # Return the class
317 return cls(path=path, hash_=hash_value, size=size_value)
318
319def setup(**kwargs):
320 """
321 Poetry installation helper interface
322 """
323 # Bring everything up-to-date
324 if not kwargs.get("silent_install",True): _cli_git(method="update")
325 # Check if the current path supports Poetry
326 if os.path.exists("pyproject.toml") and kwargs.get("silent_install",True):
327 with open("pyproject.toml", "r") as f: content = f.readlines()
328 project = [line.split("=")[-1].strip() for line in content if line.split("=")[0].strip() in ["name"]][0]
329 packages = [line for line in content if line.split("=")[0].strip() in ["packages"]]
330 # Script relies on auto-detect, which will fail on all POSIX systems.
331 if not packages:
332 # Rename old configuration file
333 os.replace("pyproject.toml","pyproject_old.toml")
334 with open("pyproject.toml", "w") as f:
335 # Copy content of old configuration file
336 for line in content:
337 f.write(line)
338 # Add non PEP8 project name explicitly
339 if all([x in line for x in ["name",project]]):
340 # Check for layouts.
341 if not os.path.exists("src"): f.write('packages = [{include=%s}]\n' % str(project))
342 else: f.write('packages = [{include=%s, from="src"}]\n' % str(project))
343 # Install the project regardless of change using the supplied TOML file
344 subprocess.call(shlex.split("poetry install", posix=not os.name.lower() in ["nt"]), stderr=sys.stderr, stdout=sys.stdout)
345 # Clean up
346 try:
347 # Replace the old configuration file
348 os.replace("pyproject_old.toml","pyproject.toml")
349 # Delete the temporary configuration file
350 os.remove("pyproject_old.toml")
351 except: pass
352 pass
353
354def build(*args, **kwargs):
355 """
356 Poetry build compatibility helper interface
357 """
358 # Import build system dependencies
359 from poetry import core #@UnresolvedImport
360 # Read content of pyproject file
361 packages = kwargs.get("packages",[])
362 if os.path.exists("pyproject.toml"):
363 with open("pyproject.toml", "r") as f: content = f.readlines()
364 else: content = str("")
365 # Check if a poetry file can be read
366 if not packages and content:
367 packages = [line for line in content if line.split("=")[0].strip() in ["packages"]]
368 config = next(iter(packages[0].split("=",1)[1:]));
369 config = config.strip(); config = config.replace("=",":")
370 config = config.replace('from','"from"'); config = config.replace('include','"include"')
371 packages = ast.literal_eval(config)
372 else: packages.append(dict())
373 # We do not build a platform specific wheel. Skipping rest of routine.
374 if not any([str("tool.poetry.build") in line for line in content]): return
375 # Clean up the initial workspace
376 _ = [os.remove(x) for x in os.listdir(os.getcwd()) if x.startswith("1.0")]
377 ## Recover the original path while running poetry install/build
378 sys.path.extend([os.path.join(x,"site-packages") for x in sys.path if "lib" in x.lower()])
379 sys.path.extend([os.path.join(os.path.dirname(x),"src","PyXMake","src") for x in sys.path if "lib" in x.lower()])
380 ## This handles all virtual environments within the current project.
381 venv = os.path.abspath(os.path.join(os.getcwd(),".venv"))
382 if os.path.exists(venv):
383 # Add all binaries
384 os.environ.update({"PATH":os.pathsep.join([os.getenv("PATH")]+[os.path.join(venv,"bin")])})
385 # Only meanningful if a virtual environment exists
386 sys.path.extend([os.path.join(venv,"src","PyXMake","src")])
387 sys.path.extend([os.path.abspath(os.path.join(venv,"lib",os.listdir(os.path.join(venv,"lib"))[0] if str(platform.system()).lower() in ["linux"] else "","site-packages"))])
388 # Create a user site directory if not already present
389 try:
390 child = subprocess.Popen([sys.executable,"-m","poetry","run","python","-c","import site; print(site.getusersitepackages())"], stdin=subprocess.PIPE, stderr=subprocess.PIPE, stdout=subprocess.PIPE)
391 usersitepackages = child.communicate()[0].decode().strip()
392 # Overwrite if command was not successful
393 if child.returncode != 0: raise RuntimeError
394 # If output cannot be recovered. Fail back to system implementation
395 if not usersitepackages: usersitepackages = site.getusersitepackages()
396 # Can happen when poetry is no longer in path
397 except: usersitepackages = site.getusersitepackages()
398 # Bugfix in poetry core for some python versions.
399 output = packages[0].get("from","") or packages[0].get("include","")
400 # Create a user site directory if not already present
401 if not os.path.exists(usersitepackages): os.makedirs(usersitepackages)
402 # Remove all outdated, deprecated files from the previous run
403 _ = [os.remove(os.path.abspath(os.path.join(usersitepackages,x))) for x in os.listdir(usersitepackages) if x.startswith("pyx_poetry_build_extension")]
404 # Write all current paths to pth files
405 for index, path in enumerate(sys.path,0):
406 # Skip empty paths
407 if not os.path.exists(path): continue
408 # Everything poetry related has to go
409 if str("poetry") in path.split("site-packages")[-1]: continue
410 # Add all paths into a seperate file
411 with open(os.path.join(usersitepackages,"pyx_poetry_build_extension_%s.pth" % str(index)),"w+") as f: f.write(path)
412 # Check if the output directory is already a valid package. Do not modify the output directory
413 if not os.path.exists(os.path.join(output,"__init__.py")):
414 # Loop over all files in the output directory
415 for x in os.listdir(output):
416 # Ignore all folders in the output directory.
417 if os.path.isdir(os.path.join(output,x)): continue
418 # Ignore generated files
419 try: subprocess.check_call(shlex.split("git ls-files --error-unmatch %s" % x,posix=not os.name.lower() in ["nt"]),
420 stdout=open(os.devnull, 'wb'), stderr=open(os.devnull, 'wb'), cwd=output)
421 except: continue
422 # Assume an unchanged index to avoid accidental updated to the repo
423 command = "git update-index --assume-unchanged %s" % x
424 # Only valid for files
425 subprocess.call(shlex.split(command,posix=not os.name.lower() in ["nt"]), cwd=output)
426 # Removes initial file. This is done only once.
427 if len(os.listdir(output)) <= 1: os.remove(os.path.join(output,x))
428 # We have an incomplete setup
429 if not os.path.exists(os.path.join(output,"__init__.py")):
430 # Add additional init file if not already present. Only required in later versions of Poetry. Never overwrite an existing one.
431 if parse(core.__version__) >= parse("1.6.1"): open(os.path.join(output,"__init__.py"),"w+")
432 # Add also a dummy init file to the output directory when running on windows or linux with interpreter version above 2.7
433 elif str(platform.system()).lower() in ["linux","windows"] and parse(".".join([str(x) for x in sys.version_info[:2]])) > parse("2.6"): open(os.path.join(output,"__init__.py"),"w+")
434
435def main(**kwargs):
436 """
437 Main command line parser.
438 """
439 if not kwargs.get("method",""):
440 parser = argparse.ArgumentParser(description='CLI wrapper options for Poetry.')
441 parser.add_argument('method', metavar='option', type=str, nargs=1,
442 help='An option identifier. Adding one new option named <setup>. All other CLI arguments are directly parsed to Poetry.')
443 # Select method
444 args, _ = parser.parse_known_args()
445 method = kwargs.get("method",str(args.method[0]))
446 else: method = kwargs.get("method")
447 # Execute wrapper within this script or parse the command to poetry
448 command = " ".join(["poetry"] + sys.argv[1:])
449 if not method in globals() or ( len(sys.argv[1:]) > 1 and not method in globals() ):
450 subprocess.call(shlex.split(command,posix=not os.name.lower() in ["nt"]),stderr=sys.stderr, stdout=sys.stdout)
451 else: globals()[method](**kwargs)
452 pass
453
454# Apply shim to poetry if version is sufficient.
455if parse(Poetry.VERSION) > parse("1.4.0"):
456 setattr(InstalledRepository,"load",load)
457
458try:
459 # Apply shim to installer package
460 from installer import __version__ as VERSION
461 from installer.records import RecordEntry
462 # Only apply shim for latest versions
463 if parse(VERSION) >= parse("0.7.0"):
464 setattr(RecordEntry, "from_elements", install)
465except ImportError: pass
466
467if __name__ == "__main__":
468 pass
activate(self, poetry, io)
Definition __poetry.py:75
load(cls, env=None, with_dependencies=False)
Definition __poetry.py:208
install(cls, path, hash_, size)
Definition __poetry.py:273
build(*args, **kwargs)
Definition __poetry.py:354
main()
Provide a custom error handler for the interruption event triggered by this wrapper to prevent multip...
Definition __init__.py:62
Module containing basic functionalities defined for convenience.
Definition __init__.py:1