OwlCyberSecurity - MANAGER

Edit File: _impl.py

import json
import os
import sys
import tempfile
from contextlib import contextmanager
from os.path import abspath
from os.path import join as pjoin
from subprocess import STDOUT, check_call, check_output

from ._in_process import _in_proc_script_path

def write_json(obj, path, **kwargs):
    with open(path, 'w', encoding='utf-8') as f:
        json.dump(obj, f, **kwargs)

def read_json(path):
    with open(path, encoding='utf-8') as f:
        return json.load(f)

class BackendUnavailable(Exception):
    """Will be raised if the backend cannot be imported in the hook process."""
    def __init__(self, traceback):
        self.traceback = traceback

class BackendInvalid(Exception):
    """Will be raised if the backend is invalid."""
    def __init__(self, backend_name, backend_path, message):
        super().__init__(message)
        self.backend_name = backend_name
        self.backend_path = backend_path

class HookMissing(Exception):
    """Will be raised on missing hooks (if a fallback can't be used)."""
    def __init__(self, hook_name):
        super().__init__(hook_name)
        self.hook_name = hook_name

class UnsupportedOperation(Exception):
    """May be raised by build_sdist if the backend indicates that it can't."""
    def __init__(self, traceback):
        self.traceback = traceback

def default_subprocess_runner(cmd, cwd=None, extra_environ=None):
    """The default method of calling the wrapper subprocess.

This uses :func:`subprocess.check_call` under the hood.
    """
    env = os.environ.copy()
    if extra_environ:
        env.update(extra_environ)

check_call(cmd, cwd=cwd, env=env)

def quiet_subprocess_runner(cmd, cwd=None, extra_environ=None):
    """Call the subprocess while suppressing output.

This uses :func:`subprocess.check_output` under the hood.
    """
    env = os.environ.copy()
    if extra_environ:
        env.update(extra_environ)

check_output(cmd, cwd=cwd, env=env, stderr=STDOUT)

def norm_and_check(source_tree, requested):
    """Normalise and check a backend path.

Ensure that the requested backend path is specified as a relative path,
    and resolves to a location under the given source tree.

Return an absolute version of the requested path.
    """
    if os.path.isabs(requested):
        raise ValueError("paths must be relative")

abs_source = os.path.abspath(source_tree)
    abs_requested = os.path.normpath(os.path.join(abs_source, requested))
    # We have to use commonprefix for Python 2.7 compatibility. So we
    # normalise case to avoid problems because commonprefix is a character
    # based comparison :-(
    norm_source = os.path.normcase(abs_source)
    norm_requested = os.path.normcase(abs_requested)
    if os.path.commonprefix([norm_source, norm_requested]) != norm_source:
        raise ValueError("paths must be inside source tree")

return abs_requested

class BuildBackendHookCaller:
    """A wrapper to call the build backend hooks for a source directory.
    """

def __init__(
            self,
            source_dir,
            build_backend,
            backend_path=None,
            runner=None,
            python_executable=None,
    ):
        """
        :param source_dir: The source directory to invoke the build backend for
        :param build_backend: The build backend spec
        :param backend_path: Additional path entries for the build backend spec
        :param runner: The :ref:`subprocess runner <Subprocess Runners>` to use
        :param python_executable:
            The Python executable used to invoke the build backend
        """
        if runner is None:
            runner = default_subprocess_runner

self.source_dir = abspath(source_dir)
        self.build_backend = build_backend
        if backend_path:
            backend_path = [
                norm_and_check(self.source_dir, p) for p in backend_path
            ]
        self.backend_path = backend_path
        self._subprocess_runner = runner
        if not python_executable:
            python_executable = sys.executable
        self.python_executable = python_executable

@contextmanager
    def subprocess_runner(self, runner):
        """A context manager for temporarily overriding the default
        :ref:`subprocess runner <Subprocess Runners>`.

.. code-block:: python

hook_caller = BuildBackendHookCaller(...)
            with hook_caller.subprocess_runner(quiet_subprocess_runner):
                ...
        """
        prev = self._subprocess_runner
        self._subprocess_runner = runner
        try:
            yield
        finally:
            self._subprocess_runner = prev

def _supported_features(self):
        """Return the list of optional features supported by the backend."""
        return self._call_hook('_supported_features', {})

def get_requires_for_build_wheel(self, config_settings=None):
        """Get additional dependencies required for building a wheel.

:returns: A list of :pep:`dependency specifiers <508>`.
        :rtype: list[str]

.. admonition:: Fallback

If the build backend does not defined a hook with this name, an
            empty list will be returned.
        """
        return self._call_hook('get_requires_for_build_wheel', {
            'config_settings': config_settings
        })

def prepare_metadata_for_build_wheel(
            self, metadata_directory, config_settings=None,
            _allow_fallback=True):
        """Prepare a ``*.dist-info`` folder with metadata for this project.

:returns: Name of the newly created subfolder within
                  ``metadata_directory``, containing the metadata.
        :rtype: str

.. admonition:: Fallback

If the build backend does not define a hook with this name and
            ``_allow_fallback`` is truthy, the backend will be asked to build a
            wheel via the ``build_wheel`` hook and the dist-info extracted from
            that will be returned.
        """
        return self._call_hook('prepare_metadata_for_build_wheel', {
            'metadata_directory': abspath(metadata_directory),
            'config_settings': config_settings,
            '_allow_fallback': _allow_fallback,
        })

def build_wheel(
            self, wheel_directory, config_settings=None,
            metadata_directory=None):
        """Build a wheel from this project.

:returns:
            The name of the newly created wheel within ``wheel_directory``.

.. admonition:: Interaction with fallback

If the ``build_wheel`` hook was called in the fallback for
            :meth:`prepare_metadata_for_build_wheel`, the build backend would
            not be invoked. Instead, the previously built wheel will be copied
            to ``wheel_directory`` and the name of that file will be returned.
        """
        if metadata_directory is not None:
            metadata_directory = abspath(metadata_directory)
        return self._call_hook('build_wheel', {
            'wheel_directory': abspath(wheel_directory),
            'config_settings': config_settings,
            'metadata_directory': metadata_directory,
        })

def get_requires_for_build_editable(self, config_settings=None):
        """Get additional dependencies required for building an editable wheel.

:returns: A list of :pep:`dependency specifiers <508>`.
        :rtype: list[str]

.. admonition:: Fallback

If the build backend does not defined a hook with this name, an
            empty list will be returned.
        """
        return self._call_hook('get_requires_for_build_editable', {
            'config_settings': config_settings
        })

def prepare_metadata_for_build_editable(
            self, metadata_directory, config_settings=None,
            _allow_fallback=True):
        """Prepare a ``*.dist-info`` folder with metadata for this project.

:returns: Name of the newly created subfolder within
                  ``metadata_directory``, containing the metadata.
        :rtype: str

.. admonition:: Fallback

If the build backend does not define a hook with this name and
            ``_allow_fallback`` is truthy, the backend will be asked to build a
            wheel via the ``build_editable`` hook and the dist-info
            extracted from that will be returned.
        """
        return self._call_hook('prepare_metadata_for_build_editable', {
            'metadata_directory': abspath(metadata_directory),
            'config_settings': config_settings,
            '_allow_fallback': _allow_fallback,
        })

def build_editable(
            self, wheel_directory, config_settings=None,
            metadata_directory=None):
        """Build an editable wheel from this project.

:returns:
            The name of the newly created wheel within ``wheel_directory``.

.. admonition:: Interaction with fallback

If the ``build_editable`` hook was called in the fallback for
            :meth:`prepare_metadata_for_build_editable`, the build backend
            would not be invoked. Instead, the previously built wheel will be
            copied to ``wheel_directory`` and the name of that file will be
            returned.
        """
        if metadata_directory is not None:
            metadata_directory = abspath(metadata_directory)
        return self._call_hook('build_editable', {
            'wheel_directory': abspath(wheel_directory),
            'config_settings': config_settings,
            'metadata_directory': metadata_directory,
        })

def get_requires_for_build_sdist(self, config_settings=None):
        """Get additional dependencies required for building an sdist.

:returns: A list of :pep:`dependency specifiers <508>`.
        :rtype: list[str]
        """
        return self._call_hook('get_requires_for_build_sdist', {
            'config_settings': config_settings
        })

def build_sdist(self, sdist_directory, config_settings=None):
        """Build an sdist from this project.

:returns:
            The name of the newly created sdist within ``wheel_directory``.
        """
        return self._call_hook('build_sdist', {
            'sdist_directory': abspath(sdist_directory),
            'config_settings': config_settings,
        })

def _call_hook(self, hook_name, kwargs):
        extra_environ = {'PEP517_BUILD_BACKEND': self.build_backend}

if self.backend_path:
            backend_path = os.pathsep.join(self.backend_path)
            extra_environ['PEP517_BACKEND_PATH'] = backend_path

with tempfile.TemporaryDirectory() as td:
            hook_input = {'kwargs': kwargs}
            write_json(hook_input, pjoin(td, 'input.json'), indent=2)

# Run the hook in a subprocess
            with _in_proc_script_path() as script:
                python = self.python_executable
                self._subprocess_runner(
                    [python, abspath(str(script)), hook_name, td],
                    cwd=self.source_dir,
                    extra_environ=extra_environ
                )

data = read_json(pjoin(td, 'output.json'))
            if data.get('unsupported'):
                raise UnsupportedOperation(data.get('traceback', ''))
            if data.get('no_backend'):
                raise BackendUnavailable(data.get('traceback', ''))
            if data.get('backend_invalid'):
                raise BackendInvalid(
                    backend_name=self.build_backend,
                    backend_path=self.backend_path,
                    message=data.get('backend_error', '')
                )
            if data.get('hook_missing'):
                raise HookMissing(data.get('missing_hook_name') or hook_name)
            return data['return_val']