本文整理汇总了Python中rdflib.Graph类的典型用法代码示例。如果您正苦于以下问题:Python Graph类的具体用法?Python Graph怎么用?Python Graph使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了Graph类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_post_no_type_to_root
def test_post_no_type_to_root(self):
graph = Graph()
created = BNode()
graph.add((self.my_ktbs.uri, RDFS.seeAlso, created))
graph.add((created, RDF.type, KTBS.Base))
with assert_raises(RdfRestException):
self.my_ktbs.post_graph(graph)
开发者ID:ktbs,项目名称:ktbs,代码行数:7,代码来源:test_ktbs_engine.py
示例2: _read_id_from_install_rdf
def _read_id_from_install_rdf(self, installrdfpath):
from rdflib import Graph
rdf = Graph()
installrdf = rdf.parse(file=file(installrdfpath))
for i in installrdf.all_nodes():
if re.search(".*@.*\..*", i):
return i.decode()
开发者ID:yongpeishi,项目名称:Pair-Stair,代码行数:7,代码来源:firefox_profile.py
示例3: generate_productlist
def generate_productlist(config, data):
g = Graph('IOMemory', BNode())
for product_data in data['products']:
product = URIRef(link(product_data['link'])+"#subject")
add_lang_names(g, product, product_data['names'], rel=[SCHEMA.name, DCTERMS.title])
g.add((product, RDF.type, SCHEMA.CreativeWork))
return g
开发者ID:qood,项目名称:vgmdb,代码行数:7,代码来源:rdf.py
示例4: fragment
def fragment(gp, broker, agora, channel, updating, gen):
try:
gp_match = re.search(r'\{(.*)\}', gp).groups(0)
if len(gp_match) != 1:
raise click.ClickException('Invalid graph pattern')
STOA = {
"broker_host": broker[0],
"broker_port": broker[1],
"agora_host": agora[0],
"agora_port": agora[1],
"exchange": channel[0],
"topic_pattern": channel[1],
"response_prefix": channel[2]
}
tps = re.split('\. ', gp_match[0])
prefixes, fragment_gen = get_fragment_generator(*tps, monitoring=30, STOA=STOA, updating=updating, gen=gen)
graph = Graph()
for prefix in prefixes:
graph.bind(prefix, prefixes[prefix])
click.echo('@prefix {}: <{}> .'.format(prefix, prefixes[prefix]))
click.echo('')
for chunk in fragment_gen:
if chunk is not None:
headers, (c, s, p, o) = chunk
triple = u'{} {} {} .'.format(s.n3(graph.namespace_manager), p.n3(graph.namespace_manager),
o.n3(graph.namespace_manager))
click.echo(triple)
except Exception as e:
raise click.ClickException('There was a problem with the request: {}'.format(e.message))
开发者ID:fserena,项目名称:agora-stoa-client,代码行数:33,代码来源:cli.py
示例5: annotateConfidence
def annotateConfidence(target, un, con, com):
# thisAnnotation id is the full string, eg:
# http://chartex.org/user/jjon/annotation/dc9d7cbdd0ebefb583e46fc2b79bc8cedde34d68
# the last element being a hash (hashlib.sha1(oa:hastarget).hexdigest()) of this full string:
# http://chartex.org/graphid/Person_11139might_bePerson_11339 (this triple is actually in there, why?, weird!
target = re.sub('[<>]', '', target)
thisAnnotationURI = "http://chartex.org/user/%s/annotation/%s" % (un, sha1(target).hexdigest())
confidence = Literal(con) if con == 'nochange' else Literal(con,datatype=XSD.decimal)
#TODO: if no change, create no confidenceMetric triple for the annotation OR insert original decimal value
if (int(annotationExists('<' + thisAnnotationURI + '>')) > 0):
return ("You've already annotated this statement: %s \nPresumably you could make a separate annotation with a different username. If you start doing that, you should keep track of all your usernames. When we have authentication and session logic, this won't be necessary.\n\nAnnotation triples:\n" % (target,), getSingleConfidenceAnnotation('<' + thisAnnotationURI + '>', 'application/rdf+xml'))
else:
thisann = URIRef(thisAnnotationURI)
g = Graph()
bodyNode = BNode()
triples = [
(thisann, RDF.type, oa.Annotation),
(thisann, oa.hasTarget, URIRef(target)),
(thisann, oa.hasBody, bodyNode),
(bodyNode, chartex.suggestedConfidenceMetric, confidence),
(bodyNode, chartex.userComment, Literal(com))
]
for t in triples: g.add(t)
r = requests.post(
AGVM_VC_REPO + "/statements",
headers={'Content-Type': 'text/turtle'},
data=g.serialize(format='turtle'),
auth=AG_AUTH
)
return (g.serialize(format='pretty-xml'), r.__dict__)
开发者ID:jjon,项目名称:Graphing-ChartEx,代码行数:32,代码来源:cartametallon.py
示例6: testConjunction
def testConjunction(self):
self.addStuffInMultipleContexts()
triple = (self.pizza, self.likes, self.pizza)
# add to context 1
graph = Graph(self.graph.store, self.c1)
graph.add(triple)
self.assertEquals(len(self.graph), len(graph))
开发者ID:RDFLib,项目名称:rdfextras,代码行数:7,代码来源:test_context.py
示例7: ext_json
def ext_json():
rdfUrl = ''
tok = Tokenizer()
if request.method == 'POST':
rdf = request.form['data']
status_test = "0"#request.form['status']
filters = ""#request.form['exculdeurls']
#rdf = "http://jpp.no-ip.org/MAD_J.rdf"
try:
#r = requests.get(rdf)
gg = Graph()
#g.load(rdfUrl)
rdf_content = StringIO.StringIO(rdf.encode('utf-8'))
#print rdf_content.readline()
gg.parse(rdf_content, format="xml")
ext = Extractor(gg)
uris = ext.getUris()
mapping = MapFactory()
for uri in uris:
term = tok.tokenized_url(uri)
uri_status = ""
if status_test == "1":
uri_status = ext.testUri(uri)
else:
uri_status = "N/A"
uri_lookup = str(uri)+"\""
lnum = ext.get_lines(rdf_content, uri_lookup)
ent = MapEntry(uri, term, "", lnum, uri_status)
mapping.add(ent)
jsonized_result = json.dumps(mapping.get())
return Response(jsonized_result, mimetype='application/json')
except requests.exceptions.ConnectionError:
X2Rwarning = 'X2R Warning: The requested URL raises ConnectionError~!!!'
return X2Rwarning
开发者ID:FengPu,项目名称:x2r-me,代码行数:34,代码来源:x2r-me.py
示例8: test_dataset_description_linksets
def test_dataset_description_linksets(self):
res = self.client.get('/.well-known/void')
self.assertEqual(res.status_code, http.client.OK)
self.assertEqual(res.headers['Content-Type'], 'text/turtle')
g = Graph()
g.parse(format='turtle', data=res.get_data(as_text=True))
# http://dbpedia.org/void/Dataset
q = sparql.prepareQuery('''
SELECT ?triples
WHERE {
?linkset a void:Linkset .
?linkset void:subset <http://n2t.net/ark:/99152/p0d> .
?linkset void:subjectsTarget <http://n2t.net/ark:/99152/p0d> .
?linkset void:linkPredicate ?predicate .
?linkset void:objectsTarget ?dataset .
?linkset void:triples ?triples .
}
''', initNs={'void': VOID})
dbpedia = URIRef('http://dbpedia.org/void/Dataset')
triples = next(iter(g.query(
q, initBindings={'dataset': dbpedia,
'predicate': DCTERMS.spatial})))['triples'].value
self.assertEqual(triples, 3)
worldcat = URIRef('http://purl.oclc.org/dataset/WorldCat')
triples = next(iter(g.query(
q, initBindings={'dataset': worldcat,
'predicate': DCTERMS.isPartOf})))['triples'].value
self.assertEqual(triples, 1)
开发者ID:periodo,项目名称:periodo-server,代码行数:29,代码来源:test_representation.py
示例9: all_products
def all_products():
params = request.args.get('categoria')
g = Graph()
try:
g.parse('prueba.rdf', format='xml')
except Exception,e:
print str(e)
开发者ID:laura-chacon,项目名称:ECSDI_project,代码行数:7,代码来源:agente_tienda.py
示例10: __init__
def __init__(self, err, data, namespace=None):
self.err = err
self.manifest = u"urn:mozilla:install-manifest"
self.namespace = namespace or "http://www.mozilla.org/2004/em-rdf"
if isinstance(data, types.StringTypes):
data = StringIO(data) # Wrap data in a pseudo-file
from rdflib.plugins.parsers import rdfxml
orig_create_parser = rdfxml.create_parser
try:
# Patch rdflib to not resolve URL entities.
def create_parser(*args, **kwargs):
parser = orig_create_parser(*args, **kwargs)
parser.setEntityResolver(AddonRDFEntityResolver(err))
return parser
rdfxml.create_parser = create_parser
# Load up and parse the file in XML format.
graph = Graph()
graph.parse(data, format="xml")
self.rdf = graph
except ParserError as ex:
# Re-raise the exception in a local exception type.
raise RDFException(message=ex.message)
except SAXParseException as ex:
# Raise the SAX parse exceptions so we get some line info.
raise RDFException(orig_exception=ex)
finally:
# If we fail, we don't want to sully up the creation function.
rdfxml.create_parser = orig_create_parser
开发者ID:Rob--W,项目名称:amo-validator,代码行数:33,代码来源:rdf.py
示例11: test_load_from_model
def test_load_from_model(self):
"""Can we round trip through a RDF model"""
model = Graph()
path = '/root/42BW9AAXX/C1-38/Project_12345/'
filename = '12345_AAATTT_L003_R1_001.fastq.gz'
seq = sequences.parse_fastq(path, filename)
seq.save_to_model(model)
seq_id = 'file://'+path+filename
seqNode = URIRef(seq_id)
libNode = URIRef('http://localhost/library/12345')
model.add((seqNode, libraryOntology['library'], libNode))
seq2 = sequences.SequenceFile.load_from_model(model, seq_id)
self.assertEqual(seq.flowcell, seq2.flowcell)
self.assertEqual(seq.flowcell, '42BW9AAXX')
self.assertEqual(seq.filetype, seq2.filetype)
self.assertEqual(seq2.filetype, 'split_fastq')
self.assertEqual(seq.lane, seq2.lane)
self.assertEqual(seq2.lane, '3')
self.assertEqual(seq.read, seq2.read)
self.assertEqual(seq2.read, 1)
self.assertEqual(seq.project, seq2.project)
self.assertEqual(seq2.project, '12345')
self.assertEqual(seq.index, seq2.index)
self.assertEqual(seq2.index, 'AAATTT')
self.assertEqual(seq.split, seq2.split)
self.assertEqual(seq2.split, '001')
self.assertEqual(seq.cycle, seq2.cycle)
self.assertEqual(seq.pf, seq2.pf)
self.assertEqual(seq2.libraryNode, libNode)
self.assertEqual(seq.path, seq2.path)
开发者ID:detrout,项目名称:htsworkflow,代码行数:32,代码来源:test_sequences.py
示例12: test_post_no_type_to_base
def test_post_no_type_to_base(self):
graph = Graph()
created = BNode()
graph.add((self.my_ktbs.uri, RDFS.seeAlso, created))
graph.add((created, RDF.type, KTBS.hasModel)) # in correct NS
with assert_raises(RdfRestException):
self.my_ktbs.post_graph(graph)
开发者ID:ktbs,项目名称:ktbs,代码行数:7,代码来源:test_ktbs_engine.py
示例13: test_post_bad_type_to_base
def test_post_bad_type_to_base(self):
graph = Graph()
created = BNode()
graph.add((self.my_ktbs.uri, KTBS.contains, created))
graph.add((created, RDF.type, RDFS.Resource))
with assert_raises(RdfRestException):
self.my_ktbs.post_graph(graph)
开发者ID:ktbs,项目名称:ktbs,代码行数:7,代码来源:test_ktbs_engine.py
示例14: open_store
def open_store(identifier):
ident = URIRef(identifier)
store = plugin.get("SQLAlchemy", Store)(identifier=ident)
graph = Graph(store, identifier=ident)
uri = Literal(os.environ.get("DATABASE_URL"))
graph.open(uri, create=False)
return graph
开发者ID:skebix,项目名称:ocl-manager,代码行数:7,代码来源:views.py
示例15: test_period_json
def test_period_json(self):
res1 = self.client.get('/trgkvwbjd.json')
self.assertEqual(res1.status_code, http.client.OK)
self.assertEqual(res1.headers['Content-Type'], 'application/json')
self.assertEqual(
res1.headers['Content-Disposition'],
'attachment; filename="periodo-period-trgkvwbjd.json"')
context = json.loads(res1.get_data(as_text=True))['@context']
self.assertEqual(context, [
'http://localhost.localdomain:5000/c',
{'@base': 'http://n2t.net/ark:/99152/'}])
res2 = self.client.get('/trgkvwbjd.jsonld')
self.assertEqual(res2.status_code, http.client.OK)
self.assertEqual(res2.headers['Content-Type'], 'application/ld+json')
self.assertEqual(
res2.headers['Content-Disposition'],
'attachment; filename="periodo-period-trgkvwbjd.json"')
jsonld = json.loads(res1.get_data(as_text=True))
context = json.loads(self.client.get('/c', buffered=True)
.get_data(as_text=True))
g = Graph().parse(
data=json.dumps({**jsonld, **context}), format='json-ld')
self.assertIsNone(
g.value(predicate=RDF.type, object=SKOS.ConceptScheme))
self.assertIn((PERIODO['p0trgkvwbjd'],
FOAF.isPrimaryTopicOf, HOST['trgkvwbjd.json']), g)
self.assertIn((HOST['trgkvwbjd.json'],
VOID.inDataset, HOST['d']), g)
self.assertIn((PERIODO['p0trgkvwbjd'],
SKOS.inScheme, PERIODO['p0trgkv']), g)
res3 = self.client.get('/trgkvwbjd.json.html')
self.assertEqual(res3.status_code, http.client.OK)
self.assertEqual(res3.headers['Content-Type'], 'text/html')
开发者ID:periodo,项目名称:periodo-server,代码行数:35,代码来源:test_representation.py
示例16: create_store_with_identifier
def create_store_with_identifier(identifier):
ident = URIRef(identifier)
store = plugin.get("SQLAlchemy", Store)(identifier=ident)
graph = Graph(store, identifier=ident)
uri = Literal(os.environ.get("DATABASE_URL"))
graph.open(uri, create=True)
graph.parse(join(join(settings.BASE_DIR, 'static'), 'output.xml'))
开发者ID:skebix,项目名称:ocl-manager,代码行数:7,代码来源:views.py
示例17: test_period_turtle
def test_period_turtle(self):
res1 = self.client.get('/trgkvwbjd.ttl')
self.assertEqual(res1.status_code, http.client.OK)
self.assertEqual(res1.headers['Content-Type'], 'text/turtle')
self.assertEqual(
res1.headers['Cache-Control'],
'public, max-age={}'.format(cache.SHORT_TIME))
self.assertEqual(
res1.headers['Content-Disposition'],
'attachment; filename="periodo-period-trgkvwbjd.ttl"')
g = Graph().parse(data=res1.get_data(as_text=True), format='turtle')
self.assertIsNone(
g.value(predicate=RDF.type, object=SKOS.ConceptScheme))
self.assertIn((PERIODO['p0trgkvwbjd'],
FOAF.isPrimaryTopicOf, HOST['trgkvwbjd.ttl']), g)
self.assertIn((HOST['trgkvwbjd.ttl'],
VOID.inDataset, HOST['d']), g)
self.assertIn((PERIODO['p0trgkvwbjd'],
SKOS.inScheme, PERIODO['p0trgkv']), g)
res2 = self.client.get('/trgkvwbjd.ttl.html')
self.assertEqual(res2.status_code, http.client.OK)
self.assertEqual(res2.headers['Content-Type'], 'text/html')
self.assertEqual(
res2.headers['Cache-Control'],
'public, max-age={}'.format(cache.SHORT_TIME))
开发者ID:periodo,项目名称:periodo-server,代码行数:26,代码来源:test_representation.py
示例18: main
def main():
graph = Graph()
graph.parse(sys.argv[1], format="n3")
if len(sys.argv) > 2:
doc = URIRef(sys.argv[2])
else:
docs = []
for c in (RIF.Document, RIF.BLDDocument,
RIF.PRDDocument, RIF.CoreDocument):
for x in graph.subjects(RDF.type, c):
docs.append(x)
if len(docs) == 1:
doc = docs[0]
elif len(docs) > 1:
print >>sys.stderr, "Input contains multiple Document nodes"
print >>sys.stderr, indent+",".join([repr(x) for x in docs])
print >>sys.stderr, "Name one on the command line to select it"
sys.exit(1)
elif len(docs) < 1:
print >>sys.stderr, "Input contains no Document nodes"
for (s,p,o) in graph:
print s,p,o
sys.exit(1)
out = sys.stdout
to_rif(out, graph, doc, root=True)
开发者ID:mpetyx,项目名称:pyrif,代码行数:28,代码来源:xtr.py
示例19: test_history_turtle
def test_history_turtle(self):
with self.client as client:
res = client.patch(
'/d/',
data=self.patch,
content_type='application/json',
headers={'Authorization': 'Bearer '
+ 'NTAwNWViMTgtYmU2Yi00YWMwLWIwODQtMDQ0MzI4OWIzMzc4'})
res = client.post(
urlparse(res.headers['Location']).path + 'merge',
buffered=True,
headers={'Authorization': 'Bearer '
+ 'ZjdjNjQ1ODQtMDc1MC00Y2I2LThjODEtMjkzMmY1ZGFhYmI4'})
res1 = self.client.get('/history.ttl')
self.assertEqual(res1.status_code, http.client.OK)
self.assertEqual(res1.headers['Content-Type'], 'text/turtle')
self.assertEqual(
res1.headers['Cache-Control'],
'public, max-age={}'.format(cache.SHORT_TIME))
self.assertEqual(
res1.headers['Content-Disposition'],
'attachment; filename="periodo-history.ttl"')
g = Graph()
g.parse(data=res1.get_data(as_text=True), format='turtle')
self.assertIn((HOST['h#patch-1'],
FOAF.page, HOST['patches/1/patch.jsonpatch']), g)
self.assertIn((HOST['d'],
DCTERMS.provenance, HOST['h#changes']), g)
res3 = self.client.get('/history.ttl/')
self.assertEqual(res3.status_code, http.client.NOT_FOUND)
开发者ID:periodo,项目名称:periodo-server,代码行数:33,代码来源:test_representation.py
示例20: parse
def parse(self):
if "workflowBundle.ttl" in self.zip.namelist():
format = "n3"
rootfile = "workflowBundle.ttl"
elif "workflowBundle.rdf" in self.zip.namelist():
rootfile = "workflowBundle.rdf"
format = "xml"
else:
raise Scufl2Error("Can't find workflowBundle.ttl or "
"workflowBundle.rdf")
self.uri = "file://" + urllib.pathname2url(os.path.abspath(self.filename)) + "/"
early_graph = Graph()
rdf_file = self.zip.open(rootfile)
early_graph.parse(rdf_file, self.uri, format=format)
sameBaseAs = list(early_graph.objects(subject=URIRef(self.uri), predicate=Scufl2NS.sameBaseAs))
if not sameBaseAs:
# Fall back to the file:/// URIs
self.graph = early_graph
else:
# Use the sameBaseAs as the base
self.uri = sameBaseAs[0]
self.graph = Graph()
# Reparse it
rdf_file = self.zip.open(rootfile)
self.graph.parse(rdf_file, self.uri, format=format)
self.parse_all_graphs(self.uri)
开发者ID:egarcia1357,项目名称:scufl2,代码行数:29,代码来源:processorNames.py
注:本文中的rdflib.Graph类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论