本文整理汇总了Python中rdflib.graph.ConjunctiveGraph类的典型用法代码示例。如果您正苦于以下问题:Python ConjunctiveGraph类的具体用法?Python ConjunctiveGraph怎么用?Python ConjunctiveGraph使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了ConjunctiveGraph类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_file
def test_file(self):
if self.html5lib_installed():
g = ConjunctiveGraph()
g.parse(location='test/rdfa/oreilly.html',
format='rdfa',
lax=True)
self.assertEqual(len(g), 77)
开发者ID:alcides,项目名称:rdflib,代码行数:7,代码来源:non_xhtml.py
示例2: test_url
def test_url(self):
if self.html5lib_installed():
g = ConjunctiveGraph()
g.parse(location='http://oreilly.com/catalog/9780596516499/',
format='rdfa',
lax=True)
self.assertTrue(len(g) > 0)
开发者ID:alcides,项目名称:rdflib,代码行数:7,代码来源:non_xhtml.py
示例3: TestSparqlJsonResults
class TestSparqlJsonResults(unittest.TestCase):
def setUp(self):
self.graph = ConjunctiveGraph()
self.graph.parse(StringIO(test_data), format="n3")
def _query_result_contains(self, query, correct):
results = self.graph.query(query)
result_json = json.loads(results.serialize(format='json'))
msg = "Expected:\n %s \n- to contain:\n%s" % (result_json, correct)
self.failUnless(result_json["head"]==correct["head"], msg)
result_bindings = sorted(result_json["results"]["bindings"])
correct_bindings = sorted(correct["results"]["bindings"])
msg = "Expected:\n %s \n- to contain:\n%s" % (result_bindings, correct_bindings)
self.failUnless(result_bindings==correct_bindings, msg)
testOptional = make_method('optional')
testWildcard = make_method('wildcard')
testUnion = make_method('union')
testUnion3 = make_method('union3')
testSelectVars = make_method('select_vars')
testWildcardVars = make_method('wildcard_vars')
开发者ID:pombredanne,项目名称:rdfextras,代码行数:29,代码来源:test_sparql_json_results.py
示例4: TestIssue06
class TestIssue06(unittest.TestCase):
debug = False
sparql = True
def setUp(self):
self.graph = ConjunctiveGraph()
self.graph.parse(data=testgraph, publicID="testgraph")
def test_issue_6(self):
query = """
PREFIX ex: <http://temp.example.org/terms/>
PREFIX loc: <http://simile.mit.edu/2005/05/ontologies/location#>
PREFIX rdfs: <http://www.w3.org/2000/01/rdf-schema#>
PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
SELECT *
WHERE {
{?event ex:date ?date .
FILTER (xsd:date(?date) >= xsd:date("2007-12-31") && xsd:date(?date) <= xsd:date("2008-01-11"))}
UNION
{?event ex:starts ?start; ex:finishes ?end .
FILTER (xsd:date(?start) >= xsd:date("2008-01-02") && xsd:date(?end) <= xsd:date("2008-01-10"))}
}
ORDER BY ?event
"""
self.graph.query(query, DEBUG=False)
开发者ID:RDFLib,项目名称:rdfextras,代码行数:28,代码来源:test_issue_06.py
示例5: GraphAggregates2
class GraphAggregates2(unittest.TestCase):
known_issue = True
sparql = True
def setUp(self):
memStore = plugin.get('IOMemory',Store)()
self.graph1 = Graph(memStore,URIRef("http://example.com/graph1"))
self.graph2 = Graph(memStore,URIRef("http://example.com/graph2"))
self.graph3 = Graph(memStore,URIRef("http://example.com/graph3"))
for n3Str,graph in [(testGraph1N3,self.graph1),
(testGraph2N3,self.graph2),
(testGraph3N3,self.graph3)]:
graph.parse(StringIO(n3Str),format='n3')
self.graph4 = Graph(memStore,RDFS)
self.graph4.parse(RDFS.uri)
self.G = ConjunctiveGraph(memStore)
def testAggregateSPARQL(self):
print sparqlQ
rt = self.G.query(sparqlQ)
assert len(rt) > 1
#print rt.serialize(format='xml')
LOG_NS = Namespace(u'http://www.w3.org/2000/10/swap/log#')
rt=self.G.query(sparqlQ2,initBindings={u'?graph' : URIRef("http://example.com/graph3")})
#print rt.serialize(format='json')
assert rt.serialize('python')[0] == LOG_NS.N3Document,repr(list(rt.serialize('python')))
开发者ID:alcides,项目名称:rdflib,代码行数:29,代码来源:test_aggregate_graphs.py
示例6: GraphAggregates2
class GraphAggregates2(unittest.TestCase):
def setUp(self):
memStore = plugin.get('SQLAlchemy', Store)(
identifier="rdflib_test", configuration=Literal("sqlite://"))
self.graph1 = Graph(memStore, URIRef("http://example.com/graph1"))
self.graph2 = Graph(memStore, URIRef("http://example.com/graph2"))
self.graph3 = Graph(memStore, URIRef("http://example.com/graph3"))
for n3Str,graph in [(testGraph1N3, self.graph1),
(testGraph2N3, self.graph2),
(testGraph3N3, self.graph3)]:
graph.parse(StringIO(n3Str), format='n3')
self.graph4 = Graph(memStore, RDFS.uri)
self.graph4.parse(RDFS.uri)
self.G = ConjunctiveGraph(memStore)
def testAggregateSPARQL(self):
rt = self.G.query(sparqlQ)
assert len(rt) > 1
rt = self.G.query(sparqlQ2, initBindings={u'?graph' : URIRef(u"http://example.com/graph3")})
try:
import json
except ImportError:
import simplejson as json
res = json.loads(rt.serialize(format='json'))
assert len(res['results']['bindings']) == 20, len(res['results']['bindings'])
开发者ID:pebbie,项目名称:rdflib-sqlalchemy,代码行数:27,代码来源:test_aggregate_graphs.py
示例7: test_01_query
def test_01_query(self):
g = ConjunctiveGraph(self.store)
count = 0
for statement in g.triples((None, None, None)):
count += 1
break
assert count == 1, "Should have found at least one triple"
开发者ID:PMR2,项目名称:virtuoso-python,代码行数:7,代码来源:test_rdflib3.py
示例8: test_issue_250
def test_issue_250(self):
"""
https://github.com/RDFLib/rdflib/issues/250
When I have a ConjunctiveGraph with the default namespace set,
for example
import rdflib
g = rdflib.ConjunctiveGraph()
g.bind(None, "http://defaultnamespace")
then the Trix serializer binds the default namespace twice in its XML
output, once for the Trix namespace and once for the namespace I used:
print(g.serialize(format='trix').decode('UTF-8'))
<?xml version="1.0" encoding="utf-8"?>
<TriX
xmlns:xml="http://www.w3.org/XML/1998/namespace"
xmlns="http://defaultnamespace"
xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"
xmlns="http://www.w3.org/2004/03/trix/trix-1/"
/>
"""
graph = ConjunctiveGraph()
graph.bind(None, "http://defaultnamespace")
sg = graph.serialize(format='trix').decode('UTF-8')
self.assertTrue(
'xmlns="http://defaultnamespace"' not in sg, sg)
self.assertTrue(
'xmlns="http://www.w3.org/2004/03/trix/trix-1/' in sg, sg)
开发者ID:RDFLib,项目名称:rdflib,代码行数:35,代码来源:test_trix_serialize.py
示例9: TestSearchAnnotations
class TestSearchAnnotations(unittest.TestCase):
def test_search_for_uri(self):
for url in annotation_urls:
g, target, selector = specific_resource(self.canvas,
res=URIRef(uuid.uuid4()),
selector=URIRef(uuid.uuid4()))
g, anno, body, target = annotation(g=g,
anno=URIRef(uuid.uuid4()),
target=target,
body=URIRef(uuid.uuid4()))
response = self.client.post(url, data=g.serialize(),
content_type="text/xml")
self.assertEqual(response.status_code, 201)
for uri in [anno, body, target, selector, self.canvas]:
response = self.client.get(url, {'uri': uri})
self.assertEqual(response.status_code, 200)
validate_return_content(self, response, g)
def tearDown(self):
pass
def setUp(self):
url = reverse('semantic_store_annotations', kwargs=dict())
self.client = Client()
fixture_filename = os.path.join(os.path.dirname(os.path.realpath(__file__)),
"semantic_store_test_fixture.xml")
self.g = ConjunctiveGraph(rdfstore.rdfstore(),
identifier=rdfstore.default_identifier)
self.g.parse(fixture_filename)
canvases = self.g.subjects(URIRef(NS.rdf['type']), URIRef(NS.dms['Canvas']))
self.canvas = list(canvases)[0]
开发者ID:bobclewell,项目名称:DM,代码行数:32,代码来源:tests.py
示例10: Db
class Db(_shared):
def __init__(self):
self.lastTimes = []
def getGraph(self):
t1 = time.time()
mtimes = []
for f in (["/my/proj/openid_proxy/access.n3"] +
glob.glob("commentstore/*.nt")):
mtimes.append(os.path.getmtime(f))
if mtimes == self.lastTimes and hasattr(self, 'currentGraph'):
return self.currentGraph
self.lastTimes = mtimes
tf = tempfile.NamedTemporaryFile()
os.system("cat /my/proj/openid_proxy/access.n3 commentstore/*.nt > %s" % tf.name)
self.currentGraph = ConjunctiveGraph()
self.currentGraph.parse(tf.name, format="n3")
log.info("reloaded comments from disk in %f sec" % (time.time() - t1))
return self.currentGraph
def writeFile(self, stmts, ctx, fileWords):
outfile = "commentstore/post-%s.nt" % ("-".join(fileWords))
graph = ConjunctiveGraph()
graph.add(*stmts, **{'context' : ctx})
graph.graph.serialize(outfile, format='n3')
log.info("wrote new comment to %s", outfile)
开发者ID:drewp,项目名称:commentserve,代码行数:32,代码来源:db.py
示例11: TestSparqlJsonResults
class TestSparqlJsonResults(unittest.TestCase):
def setUp(self):
self.graph = ConjunctiveGraph()
self.graph.parse(StringIO(test_data), format="n3")
def _query_result_contains(self, query, correct):
results = self.graph.query(query)
result_json = json.loads(results.serialize(format='json').decode('utf-8'))
msg = "Expected:\n %s \n- to contain:\n%s" % (result_json, correct)
self.assertEqual(sorted(result_json["head"], key=repr),
sorted(correct["head"], key=repr), msg)
# Sort by repr - rather a hack, but currently the best way I can think
# of to ensure the results are in the same order.
result_bindings = sorted(result_json["results"]["bindings"], key=repr)
correct_bindings = sorted(correct["results"]["bindings"], key=repr)
msg = "Expected:\n %s \n- to contain:\n%s" % (result_bindings, correct_bindings)
self.failUnless(result_bindings==correct_bindings, msg)
testOptional = make_method('optional')
testWildcard = make_method('wildcard')
testUnion = make_method('union')
testUnion3 = make_method('union3')
testSelectVars = make_method('select_vars')
testWildcardVars = make_method('wildcard_vars')
开发者ID:RDFLib,项目名称:rdfextras,代码行数:32,代码来源:test_sparql_json_results.py
示例12: TestSparqlEquals
class TestSparqlEquals(unittest.TestCase):
PREFIXES = {
'rdfs': "http://www.w3.org/2000/01/rdf-schema#"
}
def setUp(self):
testContent = """
@prefix rdfs: <%(rdfs)s> .
<http://example.org/doc/1> rdfs:label "Document 1"@en .
<http://example.org/doc/2> rdfs:label "Document 2"@en .
<http://example.org/doc/3> rdfs:label "Document 3"@en .
""" % self.PREFIXES
self.graph = ConjunctiveGraph()
self.graph.load(StringIO(testContent), format='n3')
def test_uri_equals(self):
uri = URIRef("http://example.org/doc/1")
query = ("""
PREFIX rdfs: <%(rdfs)s>
SELECT ?uri WHERE {
?uri rdfs:label ?label .
FILTER( ?uri = <"""+uri+"""> )
}
""") % self.PREFIXES
res = self.graph.query(query)
expected = [(uri,)]
self.assertEqual(list(res),expected)
开发者ID:pombredanne,项目名称:rdfextras,代码行数:29,代码来源:test_sparql_equals.py
示例13: GraphAggregates2
class GraphAggregates2(unittest.TestCase):
# known_issue = True
def setUp(self):
memStore = plugin.get("SQLAlchemy", Store)(identifier="rdflib_test", configuration=Literal("sqlite://"))
self.graph1 = Graph(memStore, URIRef("http://example.com/graph1"))
self.graph2 = Graph(memStore, URIRef("http://example.com/graph2"))
self.graph3 = Graph(memStore, URIRef("http://example.com/graph3"))
for n3Str, graph in [(testGraph1N3, self.graph1), (testGraph2N3, self.graph2), (testGraph3N3, self.graph3)]:
graph.parse(StringIO(n3Str), format="n3")
self.graph4 = Graph(memStore, RDFS.uri)
self.graph4.parse(RDFS.uri)
self.G = ConjunctiveGraph(memStore)
def testAggregateSPARQL(self):
raise SkipTest("known_issue with SELECT from NAMED")
rt = self.G.query(sparqlQ)
assert len(rt) > 1
rt = self.G.query(sparqlQ2, initBindings={u"?graph": URIRef(u"http://example.com/graph3")})
try:
import json
assert json
except ImportError:
import simplejson as json
res = json.loads(rt.serialize(format="json").decode("utf-8"))
assert len(res["results"]["bindings"]) == 20, len(res["results"]["bindings"])
开发者ID:RDFLib,项目名称:rdflib-sqlalchemy,代码行数:28,代码来源:test_aggregate_graphs.py
示例14: fromuri
def fromuri(self, uri):
self.uri = uri
if not uri.startswith('http://rdf.freebase.com'):
self.checkuri()
try:
g = ConjunctiveGraph()
g.load(self.uri)
if g:
logger.info("INFO process_rdf.py - returning graph for " + self.uri)
return g
else:
raise Exception('Nothing was returned, probably caused URL serving no RDF or bad RDF (eg. Freebase): '
'"No handlers could be found for logger "process_rdf.py" -- uri was ' + self.uri)
except URLError as e:
logger.error("URLError process_rdf.py - " + e.message)
raise Exception('URLError, cause either bad URL or no internet connection - ' + e.message + '(uri was ' + self.uri + ')')
except SAXParseException as e:
logger.error("SAXParseException process_rdf.py - " + e.message + '(uri was' + self.uri + ')')
raise Exception('SAXParseException')
except AttributeError as e:
logger.error("AttributeError process_rdf.py - " + e.message + '(uri was' + self.uri + ')')
raise Exception('AttributeError')
else:
self.fromfreebaseuri()
开发者ID:geekscruff,项目名称:peoplesparql,代码行数:29,代码来源:process_rdf.py
示例15: has_correct_hash
def has_correct_hash(self, resource):
f = RdfUtils.get_format(resource.get_filename())
cg = ConjunctiveGraph()
cg.parse(data=resource.get_content(), format=f)
quads = RdfUtils.get_quads(cg)
h = RdfHasher.make_hash(quads, resource.get_hashstr())
return resource.get_hashstr() == h
开发者ID:trustyuri,项目名称:trustyuri-python,代码行数:7,代码来源:RdfModule.py
示例16: RecursionTests
class RecursionTests(unittest.TestCase):
def setUp(self):
self.graph = ConjunctiveGraph()
self.graph.load(StringIO(testContent), format='n3')
def test_simple_recursion(self):
graph = ConjunctiveGraph()
graph.load(StringIO(BASIC_KNOWS_DATA), format='n3')
results = graph.query(KNOWS_QUERY,
DEBUG=False).serialize(format='python')
results = set([tuple(result) for result in results])
person1 = URIRef('ex:person.1')
person2 = URIRef('ex:person.2')
nose.tools.assert_equal(
results,
set([(person1, None), (person1, Literal('person 3')),
(person2, Literal('person 3'))]))
def test_secondary_recursion(self):
graph = ConjunctiveGraph()
graph.load(StringIO(SUBCLASS_DATA), format='n3')
results = graph.query(SUBCLASS_QUERY,
DEBUG=False).serialize(format='python')
results = set([tuple(result) for result in results])
ob = URIRef('ex:ob')
class1 = URIRef('ex:class.1')
class2 = URIRef('ex:class.2')
class3 = URIRef('ex:class.3')
nose.tools.assert_equal(
results,
set([(ob, class1), (ob, class2), (ob, class3)]))
开发者ID:janaya,项目名称:pubsubsemhub,代码行数:31,代码来源:test_sparql_recurse.py
示例17: parse_n3
def parse_n3(term_n3):
''' Disclaimer: Quick and dirty hack using the n3 parser. '''
prepstr = ("@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .\n"
"<urn:no_use> <urn:no_use> %s.\n" % term_n3)
g = ConjunctiveGraph()
g.parse(data=prepstr, format='n3')
return [t for t in g.triples((None, None, None))][0][2]
开发者ID:JesusPatate,项目名称:rdflib,代码行数:7,代码来源:test_util.py
示例18: main
def main():
parser = argparse.ArgumentParser(
description='OMIA integration test',
formatter_class=argparse.RawTextHelpFormatter)
parser.add_argument(
'--input', '-i', type=str, required=True,
help='Location of input ttl file')
args = parser.parse_args()
graph = ConjunctiveGraph()
graph.parse(args.input, format=rdflib_util.guess_format(args.input))
model_of = URIRef('http://purl.obolibrary.org/obo/RO_0003301')
models = graph.subject_objects(model_of)
model_len = len(list(models))
if model_len < EXPECTED_PAIRS:
logger.error("Not enough model_of predicates in graph:"
" {} expected {} check omia log for"
" warnings".format(model_len, EXPECTED_PAIRS))
exit(1)
omim_diseases = graph.objects(
subject=URIRef('https://monarchinitiative.org/model/OMIA-breed:18'),
predicate=model_of
)
if list(omim_diseases) != [URIRef('http://purl.obolibrary.org/obo/OMIM_275220')]:
logger.error("Missing breed to omim triple for {}".format('OMIA-breed:18'))
exit(1)
logger.info("PASSED")
开发者ID:DoctorBud,项目名称:dipper,代码行数:35,代码来源:omia-integration.py
示例19: testParse
def testParse(self):
g = ConjunctiveGraph()
try:
g.parse("http://groups.csail.mit.edu/dig/2005/09/rein/examples/troop42-policy.n3", format="n3")
except URLError:
from nose import SkipTest
raise SkipTest('No network to retrieve the information, skipping test')
开发者ID:takluyver,项目名称:rdflib,代码行数:7,代码来源:test_n3.py
示例20: writeFile
def writeFile(self, stmts, ctx, fileWords):
outfile = "commentstore/post-%s.nt" % ("-".join(fileWords))
graph = ConjunctiveGraph()
graph.add(*stmts, **{'context' : ctx})
graph.graph.serialize(outfile, format='n3')
log.info("wrote new comment to %s", outfile)
开发者ID:drewp,项目名称:commentserve,代码行数:7,代码来源:db.py
注:本文中的rdflib.graph.ConjunctiveGraph类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论