req2flatpak

pyproject.toml

[tool.poetry]
name = "req2flatpak"
version = "0.1.0"
description = "Generates a flatpak build manifest to install python packages defined in requirements.txt files."
authors = ["johannesjh <[email protected]>"]
license = "MIT"
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.10"
packaging = "^21.3"
PyYAML = "^6.0"

[tool.poetry.extras]
yaml = ["PyYAML"]

[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

req2flatpak.py

import argparse
import json
import pathlib
import re
import shelve
import sys
import urllib.request
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from functools import cached_property
from itertools import product
from typing import List, Iterable, Optional, Union

import packaging
import pkg_resources
from packaging.tags import cpython_tags, _normalize_string, compatible_tags, Tag
from packaging.utils import parse_wheel_filename


class Arch(str, Enum):
    """Linux architectures supported in this script."""

    x86_64: str = ("linux-x86_64",)
    aarch64: str = "linux-aarch64"

    @classmethod
    def from_string(cls, arch: str):
        arch = re.sub(r"^linux-", "", arch)
        return cls.__members__[arch]


@dataclass(frozen=True, kw_only=True)
class PythonInterpreter:
    """Represents a python interpreter on 64 bit linux."""

    major: int
    minor: int
    arch: Arch

    @classmethod
    def from_string(cls, interpreter: str):
        """Constructor that parses a string representation such as '39-linux-x86_64'."""
        major, minor, arch = re.match(r"^(\d)(\d+)-linux-(.*)$", interpreter).groups()
        return PythonInterpreter(
            major=int(major), minor=int(minor), arch=Arch.from_string(arch)
        )

    @cached_property
    def tags(self) -> list[Tag]:
        """Returns a list of tags that are supported by this interpreter."""

        def _linux_platforms(arch: Arch):
            """Returns a list of python platform values for a given architecture."""
            # based on packaging.tags._linux_platforms
            linux = _normalize_string(arch)
            _, arch = linux.split("_", 1)
            yield from packaging._manylinux.platform_tags(linux, arch)
            yield from packaging._musllinux.platform_tags(arch)
            yield linux

        def _platform_tags():
            python_version = (self.major, self.minor)
            abis = [f"cp{self.major}{self.minor}"]
            platforms = _linux_platforms(self.arch)
            yield from cpython_tags(python_version, abis, platforms)
            yield from compatible_tags(python_version, abis[0], platforms)

        return list(_platform_tags())


@dataclass(frozen=True, kw_only=True)
class Download:
    """Represents a package download from Pypi"""

    filename: str
    url: str
    sha256: str

    @classmethod
    def from_pypi_json(cls, json: dict):
        """Constructor from pypi's json format parsed into a dict"""
        return cls(
            filename=json["filename"], url=json["url"], sha256=json["digests"]["sha256"]
        )

    @cached_property
    def is_wheel(self):
        return self.filename.endswith(".whl")

    @cached_property
    def is_sdist(self):
        return not self.is_wheel

    @cached_property
    def tags(self) -> list[Tag]:
        """Returns a list of tags that this download is compatible for"""
        # https://packaging.pypa.io/en/latest/utils.html#packaging.utils.parse_wheel_filename
        # https://packaging.pypa.io/en/latest/utils.html#packaging.utils.parse_sdist_filename
        if self.is_wheel:
            _, _, _, tags = parse_wheel_filename(self.filename)
            return list(tags)
        else:
            return []

    @cached_property
    def arch(self) -> Optional[Arch]:
        """Returns a wheel's target architecture, and None for sdists."""
        if self.is_wheel:
            tag = self.tags[0]
            if tag.platform.endswith("x86_64"):
                return Arch.x86_64
            if tag.platform.endswith("aarch64"):
                return Arch.aarch64
        else:
            return None

    def matches(self, platform_tag: str):
        if self.is_wheel:
            return platform_tag in self.tags
        else:
            return True

    def __eq__(self, other):
        """Two downloads are considered equal iff their url is equal."""
        return self.url == other.url


@dataclass(frozen=True, kw_only=True)
class Requirement:
    """Represents a requirement consisting of packagename and version."""

    package: str
    version: str


@dataclass(frozen=True, kw_only=True)
class Release(Requirement):
    """
    Represents a release as packagename, version and available downloads.
    Provides methods for selecting downloads that are compatible with specific python interpreters.
    """

    downloads: List[Download] = field(default_factory=list)

    def downloads_for(
        self, interpreter: PythonInterpreter, wheels_only=False, sdists_only=False
    ) -> Iterable[Download]:
        """Yields suitable downloads for a specific python interpreter, in preferred order."""
        cache: dict[Download, None] = {}
        for (platform_tag, download) in product(interpreter.tags, self.downloads):
            if download in cache:
                continue
            if wheels_only and not download.is_wheel:
                continue
            if sdists_only and not download.is_sdist:
                continue
            if download.matches(platform_tag):
                cache[download] = None
                yield download

    def wheel_for(self, interpreter: PythonInterpreter) -> Optional[Download]:
        """Returns the preferred wheel download for this release, for a specific python interpreter"""
        try:
            return next(self.downloads_for(interpreter, wheels_only=True))
        except StopIteration:
            return None

    def wheels_for(self, interpreters: List[PythonInterpreter]) -> List[Download]:
        """Returns a list of wheel downloads, for this release and for the specified target interpreters."""
        return list({self.wheel_for(interpreter=i) for i in interpreters})

    def sdist(self) -> Download:
        """Returns the source package download for this release"""
        try:
            return next(filter(lambda d: d.is_sdist, self.downloads))
        except StopIteration:
            return None

    def __eq__(self, other):
        """Two releases are considered equal iff packagename and version are equal."""
        return self.package == other.package and self.version == other.version


class PackageIndex(ABC):
    """Base class for querying package information from indices such as pypi"""

    @classmethod
    @abstractmethod
    def get_release(cls, req: Requirement) -> Release:
        """Queries release information from a package index."""
        pass

    @classmethod
    def get_releases(cls, reqs: Iterable[Requirement]) -> List[Release]:
        """Queries release information from a package index."""
        return [cls.get_release(req) for req in reqs]


# cache typealias
# this is meant for caching responses when querying package information
# a cache can either be a dict for in-memory caching, or a shelve.Shelf
Cache = Union[dict, shelve.Shelf]  # type: TypeAlias


class PyPi(PackageIndex):
    """Provides methods for querying package information from the PyPi package index."""

    _cache: Cache = {}

    @property
    def cache(self) -> Cache:
        return type(self)._cache

    @cache.setter
    def cache(self, val: Cache):
        type(self)._cache = val

    @classmethod
    def _query(cls, url) -> str:
        def _query_from_cache(url) -> Optional[str]:
            try:
                return cls._cache[url]
            except KeyError:
                return None

        def _query_from_pypi(url) -> str:
            json_string = urllib.request.urlopen((url)).read().decode("utf-8")
            cls._cache[url] = json_string
            return json_string

        return _query_from_cache(url) or _query_from_pypi(url)

    @classmethod
    def get_release(cls, req: Requirement) -> Release:
        """Queries pypi regarding available downloads for this requirement. Returns a release object."""
        url = f"https://pypi.org/pypi/{req.package}/{req.version}/json"
        json_string = cls._query(url)
        json_dict = json.loads(json_string)
        downloads = list(map(Download.from_pypi_json, json_dict["urls"]))
        return Release(package=req.package, version=req.version, downloads=downloads)


class RequirementsTxtParser:
    """
    Parses requirements.txt files in a very simple way:
    It expects all versions to be pinned.
    And it does not resolve dependencies.
    """

    # based on: https://stackoverflow.com/a/59971236
    # using functionality from pkg_resources.parse_requirements

    @classmethod
    def parse_string(cls, requirements_txt: str) -> List[Requirement]:
        """Parses requirements.txt string content into a list of Release objects."""

        def validate_requirement(req: pkg_resources.Requirement) -> None:
            assert (
                len(req.specs) == 1
            ), "Error parsing requirements: A single version numer must be specified."
            assert (
                req.specs[0][0] == "=="
            ), "Error parsing requirements: The exact version must specified as 'package==version'."

        def make_requirement(req: pkg_resources.Requirement) -> Requirement:
            validate_requirement(req)
            return Requirement(package=req.project_name, version=req.specs[0][1])

        reqs = pkg_resources.parse_requirements(requirements_txt)
        return [make_requirement(req) for req in reqs]

    @classmethod
    def parse_file(cls, file) -> List[Requirement]:
        """Parses a requirements.txt file into a list of Release objects."""
        if hasattr(file, "read"):
            req_txt = file.read()
        else:
            req_txt = pathlib.Path(file).read_text()
        return cls.parse_string(req_txt)


@dataclass(frozen=True, kw_only=True)
class FlatpakGenerator:
    """Generates a flatpak build module that installs specified python requirements."""

    interpreters: List[PythonInterpreter]
    package_index: PackageIndex = PyPi

    def _buildmodule_sources(self, reqs: Iterable[Requirement]) -> list:
        """Helper function, returns the 'sources' section of the flatpak build module"""
        releases = self.package_index.get_releases(reqs)
        downloads = {
            wheel
            for release in releases
            for wheel in release.wheels_for(self.interpreters)
        }

        def source(download: Download) -> dict:
            source = {"type": "file", "url": download.url, "sha256": download.sha256}
            if download.arch == Arch.x86_64:
                source["only-arches"] = ["x86_64"]
            elif download.arch == Arch.aarch64:
                source["only-arches"] = ["aarch64"]
            return source

        return [source(download) for download in downloads]

    def buildmodule_as_dict(
        self,
        reqs: Iterable[Requirement],
        module_name="python3-package-installation",
        pip_install_template: str = 'pip3 install --verbose --exists-action=i --no-index --find-links="file://${PWD}" --prefix=${FLATPAK_DEST} --no-build-isolation ',
    ) -> dict:
        """Returns a flatpak build module in the form of a python dict"""
        return {
            "name": module_name,
            "buildsystem": "simple",
            "build-commands": [
                pip_install_template + " ".join([req.package for req in reqs])
            ],
            "sources": self._buildmodule_sources(reqs),
        }

    def buildmodule_as_json(self, *args, **kwargs) -> str:
        module = self.buildmodule_as_dict(*args, **kwargs)
        return json.dumps(module, indent=4)

    def buildmodule_as_yaml(self, *args, **kwargs) -> str:
        try:
            import yaml
        except ImportError:
            exit('PyYAML modules is not installed. Run "pip install PyYAML"')
        module = self.buildmodule_as_dict(*args, **kwargs)
        return yaml.dump(module)


class Req2FlatpakCli:
    """The commandline interface for this scripe."""

    parser: argparse.ArgumentParser
    options: argparse.Namespace
    output: str

    @classmethod
    def setup_cli_options(cls) -> argparse.ArgumentParser:
        cls.parser = argparse.ArgumentParser(description=FlatpakGenerator.__doc__)
        cls.parser.add_argument(
            "requirements",
            nargs="*",
            help="One or more requirements can be specied as commandline arguments, e.g., 'pandas==1.4.4'.",
        )
        cls.parser.add_argument(
            "--requirements-file",
            "-r",
            nargs="?",
            type=argparse.FileType("r"),
            default=sys.stdin,
            help="Requirements can be read from a specified requirements.txt file or from stdin.",
        )
        cls.parser.add_argument(
            "--target-interpreters",
            "-t",
            nargs="+",
            help="Target interpreters can be specified as, e.g., '39-linux-x86_64'.",
        )
        cls.parser.add_argument(
            "--cache", action=argparse.BooleanOptionalAction, default=False
        )
        cls.parser.add_argument(
            "--outfile",
            "-o",
            nargs="?",
            type=argparse.FileType("w"),
            default=sys.stdout,
        )
        cls.parser.add_argument(
            "--format", "-f", default="json", choices=["json", "yaml"]
        )

    @classmethod
    def parse_cli_options(cls):
        cls.options = cls.parser.parse_args()

        # parse requirements
        if reqs := cls.options.requirements:
            cls.options.requirements = RequirementsTxtParser.parse_string("\n".join(reqs))
        elif req_file := cls.options.requirements_file:
            cls.options.requirements = RequirementsTxtParser.parse_file(req_file)
            del cls.options.requirements_file
        assert (
            len(cls.options.requirements) > 0
        ), "Error parsing requirements. At least one requirement must be specified."

        # parse target interpreters
        cls.options.target_interpreters = [
            PythonInterpreter.from_string(interpreter)
            for interpreter in cls.options.target_interpreters
        ]
        assert (
            len(cls.options.target_interpreters) > 0
        ), "Error parsing requirements. At least one requirement must be specified."

    @classmethod
    def generate_buildmodule(cls):
        generator = FlatpakGenerator(interpreters=cls.options.target_interpreters)
        if cls.options.format == "json":
            cls.output = generator.buildmodule_as_json(cls.options.requirements)
        elif cls.options.format == "yaml":
            cls.output = generator.buildmodule_as_yaml(cls.options.requirements)

    @classmethod
    def write_output(cls):
        if hasattr(file := cls.options.outfile, "write"):
            file.write(cls.output)
        else:
            pathlib.Path(file).write_text(cls.output)

    @classmethod
    def main(cls):
        # process commandline options
        cls.setup_cli_options()
        cls.parse_cli_options()

        # generate the flatpak build module
        if cls.options.cache:
            with shelve.open("pypi_cache") as cache:
                PyPi.cache = cache
                cls.generate_buildmodule()
        else:
            cls.generate_buildmodule()

        # write the build module as output
        cls.write_output()


if __name__ == "__main__":
    Req2FlatpakCli.main()