#!/usr/bin/env python
# -*- coding: utf-8 -*-
# pylint: disable-msg=C0103
################################################################################
# Copyright (c) 2006-2017 Franz Inc.
# All rights reserved. This program and the accompanying materials are
# made available under the terms of the MIT License which accompanies
# this distribution, and is available at http://opensource.org/licenses/MIT
################################################################################
from __future__ import absolute_import
from __future__ import unicode_literals
from builtins import object
from .value import Value, BNode, URI
from .literal import Literal, CompoundLiteral, RangeLiteral, GeoCoordinate
from .statement import Statement
[docs]class ValueFactory(object):
"""
A factory for creating URIs, blank nodes, literals and statements.
"""
BLANK_NODE_AMOUNT = 10
[docs] def __init__(self, store):
self.store = store
self.unusedBNodeIds = []
[docs] def getUnusedBNodeId(self):
if not self.unusedBNodeIds:
## retrieve a set of bnode ids (they include leading '_:', which we strip off later:
self.unusedBNodeIds = self.store.mini_repository.getBlankNodes(amount=ValueFactory.BLANK_NODE_AMOUNT)
return self.unusedBNodeIds.pop()[2:] ## strip off leading '_:'
[docs] def createBNode(self, nodeID=None):
"""
See :meth:`.RepositoryConnection.createBNode`.
"""
if not nodeID:
nodeID = self.getUnusedBNodeId()
return BNode(nodeID)
[docs] def createLiteral(self, value, datatype=None, language=None):
"""
See :meth:`.RepositoryConnection.createLiteral`.
"""
if isinstance(value, (tuple, list)) and len(value) == 2:
return self.createRange(value[0], value[1])
return Literal(value, datatype=datatype, language=language)
[docs] def createStatement(self, subject, predicate, _object, context=None):
"""
See :meth:`.RepositoryConnection.createStatement`.
"""
return Statement(subject, predicate, _object, context=context)
[docs] def createURI(self, uri=None, namespace=None, localname=None):
"""
See :meth:`.RepositoryConnection.createURI`.
"""
if namespace and not localname:
return URI(namespace=uri, localname=namespace)
else:
return URI(uri=uri, namespace=namespace, localname=localname)
#############################################################################
## Extension to RDF4J API
#############################################################################
[docs] def validateRangeConstant(self, term, predicate):
"""Validate an individual range constant"""
datatype = term.getDatatype()
if not datatype:
raise Exception('Illegal term in range expression "%s" needs to '
'have a datatype.' % term.getValue())
[docs] def validateCompoundLiteral(self, term, predicate):
"""
Check to see if range boundaries are mapped.
TODO: ADD VALIDATION FOR GEO TERMS
"""
if isinstance(term, RangeLiteral):
self.validateRangeConstant(term.lowerBound, predicate)
self.validateRangeConstant(term.upperBound, predicate)
elif isinstance(term, GeoCoordinate):
pass
[docs] def object_position_term_to_openrdf_term(self, term, predicate=None):
"""
If 'term' is a string, integer, float, etc, convert it to
a Literal term. Otherwise, if its a Value, just pass it through.
"""
if term is not None:
if isinstance(term, CompoundLiteral):
self.validateCompoundLiteral(term, predicate)
elif not isinstance(term, Value):
term = self.createLiteral(term)
return term
[docs] def createRange(self, lowerBound, upperBound):
"""
Create a compound literal representing a range from 'lowerBound' to 'upperBound'
"""
lowerBound = self.object_position_term_to_openrdf_term(lowerBound)
upperBound = self.object_position_term_to_openrdf_term(upperBound)
return RangeLiteral(lowerBound=lowerBound, upperBound=upperBound)
