Source code for pacifica.dispatcher.models

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# pacifica-dispatcher: pacifica/dispatcher/models.py
#
# Copyright (c) 2019, Battelle Memorial Institute
# All rights reserved.
#
# See LICENSE for details.
"""
Models module.

This class contains the data structure models for the dispatcher.
"""
import abc
import json
import os
import typing

from cloudevents.model import Event
from jsonpath2.path import Path

from .exceptions import TransactionDuplicateAttributeError


# pylint: disable=too-few-public-methods
class PacificaModel(abc.ABC):
    """
    Abstract base class for all Dispatcher models.

    Contains method interface definitions for all model classes.
    """

    # pylint: disable=line-too-long
    @classmethod
    @abc.abstractmethod
    def from_cloudevents_model(cls, event: Event) -> typing.Union['PacificaModel', typing.List['PacificaModel']]:  # NOQA: E501
        """Abstract method for creating a model from a cloud event."""
        raise NotImplementedError()  # pragma: no cover
    # pylint: enable=line-too-long


class File(PacificaModel):
    """
    File model class.

    This holds all information about a file from a cloud event.
    """

    _id = None
    ctime = None
    encoding = None
    hashsum = None
    hashtype = None
    mimetype = None
    mtime = None
    name = None
    size = None
    subdir = None
    suspense_date = None

    def __init__(self, **attrs: typing.Dict[str, typing.Any]) -> None:
        """Build the file attributes from the dictionary keywords."""
        super(File, self).__init__()
        attr_names = [
            '_id', 'ctime', 'encoding', 'hashsum', 'hashtype', 'mimetype',
            'mtime', 'name', 'size', 'subdir', 'suspense_date'
        ]
        for name in attr_names:
            setattr(self, name, attrs.get(name, None))

    @classmethod
    def from_cloudevents_model(cls, event: Event) -> typing.List['File']:
        """Factory creating instances of File from a cloud event."""
        insts = []

        for match_data in Path.parse_str('$[*][?(@["destinationTable"] = "Files")]').match(event.data):
            inst = cls(**match_data.current_value)

            insts.append(inst)

        return insts

    @property
    def path(self) -> str:
        """Path property accessor."""
        if self.name is None:
            raise AttributeError('field \'name\' is None')
        if self.subdir is None:
            return self.name
        return os.path.join(self.subdir, self.name)


class Transaction(PacificaModel):
    """
    Transaction model clas.

    This class holds all the information about a transaction from a
    cloud event.
    """

    _id = None
    analytical_tool = None
    description = None
    instrument = None
    project = None
    submitter = None
    suspense_date = None

    def __init__(self, **attrs: typing.Dict[str, typing.Any]) -> None:
        """Build the transaction from the attrs in keywords."""
        super(Transaction, self).__init__()

        for name in ['_id', 'analytical_tool', 'description', 'instrument', 'project', 'submitter', 'suspense_date']:
            setattr(self, name, attrs.get(name, None))

    @classmethod
    def from_cloudevents_model(cls, event: Event) -> 'Transaction':
        """Factory creating a transaction class from a cloud event."""
        attrs = {}
        attr_names = [
            '_id', 'analytical_tool', 'description', 'instrument',
            'project', 'submitter', 'suspense_date'
        ]
        for name in attr_names:
            matches = Path.parse_str(
                '$[*][?(@["destinationTable"] = {0})]["value"]'.format(
                    json.dumps('Transactions.{0}'.format(name))
                )
            ).match(event.data)
            for match_data in matches:
                if name in attrs:
                    raise TransactionDuplicateAttributeError(event, name)

                attrs[name] = match_data.current_value

            if name not in attrs:
                attrs[name] = None

        return cls(**attrs)


class TransactionKeyValue(PacificaModel):
    """
    Transaction key value model.

    This holds the transaction key value pairs from a cloud event.
    """

    def __init__(self, **attrs: typing.Dict[str, typing.Any]) -> None:
        """Construct the attributes from keywords."""
        super(TransactionKeyValue, self).__init__()

        for name in ['key', 'value']:
            setattr(self, name, attrs.get(name, None))

    @classmethod
    def from_cloudevents_model(cls, event: Event) -> typing.List['TransactionKeyValue']:
        """Factory creating all the key value pairs for a cloud event."""
        insts = []

        for match_data in Path.parse_str('$[*][?(@["destinationTable"] = "TransactionKeyValue")]').match(event.data):
            inst = cls(**match_data.current_value)

            insts.append(inst)

        return insts


__all__ = ('PacificaModel', 'File', 'Transaction', 'TransactionKeyValue', )