Skip to content
Snippets Groups Projects
setup_to_meta.py 8.09 KiB
import os
import json
import logging
import subprocess
from typing import List, Any, Dict, Callable, Optional

from zc.buildout.buildout import Buildout, Options

PYTHON_VERSION = '3.8'
logger = logging.getLogger("buildout/setup_to_meta")


def write_metafile(metadata: str, filepath: str):
    """
    Writes given metadata to file with given path.
    :param metadata: String containing conda recipe metadata to be written
    :param filepath: String containing the path to conda recipe file (meta.yaml)
    """
    logger.debug(f"Writing meta.yaml file at {filepath}...")
    try:
        os.makedirs(filepath[:-10])
    except FileExistsError:
        pass

    with open(filepath, 'w') as f:
        f.write(metadata)
    logger.debug("Done writing.")


class MetadataGenerator:
    """
    Uses given info extracted from setup.py file to fill out metadata template.
    """
    def __init__(self, setup: Dict[str, str], path: str):
        self.setup = setup
        self.path = path
    
    def fmt_ep(self) -> str:
        """
        Format entry points section of metadata.
        :return: Formatted string if entry points exists; else empty string.
        """
        ep_string = ''
        if 'entry_points' in self.setup.keys() and 'console_scripts' in self.setup['entry_points']:
            ep_string += 'entry_points:\n'
            for ep in self.setup['entry_points']['console_scripts']:
                ep_string += '    - {}\n'.format(ep)
            ep_string += '  '
        return ep_string

    def fmt_reqs(self) -> str:
        """
        Format requirements section of metadata.
        :return: Formatted string if requirements exists; else empty string.
        """
        reqs_string = ''
        reqs_list = ''
        reqs_string += 'requirements:\n'
        build_reqs = '  build:\n'
        run_reqs = '  run:\n'
        host_reqs = '  host:\n'
        reqs_list += '    - python={}\n'.format(PYTHON_VERSION)

        if 'install_requires' in self.setup.keys():
            for req in self.setup['install_requires']:
                reqs_list += '    - {}\n'.format(req)

        reqs_string += build_reqs + reqs_list + \
                       run_reqs + reqs_list + \
                       host_reqs + reqs_list + \
                       '\n'
        return reqs_string

    def fmt_test(self) -> str:
        """
        Format test section of metadata.
        NOTE: May need further tweaking to be smarter based on individual project
        needs. For now, it's pretty dumb.
        :return: Formatted string if tests_require exists; else empty string.
        """
        test_string = ''
        if 'tests_require' in self.setup.keys():
            test_string += (
                'test:\n'
                '  source_files:\n'
                '    - test/\n'
                '  requires:\n'
            )

            for req in self.setup['tests_require']:
                test_string += '    - {}\n'.format(req)

            test_string += (
                '  commands:\n'
                '    - pytest -vv --log-level=DEBUG --showlocals\n'
                '\n'
            )
        return test_string

    def generate(self) -> str:
        logger.debug(f"Generating meta.yaml file from {self.path}...")
        # Filter numpy etc. out of the requirements
        try:
            self.setup['install_requires'] = [req for req in self.setup['install_requires'] if req != 'numpy']
        except KeyError:
            pass

        name = self.setup['name']
        version = self.setup['version']
        entry_points = self.fmt_ep()
        pth = self.path.replace("./", "")
        requirements = self.fmt_reqs()
        test = self.fmt_test()
        lic = self.setup['license']
        summary = self.setup['description']
    
        with open('build/tools/metafile_template.txt', 'r') as f:
            metadata = f.read()

        logger.debug("Done generating.")
        return metadata.format(
            name = name,
            version = version,
            entry_points = entry_points,
            path = "../../../" + pth,
            requirements = requirements,
            test = test,
            license = lic,
            summary = summary
        )


def parse_setup(d: str) -> Dict[str, str]:
    """
    Function for running parse_setup.py on each directory with a setup.py file.
    NOTE: Contains a hack for getting parse_setup.py to run in each directory.
    :param d: Directory with a setup.py file.
    :return: Data collected from parse_setup.py.
    """
    logger.debug(f"Parsing setup.py at {d}...")
    subprocess.run(['cp', 'build/tools/parse_setup.py', d])
    os.chdir(d)
    proc = subprocess.run(['python3', 'parse_setup.py'], stdout=subprocess.PIPE)
    os.chdir(root)
    subprocess.run(['rm', '{}/parse_setup.py'.format(d)])

    logger.debug("Done parsing.")
    return json.loads(proc.stdout)


def get_outputs(names: List[str]) -> List[str]:
    """
    Generate list of metadata files that will be created.
    :param dirs: List of dirs of all subprojects with a setup.py file.
    :return: List of paths to output files as strings.
    """
    outputs = []
    for name in names:
        outputs.append("build/metadata/{}/meta.yaml".format(name))

    return outputs

def get_dirs() -> List[str]:
    """
    Finds all subdirectories containing setup.py files.
    :return: List of directories as strings.
    """
    logger.debug("Finding list of directories containing setup.py files...")
    find = subprocess.run([
        'find', '.', '-name', 'setup.py', '-not', '-path', './build/recipes/*'
    ], stdout=subprocess.PIPE)
    dirs = find.stdout.decode('utf-8').split('\n')
    dirs_cpy = dirs

    for i, d in enumerate(dirs_cpy):
        dirs[i] = d.replace('/setup.py', '')

    logger.debug("Done finding directories.")
    return dirs


def get_names(dirs: List[str]) -> List[str]:
    """
    Generate list of subproject names based on the rule that the name of the
    subproject directory will be the name of the subproject.
    :return: List of names as strings.
    """
    logger.debug("Getting list of names...")
    names = []
    for d in dirs:
        if d != '':
            name = d.split('/')[-1]

            if name == "archive":
                # Case with ./services/archive having special dir structure
                name = "services"
            names.append(name)

    logger.debug("Done getting list of names.")
    return names


def del_substrings(s: str, substrings: List[str]):
    """
    Function for deleting multiple substrings from a string.
    :param s: String to be modified.
    :param substrings: List of substrings to be targeted for deletion.
    :return: Modified string.
    """
    for replace in substrings:
        s = s.replace(replace, '')

    return s


root = os.getcwd()


class Recipe:
    """
    Buildout Recipe class.
    For more detailed information, see the link.
    http://www.buildout.org/en/latest/topics/writing-recipes.html
    """
    def __init__(self, buildout: Optional[Buildout], name: Optional[str], options: Options):
        """
        Initializes fields needed for recipe.
        :param buildout: (Boilerplate) Dictionary of options from buildout section
        of buildout.cfg
        :param name: (Boilerplate) Name of section that uses this recipe.
        :param options: (Boilerplate) Options of section that uses this recipe.
        """
        self.dirs = get_dirs()
        self.names = get_names(self.dirs)
        self.outputs = get_outputs(self.names)
        self.options = options

    def install(self) -> Any:
        """
        Install method that runs when recipe has components it needs to install.
        :return: Paths to files, as strings, created by the recipe.
        """
        for i, d in enumerate(self.dirs):
            if d != '':
                setup_data = parse_setup(d)
                metadata = MetadataGenerator(setup_data, d).generate()
                write_metafile(metadata, self.outputs[i])
                # Buildout-specific operation: pass created file into options.created()
                self.options.created(self.outputs[i])

        return self.options.created()

    # No special procedure for updating vs. installing
    update = install