Source code for forml.project._distribution

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

Project distribution.
import collections
import functools
import json
import logging
import pathlib
import re
import shutil
import string
import sys
import tempfile
import types
import typing
import zipfile

import forml
from forml import setup
from import asset

from . import _body

if typing.TYPE_CHECKING:
    from forml import project

LOGGER = logging.getLogger(__name__)

[docs]class Package(collections.namedtuple('Package', 'path, manifest')): """ForML artifact representing a complete project code together with all of its dependencies packaged for distribution. Args: path: File system path pointing to the package file. """ path: pathlib.Path manifest: 'project.Manifest' FORMAT = '4ml' COMPRESSION = zipfile.ZIP_DEFLATED PYSFX = re.compile(r'\.py[co]?$') def __new__(cls, path: typing.Union[str, pathlib.Path]): path = pathlib.Path(path) return super().__new__(cls, path.resolve(), def __getnewargs__(self): return tuple([self.path]) @classmethod def create( cls, source: typing.Union[str, pathlib.Path], manifest: 'project.Manifest', path: typing.Union[str, pathlib.Path], ) -> 'Package': """Create new package from the given source tree. Args: source: File system path to the root of directory tree to be packaged. manifest: Package manifest to be used. path: Target package file system path. Returns: Package instance. """ def writeall(level: pathlib.Path, archive: zipfile.ZipFile, root: typing.Optional[pathlib.Path] = None) -> None: """Recursive helper for adding directory tree content to a zip archive. Args: level: Level to be added. archive: zipfile instance opened for writing. root: Root of directory tree to be added. """ def valid(file: pathlib.Path) -> bool: """Check the item is valid package item candidate. Args: file: Item path to be validated. Returns: True if valid. """ return file != descriptor if not root: root = level for item in level.iterdir(): if not item.is_dir(): target = item.relative_to(root) if valid(target): archive.write(item, target) else: writeall(item, archive, root) descriptor = Manifest.path('.') with zipfile.PyZipFile(path, 'w', cls.COMPRESSION) as package: with tempfile.TemporaryDirectory() as temp: manifest.write(temp) package.write(Manifest.path(temp), descriptor) writeall(pathlib.Path(source), package) return cls(path) def install(self, path: typing.Union[str, pathlib.Path]) -> 'project.Artifact': """Return the project artifact based on this package mounted on the given path. Args: path: Target install path. Returns: Artifact instance. """ def uninstalled() -> bool: """Prune the existing path if not matching the target manifest. Returns: True if uninstalled. """ try: # to allow installing "virtual" packages this even ignores invalid self.path if == self.manifest: LOGGER.debug('Package %s already installed', self.path) return False except forml.InvalidError: pass if path.exists(): LOGGER.warning('Deleting existing content at %s', path) if path.is_dir(): shutil.rmtree(path) else: path.unlink() return True path = pathlib.Path(path) if path.exists() and path.samefile(self.path): LOGGER.debug('Same source-target install attempt ignored') elif uninstalled(): if not zipfile.is_zipfile(self.path): assert self.path.is_dir(), f'Expecting zip file or directory: {self.path}' LOGGER.debug('Installing directory based package %s to %s', self.path, path) shutil.copytree(self.path, path, ignore=lambda *_: {'__pycache__'}) else: with zipfile.ZipFile(self.path) as package: if all( for n in package.namelist()): # is a zip-safe LOGGER.debug('Installing zip-safe package %s to %s', self.path, path) path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(self.path.read_bytes()) else: LOGGER.debug('Extracting non zip-safe package %s to %s', self.path, path) package.extractall(path) return _body.Artifact(path, self.manifest.package, **self.manifest.modules)
[docs]class Manifest(collections.namedtuple('Manifest', 'name, version, package, modules')): """ForML distribution package metadata manifest. Args: name: Project name. version: Project release version. package: Full python package name containing the project principal components. modules: Individual project components mapping (if non-conventional). """ name: asset.Project.Key version: asset.Release.Key package: str modules: typing.Mapping[str, str] MODULE = f'__{Package.FORMAT}__' TEMPLATE = string.Template( '\n'.join( ( 'NAME = "$name"', 'VERSION = "$version"', 'PACKAGE = "$package"', 'MODULES = $modules', ) ) ) def __new__( cls, name: typing.Union[str, asset.Project.Key], version: typing.Union[str, asset.Release.Key], package: str, **modules: str, ): return super().__new__( cls, asset.Project.Key(name), asset.Release.Key(version), package, types.MappingProxyType(modules) ) def __getnewargs_ex__(self): return (, self.version, self.package), dict(self.modules) def __repr__(self): return f'{}-{self.version}' @classmethod @functools.cache def path(cls, base: typing.Union[str, pathlib.Path]) -> pathlib.Path: """Return the manifest module path. Args: base: Base path for the module. Returns: Module path. """ return pathlib.Path(base) / f'{cls.MODULE}.py' @classmethod def read(cls, path: typing.Optional[typing.Union[str, pathlib.Path]] = None) -> 'project.Manifest': """Load the manifest from the given path. Args: path: Path to read the manifest from (defaults to all of :data:`python:sys.path`). Returns: Manifest instance. Raises: forml.MissingError: Not a ForML package manifest. forml.InvalidError: Corrupt ForML package manifest. """ try: module = setup.isolated(cls.MODULE, path) manifest = cls(module.NAME, module.VERSION, module.PACKAGE, **module.MODULES) except ModuleNotFoundError as err: raise forml.MissingError(f'Unknown manifest ({err})') except AttributeError as err: raise forml.InvalidError(f'Invalid manifest ({err})') finally: if cls.MODULE in sys.modules: del sys.modules[cls.MODULE] return manifest def write(self, path: typing.Union[str, pathlib.Path]) -> None: """Write the manifest to the given path (directory). Args: path: Directory to write the manifest into. """ path = self.path(path) path.parent.mkdir(parents=True, exist_ok=True) with'w') as manifest: manifest.write( self.TEMPLATE.substitute(, version=self.version, package=self.package, modules=json.dumps(dict(self.modules)) ) )