Source code for MyCapytain.resources.prototypes.metadata

# -*- coding: utf-8 -*-
.. module:: MyCapytain.resources.prototypes.metadata
   :synopsis: Definition of Metadata Type Objects

.. moduleauthor:: Thibault Clérice <>


from MyCapytain.common.metadata import Metadata
from MyCapytain.errors import UnknownCollection
from MyCapytain.common.utils import Subgraph, LiteralToDict
from MyCapytain.common.constants import RDF_NAMESPACES, RDFLIB_MAPPING, Mimetypes, get_graph
from MyCapytain.common.base import Exportable
from rdflib import URIRef, RDF, Literal, Graph, RDFS
from rdflib.namespace import SKOS, DC

[docs]class Collection(Exportable): """ Collection represents any resource's metadata. It has members and parents :ivar properties: Properties of the collection :type properties: dict :ivar parents: Parent of the node from the direct parent to the highest ascendant :type parents: [Collection] :ivar metadata: Metadata :type metadata: Metadata """ TYPE_URI = URIRef(RDF_NAMESPACES.DTS.collection) MODEL_URI = URIRef(RDF_NAMESPACES.DTS.collection) EXPORT_TO = [Mimetypes.JSON.LD, Mimetypes.JSON.DTS.Std, Mimetypes.XML.RDF] def __init__(self, identifier="", *args, **kwargs): super(Collection, self).__init__(identifier, *args, **kwargs) self.__graph__ = get_graph() self.__node__ = URIRef(identifier) self.__metadata__ = Metadata(node=self.asNode()) self.__capabilities__ = Metadata.getOr(self.asNode(), RDF_NAMESPACES.DTS.capabilities) self.graph.set((self.asNode(), RDF.type, self.TYPE_URI)) self.graph.set((self.asNode(), RDF_NAMESPACES.DTS.model, self.MODEL_URI)) self.graph.addN( [ (self.asNode(), RDF_NAMESPACES.DTS.capabilities, self.capabilities.asNode(), self.graph) ] ) self.__parent__ = None self.__children__ = {} def __repr__(self): return "%s(%s)" % (self.__class__.__name__, @property def version(self): for x in self.graph.objects(self.asNode(), RDF_NAMESPACES.DTS.version): return x @version.setter def version(self, value): if not isinstance(value, Literal): value = Literal(value) self.graph.set((self.asNode(), RDF_NAMESPACES.DTS.version, value)) @property def type(self): return list(self.graph.objects(self.asNode(), RDF.type))[0] @type.setter def type(self, value): if not isinstance(value, URIRef): value = URIRef(value) self.graph.set((self.asNode(), RDF.type, value)) @property def model(self): return list(self.graph.objects(self.asNode(), RDF_NAMESPACES.DTS.model))[0] @model.setter def model(self, value): if not isinstance(value, URIRef): value = URIRef(value) self.graph.set((self.asNode(), RDF_NAMESPACES.DTS.model, value)) @property def size(self): return len(self.members) # Graph Related Properties @property def graph(self): """ RDFLib Graph space :rtype: Graph """ return self.__graph__ @property def metadata(self): return self.__metadata__ @property def capabilities(self): return self.__capabilities__
[docs] def asNode(self): """ Node representation of the collection in the graph :rtype: URIRef """ return self.__node__
@property def id(self): return str(self.asNode())
[docs] def get_label(self, lang=None): """ Return label for given lang or any default :param lang: Language to request :return: Label value :rtype: Literal """ x = None if lang is None: for obj in self.graph.objects(self.asNode(), RDFS.label): return obj for obj in self.graph.objects(self.asNode(), RDFS.label): x = obj if x.language == lang: return x return x
[docs] def set_label(self, label, lang): """ Add the label of the collection in given lang :param label: Label Value :param lang: Language code """ self.metadata.add(SKOS.prefLabel, Literal(label, lang=lang)) self.graph.addN([ (self.asNode(), RDFS.label, Literal(label, lang=lang), self.graph), ])
@property def children(self): """ Dictionary of childrens {Identifier: Collection} :rtype: dict """ return self.__children__ @property def parents(self): """ Iterator to find parents of current collection, from closest to furthest :rtype: Generator[:class:`Collection`] """ p = self.parent parents = [] while p is not None: parents.append(p) p = p.parent return parents @property def parent(self): """ Parent of current object :rtype: Collection """ return self.__parent__ @parent.setter def parent(self, parent): """ Parents :param parent: Parent to set for the object :type parent: Collection :return: """ self.__parent__ = parent self.graph.set( (self.asNode(), RDF_NAMESPACES.DTS.parent, parent.asNode()) ) parent.__add_member__(self) def __add_member__(self, member): """ Does not add member if it already knows it. .. warning:: It should not be called ! :param member: Collection to add to members """ if in self.children: return None else: self.children[] = member #self.graph.add((self.asNode(), RDF_NAMESPACES.DTS.child, member.asNode())) def __getitem__(self, key): """ Retrieve an item by its ID in the tree of a collection :param key: Key of the object to delete :return: Collection identified by the item """ for obj in self.descendants + [self]: if == key: return obj raise UnknownCollection("%s is not part of this object" % key) def __delitem__(self, key): """ Delete a children of the collection :param key: Key of the object to delete """ item = self[key] # Delete the graph Item self.graph.remove((item.asNode(), None, None)) self.graph.remove((None, None, item.asNode())) self.metadata.remove() self.metadata.unlink() # Delete the Python item if len(item.parents) > 0: del item.parents[0].children[] def __contains__(self, item): """ Retrieve an item by its ID in the tree of a collection :param item: :return: Collection identified by the item """ for obj in self.descendants + [self]: if == item: return True return False @property def readable(self): """ Readable property should return elements where the element can be queried for getPassage / getReffs """ return False @property def members(self): """ Children of the collection's item :rtype: [Collection] """ return list(self.children.values()) @property def descendants(self): """ Any descendant (no max level) of the collection's item :rtype: [Collection] """ return self.members + \ [submember for member in self.members for submember in member.descendants] @property def readableDescendants(self): """ List of element available which are readable :rtype: [Collection] """ return [member for member in self.descendants if member.readable] def __namespaces_header__(self, cpt=None): """ Generates Namespaces Header given the graph :return: Dictionary with XMLNS prefix and uri as key and values """ nm = self.graph.namespace_manager bindings = {} for predicate in set(self.graph.predicates()): prefix, namespace, name = nm.compute_qname(predicate) if prefix != "": bindings["xmlns:" + prefix] = str(URIRef(namespace))#[:-1] else: bindings["xmlns"] = str(URIRef(namespace))#[:-1] if cpt is True: bindings["xmlns:cpt"] = str(RDF_NAMESPACES.CAPITAINS) # Small hard coded fix for namespace that were not thought for RDF for k, v in bindings.items(): if v in [""]: bindings[k] = v[:-1] return bindings def __export__(self, output=None, domain=""): """ Export the collection item in the Mimetype required. ..note:: If current implementation does not have special mimetypes, reuses default_export method :param output: Mimetype to export to (Uses MyCapytain.common.utils.Mimetypes) :type output: str :param domain: Domain (Necessary sometime to express some IDs) :type domain: str :return: Object using a different representation """ if output == Mimetypes.JSON.DTS.Std: nm = self.graph.namespace_manager bindings = {} for predicate in set(self.graph.predicates()): prefix, namespace, name = nm.compute_qname(predicate) bindings[prefix] = str(URIRef(namespace)) RDFSLabel = self.graph.qname(RDFS.label) RDFType = self.graph.qname(RDF.type) store = Subgraph(get_graph().namespace_manager) store.graphiter(self.graph, self.metadata, ascendants=0, descendants=1) metadata = {} for _, predicate, obj in store.graph: k = self.graph.qname(predicate) if k in metadata: if isinstance(metadata[k], list): metadata[k].append(LiteralToDict(obj)) else: metadata[k] = [metadata[k], LiteralToDict(obj)] else: metadata[k] = LiteralToDict(obj) o = { "@context": bindings, "@graph": { "@id":, RDFType: str(self.type), RDFSLabel: LiteralToDict(self.get_label()) or, self.graph.qname(RDF_NAMESPACES.DTS.size): len(self.members), self.graph.qname(RDF_NAMESPACES.DTS.metadata): metadata } } version = self.version if version is not None: o["@graph"]["version"] = str(version) if len(self.members): o["@graph"][self.graph.qname(RDF_NAMESPACES.DTS.members)] = [ { "@id":, RDFSLabel: LiteralToDict(member.get_label()) or, self.graph.qname(RDF_NAMESPACES.DTS.url): domain + } for member in self.members ] if self.parent: o["@graph"][self.graph.qname(RDF_NAMESPACES.DTS.parents)] = [ { "@id":, RDFSLabel: LiteralToDict(member.get_label()) or, self.graph.qname(RDF_NAMESPACES.DTS.url): domain + } for member in self.parents ] del store return o elif output == Mimetypes.JSON.LD\ or output == Mimetypes.XML.RDF: # We create a temp graph store = Subgraph(get_graph().namespace_manager) store.graphiter(self.graph, self.asNode(), ascendants=1, descendants=-1) o = store.serialize(format=RDFLIB_MAPPING[output], auto_compact=True, indent="") del store return o
[docs]class ResourceCollection(Collection): @property def lang(self): """ Languages this text is in :return: List of available languages """ return str(self.graph.value(self.asNode(), DC.language)) @lang.setter def lang(self, lang): """ Language this text is available in :param lang: Language to add :type lang: str """ self.graph.set((self.asNode(), DC.language, Literal(lang)))
[docs] def get_creator(self, lang=None): """ Get the DC Creator literal value :param lang: Language to retrieve :return: Creator string representation :rtype: Literal """ return self.metadata.get_single(key=DC.creator, lang=lang)
[docs] def get_title(self, lang=None): """ Get the title of the object :param lang: Lang to retrieve :return: Title string representation :rtype: Literal """ return self.metadata.get_single(key=DC.title, lang=lang)
[docs] def get_description(self, lang=None): """ Get the description of the object :param lang: Lang to retrieve :return: Description string representation :rtype: Literal """ return self.metadata.get_single(key=DC.description, lang=lang)
[docs] def get_subject(self, lang=None): """ Get the subject of the object :param lang: Lang to retrieve :return: Subject string representation :rtype: Literal """ return self.metadata.get_single(key=DC.subject, lang=lang)