PyCS/pycs/database/LabelProvider.py

import json
import re
import typing as T

from pathlib import Path

from pycs import db
from pycs.database.base import NamedBaseModel
from pycs.interfaces.LabelProvider import LabelProvider as LabelProviderInterface


class LabelProvider(NamedBaseModel):
    """
    DB model for label providers
    """

    description = db.Column(db.String)
    root_folder = db.Column(db.String, nullable=False)
    configuration_file = db.Column(db.String, nullable=False)

    # relationships to other models
    projects = db.relationship("Project", backref="label_provider", lazy="dynamic")

    # contraints
    __table_args__ = (
        db.UniqueConstraint('root_folder', 'configuration_file'),
    )

    serialize_only = NamedBaseModel.serialize_only + (
        "description",
        "root_folder",
        "configuration_file",
    )

    @classmethod
    def discover(cls, root: T.Union[Path, str]):
        """
            searches for label providers under the given path
            and stores them in the database
        """

        for folder, conf_path in _find_files(root):
            with open(conf_path) as conf_file:
                config = json.load(conf_file)

            provider, _ = cls.get_or_create(
                root_folder=str(folder),
                configuration_file=conf_path.name
            )

            provider.name = config['name']

            # returns None if not present
            provider.description = config.get('description')

            provider.flush()
        db.session.commit()

    @property
    def root(self) -> Path:
        """ returns the root folder as Path object """
        return Path(self.root_folder)

    @property
    def configuration_file_path(self) -> str:
        """ return the configuration file as Path object """
        return str(self.root / self.configuration_file)

    def load(self) -> LabelProviderInterface:
        """
        load configuration.json and create an instance from the included code object

        :return: LabelProvider instance
        """
        # load configuration.json
        with open(self.configuration_file_path) as configuration_file:
            configuration = json.load(configuration_file)

        # load code
        code_path = str(self.root / configuration['code']['module'])
        module_name = code_path.replace('/', '.').replace('\\', '.')
        class_name = configuration['code']['class']

        imported_module = __import__(module_name, fromlist=[class_name])
        class_attr = getattr(imported_module, class_name)

        # return instance
        return class_attr(self.root_folder, configuration)


def _find_files(root: T.Union[Path, str], config_regex=re.compile(r'^configuration(\d+)?\.json$')):
    """ generator for config files found under the given path """

    for folder in Path(root).glob('*'):
        # list files
        for file_path in folder.iterdir():

            # filter configuration files
            if not file_path.is_file():
                continue

            if config_regex.match(file_path.name) is None:
                continue

            # yield element
            yield folder, file_path