Source code for goblin.app
"""Goblin application class and class constructor"""
import collections
import importlib
import logging
import aiogremlin
from goblin import element, provider, session
logger = logging.getLogger(__name__)
# Main API classes
[docs]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._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
[docs] @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 edges(self):
"""Registered edge classes"""
return self._edges
@property
def url(self):
"""Database url"""
return self._url
[docs] 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
[docs] 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)
[docs] def config_from_yaml(self, filename):
self._cluster.config_from_yaml(filename)
[docs] 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)
[docs] def config_from_module(self, module):
self._cluster.config_from_module(module)
[docs] 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)
[docs] 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)
[docs] async def close(self):
await self._cluster.close()