Python rdflib 模块，Graph() 实例源码

我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用rdflib.Graph()。

项目：scheduled-bots 作者：SuLab | 项目源码 | 文件源码

def get_do_metadata():
    # from the do owl file, get do labels, descriptions
    g = Graph()
    g.parse(DO_OWL_PATH)

    disease_ontology = Literal('disease_ontology', datatype=URIRef('http://www.w3.org/2001/XMLSchema#string'))
    query = """
        SELECT * WHERE {
            ?id oboInOwl:hasOBONamespace ?disease_ontology .
            ?id rdfs:label ?label .
            OPTIONAL {?id obo:IAO_0000115 ?descr}
            FILTER NOT EXISTS {?id owl:deprecated ?dep}
        }
        """
    rows = g.query(query, initBindings={'disease_ontology': disease_ontology})
    res = [{str(k): str(v) for k, v in binding.items()} for binding in rows.bindings]
    df = pd.DataFrame(res)
    df.drop_duplicates(subset=['id'], inplace=True)
    df.fillna("", inplace=True)
    do = df.to_dict("records")
    do = {purl_to_curie(x['id']): x for x in do}
    return do

项目：pyfc4 作者：ghukill | 项目源码 | 文件源码

def test_graph_parse(self):

        # collect graphs
        graphs = []
        # loop through Content-Types, save parsed graphs
        content_types = [
            'application/ld+json',
            'application/n-triples',
            'application/rdf+xml',
            'text/n3',
            'text/plain',
            'text/turtle'
        ]
        for content_type in content_types:
            logger.debug("testing parsing of Content-Type: %s" % content_type)
            foo = repo.get_resource('%s/foo' % testing_container_uri, response_format=content_type)
            # test that graph was parsed correctly
            assert type(foo.rdf.graph) == rdflib.graph.Graph


    # create child container foo/bar (basic container)

项目：QuitStore 作者：AKSW | 项目源码 | 文件源码

def __readGraphIriFile(self, graphfile):
        """Search for a graph uri in graph file and return it.

        Args:
            graphfile: String containing the path of a graph file

        Returns:
            graphuri: String with the graph URI
        """
        try:
            with open(graphfile, 'r') as f:
                graphuri = f.readline().strip()
        except FileNotFoundError:
            logger.debug("File not found {}".format(graphfile))
            return

        try:
            urlparse(graphuri)
            logger.debug("Graph URI {} found in {}".format(graphuri, graphfile))
        except Exception:
            graphuri = None
            logger.debug("No graph URI found in {}".format(graphfile))

        return graphuri

项目：QuitStore 作者：AKSW | 项目源码 | 文件源码

def getgraphfromfile(self):
        """Return a Conjunctive Graph generated from the referenced file.

        Returns:
            A ConjunctiveGraph
        """
        graph = ConjunctiveGraph()

        try:
            graph.parse(self.path, format='nquads', publicID='http://localhost:5000/')
            logger.debug('Success: File', self.path, 'parsed')
        except KeyError as e:
            # Given file contains non valid rdf data
            # logger.debug('Error: File', self.path, 'not parsed')
            # self.__setcontent([[None][None][None][None]])
            pass

        return graph

项目：genepio 作者：GenEpiO | 项目源码 | 文件源码

def __init__(self, namespace={}, prefixes='', newqueries={}):

        self.graph = rdflib.Graph()

        self.namespace.update(namespace)
        self.prefixes += prefixes
        # run all given "SELECT" queries through prepareQuery function.
        for (id, query) in newqueries.iteritems():
            leader = query.strip()[0:6]
            if leader == 'SELECT': # prepareQuery only works on SELECT ...
                self.queries[id] = rdflib.plugins.sparql.prepareQuery(query, initNs = self.namespace)
                print "Adding SELECT query"

            elif leader == 'DELETE' or leader == 'INSERT':
                self.queries[id] = query
                print "Adding DEL/INS query"
        #self.queries.update(queries)
        #print "Done query prep."

项目：py-ldnlib 作者：trellis-ldp | 项目源码 | 文件源码

def notification(self, iri, **kwargs):
        """
        Retrieve a single LDN notification and decode into a Python object.
        """
        headers = kwargs.pop("headers", dict())
        if 'accept' not in headers:
            headers['accept'] = kwargs.pop("accept", self.accept_headers)

        r = requests.get(iri, headers=headers, **kwargs)
        r.raise_for_status()
        mime_type = self.content_type_to_mime_type(r.headers['content-type'])
        if mime_type == self.JSON_LD:
            return r.json()
        else:
            g = Graph().parse(data=r.text, format=mime_type)
            return json.loads(str(g.serialize(format="json-ld"), 'utf-8'))

项目：smartcontainers 作者：crcresearch | 项目源码 | 文件源码

def __init__(self, filename='sc.config'):
        """Initialize

        Parameters
        ----------
        :param filename: string
            For renaming the configuration file name.  Default value.

        Returns
        -------
        :returns: none
        """
        self.graph = rdflib.Graph()
        self.filename = filename

        if os.environ.get('SC_HOME'):
            self.config_path = os.getenv('SC_HOME')
        else:
            os.environ['SC_HOME'] = os.environ['HOME'] + '/.sc/'
            self.config_path = os.getenv('SC_HOME')