Source code for

# -*- coding: utf-8 -*-

"""Transport functions for `Fraunhofer's OrientDB <>`_.

`Fraunhofer <>`_ hosts
an instance of `OrientDB <>`_ that contains BEL in a schema similar to
:mod:``. However, they include custom relations that do not come
from a controlled vocabulary, and have not made the schema, ETL scripts, or documentation available.

Unlike BioDati and BEL Commons, the Fraunhofer OrientDB does not allow for uploads, so only
a single function :func:`pybel.from_fraunhofer_orientdb` is provided by PyBEL.

import logging
from typing import Any, Iterable, Mapping, Optional
from urllib.parse import quote_plus

import requests
from pyparsing import ParseException

from .. import constants as pc
from ..parser import BELParser
from ..struct import BELGraph

__all__ = [

logger = logging.getLogger(__name__)

[docs]def from_fraunhofer_orientdb( # noqa:S107 database: str = 'covid', user: str = 'covid_user', password: str = 'covid', query: Optional[str] = None ) -> BELGraph: """Get a BEL graph from the Fraunhofer OrientDB. :param database: The OrientDB database to connect to :param user: The user to connect to OrientDB :param password: The password to connect to OrientDB :param query: The query to run. Defaults to the URL encoded version of ``select from E``, where ``E`` is all edges in the OrientDB edge database. Likely does not need to be changed, except in the case of selecting specific subsets of edges. Make sure you URL encode it properly, because OrientDB's RESTful API puts it in the URL's path. By default, this function connects to the ``covid`` database, that corresponds to the COVID-19 Knowledge Graph [0]_. If other databases in the Fraunhofer OrientDB are published and demo username/password combinations are given, the following table will be updated. +----------+------------+----------+ | Database | Username | Password | +==========+============+==========+ | covid | covid_user | covid | +----------+------------+----------+ The ``covid`` database can be downloaded and converted to a BEL graph like this: .. code-block:: python import pybel graph = pybel.from_fraunhofer_orientdb( database='covid', user='covid_user', password='covid', ) graph.summarize() However, because the source BEL scripts for the COVID-19 Knowledge Graph are available on `GitHub <>`_ and the authors pre-enabled it for PyBEL, it can be downloaded with ``pip install git+`` and used with the following python code: .. code-block:: python import covid19kg graph = covid19kg.get_graph() graph.summarize() .. warning:: It was initially planned to handle some of the non-standard relationships listed in the Fraunhofer OrientDB's `schema <>`_ in their OrientDB Studio instance, but none of them actually appear in the only network that is accessible. If this changes, please leave an issue at so it can be addressed. .. [0] Domingo-Fernández, D., *et al.* (2020). `COVID-19 Knowledge Graph: a computable, multi-modal, cause-and-effect knowledge model of COVID-19 pathophysiology <>`_. *bioRxiv* 2020.04.14.040667. """ graph = BELGraph(name='Fraunhofer OrientDB: {}'.format(database)) parser = BELParser(graph, skip_validation=True) results = _request_graphstore(database, user, password, select_query_template=query) for result in results: _parse_result(parser, result) return graph
def _parse_result(parser: BELParser, result: Mapping[str, Any]) -> None: citation_db, citation_id = pc.CITATION_TYPE_PUBMED, result.get('pmid') if citation_id is None: citation_db, citation_id = pc.CITATION_TYPE_PMC, result.get('pmc') if citation_id is None: if 'citation' in result: logger.warning('incorrect citation information for %s: %s', result['@rid'], result['citation']) else: logger.debug('no citation information for %s', result['@rid']) return parser.control_parser.clear() parser.control_parser.citation_db = citation_db parser.control_parser.citation_db_id = citation_id parser.control_parser.evidence = result['evidence'] parser.control_parser.annotations.update(result['annotation']) source = result['in']['bel'] relation = result['@class'] relation = RELATION_MAP.get(relation, relation) target = result['out']['bel'] statement = ' '.join([source, relation, target]) try: parser.parseString(statement) except ParseException: logger.warning('could not parse %s', statement) RELATION_MAP = { 'causes_no_change': pc.CAUSES_NO_CHANGE, 'positive_correlation': pc.POSITIVE_CORRELATION, 'negative_correlation': pc.NEGATIVE_CORRELATION, 'is_a': pc.IS_A, 'has_member': 'hasMember', 'has_members': 'hasMembers', 'has_component': 'hasComponent', 'has_components': 'hasComponents', } def _request_graphstore( database: str, user: str, password: str, count_query: Optional[str] = None, select_query_template: Optional[str] = None, page_size: int = 500, base: str = '' ) -> Iterable[Mapping[str, Any]]: """Make an API call to the OrientDB.""" if count_query is None: count_query = 'select count(@rid) from E' count_query = quote_plus(count_query) count_url = '{base}/{database}/sql/{count_query}'.format(base=base, database=database, count_query=count_query) count_res = requests.get(count_url, auth=(user, password)) count = count_res.json()['result'][0]['count'] logging.debug('fraunhofer orientdb has %d edges', count) if select_query_template is None: select_query_template = 'select from E order by @rid limit {limit} offset {offset}' offsets = count // page_size for offset in range(offsets + 1): select_query = select_query_template.format(limit=page_size, offset=offset * page_size) logger.debug('query: %s', select_query) select_query = quote_plus(select_query) select_url = '{base}/{database}/sql/{select_query}/{page_size}/*:1'.format( base=base, database=database, select_query=select_query, page_size=page_size, ) res = requests.get(select_url, auth=(user, password)) res_json = res.json() result = res_json['result'] yield from result