goblin.app

Complexity

Total Complexity

27

Size/Duplication

Total Lines	152
Duplicated Lines	0 %

Importance

Changes

0

16 Methods

Rating	Name	Duplication	Size	Complexity
A	Goblin.session()	0	9	1
A	Goblin.config_from_yaml()	0	2	1
A	Goblin.register_from_module()	0	9	4
A	Goblin.config()	0	3	1
A	Goblin.config_from_json()	0	7	1
A	Goblin.url()	0	4	1
A	Goblin.config_from_file()	0	7	1
B	Goblin.__init__()	0	19	5
A	Goblin.edges()	0	4	1
A	Goblin.vertex_properties()	0	4	1
A	Goblin.vertices()	0	4	1
A	Goblin.register()	0	13	5
A	Goblin.open()	0	18	1
A	Goblin.close()	0	2	1
A	Goblin.cluster()	0	3	1
A	Goblin.config_from_module()	0	2	1

"""goblin application class and class constructor"""

import collections
import importlib
import logging

import aiogremlin # type: ignore

from goblin import element, provider, session

logger = logging.getLogger(__name__)


# Main API classes
class Goblin:
    """
    Class used to encapsulate database connection configuration and generate
    database connections Used as a factory to create
    :py:class:`Session<goblin.session.Session>` objects.

    :param str url: Database url
    :param asyncio.BaseEventLoop loop: Event loop implementation
    :param dict features: Vendor implementation specific database features
    :param dict config: Config parameters for application
    """

    def __init__(self,
                 cluster,
                 *,
                 provider=provider.TinkerGraph,
                 get_hashable_id=None,
                 aliases=None):
        self._cluster = cluster
        self._loop = self._cluster._loop
        self._cluster = cluster
        self._vertices = collections.defaultdict(lambda: element.GenericVertex)
        self._edges = collections.defaultdict(lambda: element.GenericEdge)
        self._vertex_properties = {}
        self._provider = provider
        if not get_hashable_id:
            get_hashable_id = self._provider.get_hashable_id
        self._get_hashable_id = get_hashable_id
        if aliases is None:
            aliases = {}
        self._aliases = aliases

    @classmethod
    async def open(cls,
                   loop,
                   *,
                   provider=provider.TinkerGraph,
                   get_hashable_id=None,
                   aliases=None,
                   **config):
        # App currently only supports GraphSON 1
        # aiogremlin does not yet support providers
        cluster = await aiogremlin.Cluster.open(
            loop, aliases=aliases, **config)
        app = Goblin(
            cluster,
            provider=provider,
            get_hashable_id=get_hashable_id,
            aliases=aliases)
        return app

    @property
    def cluster(self):
        return self._cluster

    @property
    def config(self):
        return self.cluster.config

    @property
    def vertices(self):
        """Registered vertex classes"""
        return self._vertices

    @property
    def vertex_properties(self):
        """Registered vertex classes"""
        return self._vertex_properties

    @property
    def edges(self):
        """Registered edge classes"""
        return self._edges

    @property
    def url(self):
        """Database url"""
        return self._url

    def register(self, *elements):
        """
        Register user created Element classes.

        :param goblin.element.Element elements: User defined Element classes
        """
        for element in elements:
            if element.__type__ == 'vertex':
                self._vertices[element.__label__] = element
            if element.__type__ == 'edge':
                self._edges[element.__label__] = element
            if element.__type__ == 'vertexproperty':
                self._vertex_properties[element.__label__] = element

    def config_from_file(self, filename):
        """
        Load configuration from from file.

        :param str filename: Path to the configuration file.
        """
        self._cluster.config_from_file(filename)

    def config_from_yaml(self, filename):
        self._cluster.config_from_yaml(filename)

    def config_from_json(self, filename):
        """
        Load configuration from from JSON file.

        :param str filename: Path to the configuration file.
        """
        self._cluster.config_from_json(filename)

    def config_from_module(self, module):
        self._cluster.config_from_module(module)

    def register_from_module(self, module, *, package=None):
        if isinstance(module, str):
            module = importlib.import_module(module, package)
        elements = list()
        for item_name in dir(module):
            item = getattr(module, item_name)
            if isinstance(item, element.ElementMeta):
                elements.append(item)
        self.register(*elements)

    async def session(self, *, processor='', op='eval', aliases=None):
        """
        Create a session object.

        :returns: :py:class:`Session<goblin.session.Session>` object
        """
        remote_connection = await aiogremlin.DriverRemoteConnection.using(
            self._cluster, aliases=self._aliases)
        return session.Session(self, remote_connection, self._get_hashable_id)

    async def close(self):
        await self._cluster.close()