Module coscine.graph

This file provides a simple wrapper around Coscine application profiles. It abstracts the interaction with rdf graphs using rdflib and provides an intuitive, Coscine-specific interface to these graphs.

Expand source code
###############################################################################
# Coscine Python SDK
# Copyright (c) 2018-2022 RWTH Aachen University
# Licensed under the terms of the MIT License
# #############################################################################
# Coscine, short for Collaborative Scientific Integration Environment is
# a platform for research data management (RDM).
# For more information on Coscine visit https://www.coscine.de/.
#
# Please note that this python module is open source software primarily
# developed and maintained by the scientific community. It is not
# an official service that RWTH Aachen provides support for.
###############################################################################

###############################################################################
# File description
###############################################################################

"""
This file provides a simple wrapper around Coscine application profiles.
It abstracts the interaction with rdf graphs using rdflib and
provides an intuitive, Coscine-specific interface to these graphs.
"""

###############################################################################
# Dependencies
###############################################################################

from __future__ import annotations
from typing import List
import rdflib

###############################################################################
# Class
###############################################################################

class ApplicationProfile:
        """
        The ApplicationProfile class serves as a wrapper around Coscine
        application profiles. Coscine application profiles are served
        as rdf graphs which are difficult to interact with, without 3rd
        party libraries.
        The ApplicationProfile class abstracts this interaction with rdf
        graphs.

        Attributes
        ----------
        graph : rdflib.Graph
                An rdf graph parsed with the help of rdflib
        RDFTYPE : str
                A variable frequently used by methods of this class.
        """

        graph: rdflib.Graph
        RDFTYPE: str = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"

###############################################################################

        def __init__(self, graph: str, format: str = "json-ld") -> None:
                """
                Initializes an instance of the ApplicationProfile wrapper class.

                Parameters
                ----------
                graph : str
                        A Coscine application profile rdf graph in json-ld text format.
                format : str (optional), default: "json-ld"
                        The format of the graph
                """

                self.graph = rdflib.Graph()
                self.graph.bind("sh", "http://www.w3.org/ns/shacl#")
                self.graph.bind("dcterms", "http://purl.org/dc/terms/")
                self.graph.parse(data = graph, format = format)

###############################################################################

        def __str__(self) -> str:
                """
                Serializes the application profile rdf graph used internally
                for easy output to stdout.
                """
                return self.graph.serialize(format="ttl")

###############################################################################

        def target(self) -> str:
                """
                Returns a str indicating the target class of the application profile.
                This may for example be "engmeta" in case of an engmeta profile.
                """

                QUERY = \
                        """
                        SELECT ?target WHERE {
                                ?_ sh:targetClass ?target .
                        }
                        """
                result = self.query(QUERY)
                return result[0][0]

###############################################################################

        def query(self, query: str, **kwargs) -> List[List[object]]:
                """
                Performs a SPARQL query against the application profile and
                returns the result as a list of rows.
                """

                items = []
                results = self.graph.query(query, **kwargs)
                for row in results:
                        item = []
                        for val in row:
                                value = val.toPython() if val is not None else None
                                item.append(value)
                        items.append(item)
                return items

###############################################################################

        def items(self) -> List[dict]:
                """
                Returns all items contained within the application profile as a list
                of key value pairs in their order of appearance.
                """

                QUERY = \
                        """
                        SELECT ?path ?name ?order ?class ?minCount ?maxCount ?datatype (lang(?name) as ?lang) WHERE {
                                ?_ sh:path ?path ;
                                        sh:name ?name ;
                                        sh:order ?order .
                                OPTIONAL { ?_ sh:class ?class . } .
                                OPTIONAL { ?_ sh:minCount ?minCount . } .
                                OPTIONAL { ?_ sh:maxCount ?maxCount . } .
                                OPTIONAL { ?_ sh:datatype ?datatype . } .
                        }
                        ORDER BY ASC(?order)
                        """
                items_de = []
                items_en = []
                items = []
                results = self.query(QUERY)
                for result in results:
                        lang = result[7]
                        properties = {
                                "path": result[0],
                                "name": result[1],
                                "order": result[2],
                                "class": result[3],
                                "minCount": result[4] if result[4] else 0,
                                "maxCount": result[5] if result[5] else 42,
                                "datatype": result[6]
                        }
                        if lang == "de": items_de.append(properties)
                        else: items_en.append(properties)

                # Fuse language lists into one
                for index, it in enumerate(items_de):
                        it["name"] = [it["name"], items_en[index]["name"]]
                        items.append(it)
                return items

###############################################################################

        def length(self) -> int:
                """
                Returns the number of fields contained within the application profile.
                """

                QUERY = \
                        """
                        SELECT ?path WHERE {
                                ?_ sh:path ?path ;
                                        sh:order ?order .
                        }
                        """
                results = self.query(QUERY)
                return len(results)

###############################################################################

Classes

class ApplicationProfile (graph: str, format: str = 'json-ld')

The ApplicationProfile class serves as a wrapper around Coscine application profiles. Coscine application profiles are served as rdf graphs which are difficult to interact with, without 3rd party libraries. The ApplicationProfile class abstracts this interaction with rdf graphs.

Attributes

graph : rdflib.Graph
An rdf graph parsed with the help of rdflib
RDFTYPE : str
A variable frequently used by methods of this class.

Initializes an instance of the ApplicationProfile wrapper class.

Parameters

graph : str
A Coscine application profile rdf graph in json-ld text format.
format : str (optional), default: "json-ld"
The format of the graph
Expand source code
class ApplicationProfile:
        """
        The ApplicationProfile class serves as a wrapper around Coscine
        application profiles. Coscine application profiles are served
        as rdf graphs which are difficult to interact with, without 3rd
        party libraries.
        The ApplicationProfile class abstracts this interaction with rdf
        graphs.

        Attributes
        ----------
        graph : rdflib.Graph
                An rdf graph parsed with the help of rdflib
        RDFTYPE : str
                A variable frequently used by methods of this class.
        """

        graph: rdflib.Graph
        RDFTYPE: str = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"

###############################################################################

        def __init__(self, graph: str, format: str = "json-ld") -> None:
                """
                Initializes an instance of the ApplicationProfile wrapper class.

                Parameters
                ----------
                graph : str
                        A Coscine application profile rdf graph in json-ld text format.
                format : str (optional), default: "json-ld"
                        The format of the graph
                """

                self.graph = rdflib.Graph()
                self.graph.bind("sh", "http://www.w3.org/ns/shacl#")
                self.graph.bind("dcterms", "http://purl.org/dc/terms/")
                self.graph.parse(data = graph, format = format)

###############################################################################

        def __str__(self) -> str:
                """
                Serializes the application profile rdf graph used internally
                for easy output to stdout.
                """
                return self.graph.serialize(format="ttl")

###############################################################################

        def target(self) -> str:
                """
                Returns a str indicating the target class of the application profile.
                This may for example be "engmeta" in case of an engmeta profile.
                """

                QUERY = \
                        """
                        SELECT ?target WHERE {
                                ?_ sh:targetClass ?target .
                        }
                        """
                result = self.query(QUERY)
                return result[0][0]

###############################################################################

        def query(self, query: str, **kwargs) -> List[List[object]]:
                """
                Performs a SPARQL query against the application profile and
                returns the result as a list of rows.
                """

                items = []
                results = self.graph.query(query, **kwargs)
                for row in results:
                        item = []
                        for val in row:
                                value = val.toPython() if val is not None else None
                                item.append(value)
                        items.append(item)
                return items

###############################################################################

        def items(self) -> List[dict]:
                """
                Returns all items contained within the application profile as a list
                of key value pairs in their order of appearance.
                """

                QUERY = \
                        """
                        SELECT ?path ?name ?order ?class ?minCount ?maxCount ?datatype (lang(?name) as ?lang) WHERE {
                                ?_ sh:path ?path ;
                                        sh:name ?name ;
                                        sh:order ?order .
                                OPTIONAL { ?_ sh:class ?class . } .
                                OPTIONAL { ?_ sh:minCount ?minCount . } .
                                OPTIONAL { ?_ sh:maxCount ?maxCount . } .
                                OPTIONAL { ?_ sh:datatype ?datatype . } .
                        }
                        ORDER BY ASC(?order)
                        """
                items_de = []
                items_en = []
                items = []
                results = self.query(QUERY)
                for result in results:
                        lang = result[7]
                        properties = {
                                "path": result[0],
                                "name": result[1],
                                "order": result[2],
                                "class": result[3],
                                "minCount": result[4] if result[4] else 0,
                                "maxCount": result[5] if result[5] else 42,
                                "datatype": result[6]
                        }
                        if lang == "de": items_de.append(properties)
                        else: items_en.append(properties)

                # Fuse language lists into one
                for index, it in enumerate(items_de):
                        it["name"] = [it["name"], items_en[index]["name"]]
                        items.append(it)
                return items

###############################################################################

        def length(self) -> int:
                """
                Returns the number of fields contained within the application profile.
                """

                QUERY = \
                        """
                        SELECT ?path WHERE {
                                ?_ sh:path ?path ;
                                        sh:order ?order .
                        }
                        """
                results = self.query(QUERY)
                return len(results)

Class variables

var RDFTYPE : str
var graph : rdflib.graph.Graph

Methods

def items(self) ‑> List[dict]

Returns all items contained within the application profile as a list of key value pairs in their order of appearance.

Expand source code
def items(self) -> List[dict]:
        """
        Returns all items contained within the application profile as a list
        of key value pairs in their order of appearance.
        """

        QUERY = \
                """
                SELECT ?path ?name ?order ?class ?minCount ?maxCount ?datatype (lang(?name) as ?lang) WHERE {
                        ?_ sh:path ?path ;
                                sh:name ?name ;
                                sh:order ?order .
                        OPTIONAL { ?_ sh:class ?class . } .
                        OPTIONAL { ?_ sh:minCount ?minCount . } .
                        OPTIONAL { ?_ sh:maxCount ?maxCount . } .
                        OPTIONAL { ?_ sh:datatype ?datatype . } .
                }
                ORDER BY ASC(?order)
                """
        items_de = []
        items_en = []
        items = []
        results = self.query(QUERY)
        for result in results:
                lang = result[7]
                properties = {
                        "path": result[0],
                        "name": result[1],
                        "order": result[2],
                        "class": result[3],
                        "minCount": result[4] if result[4] else 0,
                        "maxCount": result[5] if result[5] else 42,
                        "datatype": result[6]
                }
                if lang == "de": items_de.append(properties)
                else: items_en.append(properties)

        # Fuse language lists into one
        for index, it in enumerate(items_de):
                it["name"] = [it["name"], items_en[index]["name"]]
                items.append(it)
        return items
def length(self) ‑> int

Returns the number of fields contained within the application profile.

Expand source code
def length(self) -> int:
        """
        Returns the number of fields contained within the application profile.
        """

        QUERY = \
                """
                SELECT ?path WHERE {
                        ?_ sh:path ?path ;
                                sh:order ?order .
                }
                """
        results = self.query(QUERY)
        return len(results)
def query(self, query: str, **kwargs) ‑> List[List[object]]

Performs a SPARQL query against the application profile and returns the result as a list of rows.

Expand source code
def query(self, query: str, **kwargs) -> List[List[object]]:
        """
        Performs a SPARQL query against the application profile and
        returns the result as a list of rows.
        """

        items = []
        results = self.graph.query(query, **kwargs)
        for row in results:
                item = []
                for val in row:
                        value = val.toPython() if val is not None else None
                        item.append(value)
                items.append(item)
        return items
def target(self) ‑> str

Returns a str indicating the target class of the application profile. This may for example be "engmeta" in case of an engmeta profile.

Expand source code
def target(self) -> str:
        """
        Returns a str indicating the target class of the application profile.
        This may for example be "engmeta" in case of an engmeta profile.
        """

        QUERY = \
                """
                SELECT ?target WHERE {
                        ?_ sh:targetClass ?target .
                }
                """
        result = self.query(QUERY)
        return result[0][0]