Source code for

from logging import getLogger
from rdflib.plugin import PluginException
import requests
from rdflib import Graph

from .datastore import DataStore
from oldman.model.manager import ModelManager
from import JSON_TYPES

[docs]class HttpDataStore(DataStore): """ Read only. No search feature. """ def __init__(self, schema_graph=None, cache_region=None, session=None): DataStore.__init__(self, ModelManager(schema_graph=schema_graph), cache_region) self._session = session if session is not None else requests.session() self._logger = getLogger(__name__) @property
[docs] def session(self): return self._session
def _get_by_id(self, id): r = self._session.get(id, headers=dict(Accept='text/turtle;q=1.0, ' 'application/rdf+xml;q=1.0, ' 'application/ld+json;q=.0.8, ' 'application/json;q=0.1')) if r.status_code != 200: self._logger.warn("Resource %s not retrieved (Status code: %d)." % (id, r.status_code)) return None hashless_iri = id.split('#')[0] content_type = r.headers.get('content-type') if content_type is None: self._logger.warn("No content-type returned for the resource %s." % id) return None elif content_type in JSON_TYPES: ctx_header = r.links.get("") if ctx_header is None: self._logger.warn("No context header given with the JSON representation of %s." % id) return None ctx_url = ctx_header["url"] resource_graph = Graph().parse(data=r.content, context=ctx_url, publicID=hashless_iri, format=content_type) else: try: resource_graph = Graph().parse(data=r.content, publicID=hashless_iri, format=content_type) except PluginException: self._logger.warn("Content-type %s is not supported. Impossible to get %s." % (content_type, id)) return None return self._new_resource_object(id, resource_graph)