packaging-scripts/packaging_scripts/pkgfiles.py

"""Helper functions for dealing with package files."""

import logging
import re
import shutil

from pathlib import Path

# ========== Constants ==========
# Some versions might have a colon (:)
# Match any pkgfile of any name with the .pkg.tar.* extension
PKGREGEX = r"^[\:\w.+/-]+\.pkg\.tar(\.\w+)?$"
# Match any sigfile of any name with the .pkg.tar.*.sig extension
SIGREGEX = r"^[\:\w.+/-]+\.pkg\.tar(\.\w+)?\.sig$"


# ========== Logging Setup ===========
syslog = logging.getLogger(__name__)


# ========== Functions ==========
def get(query=None, *, directory=None, signatures_only=False):
    """Retrieve all package-related files from the filesystem.

    :param query: names of package files to search for
    :type query: str
    :param directory: a directory to search in
    :type directory: str or path-like object
    :param signatures_only: include only signature files
    :type signatures_only: bool
    :yields: paths of package files
    """
    path = Path.cwd() if directory is None else Path(directory)
    filequery = f"*{query}*" if query else "*"

    if signatures_only:
        yield from _filter_by_regex(SIGREGEX, path.glob(filequery))
    else:
        yield from _filter_by_regex(PKGREGEX, path.glob(filequery))


def filter(iterable, *, signatures_only=False):
    """Retrieve package files from a predefined iterable.

    :param iterable: any iterable
    :type iterable: anything that returns str or path-like objects
    :param signatures_only: include only signature files
    :type signatures_only: bool
    :yields: paths of package files
    """
    if signatures_only:
        yield from _filter_by_regex(SIGREGEX, sorted(set(iterable)))
    else:
        yield from _filter_by_regex(PKGREGEX, sorted(set(iterable)))


def add(pkgfile, cachedir):
    """Add package file to the repository cache directory.

    :param pkg: path of package to add
    :type pkg: path-like object
    :param cachedir: cache directory to move package to
    :type cachedir: path-like object
    """
    syslog.info("Adding %s to %s", pkgfile, cachedir)
    shutil.move(pkgfile, cachedir / pkgfile.name)


def delete(pkg):
    """Remove package file.

    :param pkg: path of package to remove
    :type pkg: path-like object
    """
    pkg.unlink()
    syslog.info("Removed %s", pkg)


def _filter_by_regex(regex_expression, iterable):
    """Filter by regular expression.

    :param regex_expression: the expression to filter by
    :type regex_expression: str
    :param iterable: iterable to filter through
    :type iterable: str or path-like objects
    :yields: pathlib.Path objects to package files
    """
    for item in iterable:
        if re.match(regex_expression, str(item)):
            yield Path(item)