Coverage for oc_ocdm/abstract_entity.py: 82%
74 statements
« prev ^ index » next coverage.py v6.5.0, created at 2025-05-30 22:05 +0000
1#!/usr/bin/python
2# -*- coding: utf-8 -*-
3# Copyright (c) 2016, Silvio Peroni <essepuntato@gmail.com>
4#
5# Permission to use, copy, modify, and/or distribute this software for any purpose
6# with or without fee is hereby granted, provided that the above copyright notice
7# and this permission notice appear in all copies.
8#
9# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT,
12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE,
13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS
15# SOFTWARE.
16from __future__ import annotations
18from abc import ABC
19from typing import TYPE_CHECKING
21from oc_ocdm.support.support import create_type, create_literal, get_short_name, is_dataset, is_string_empty
22from rdflib import URIRef, RDFS, RDF, Literal, Graph
24if TYPE_CHECKING:
25 from typing import Optional, List, ClassVar, Dict, Tuple, Iterable
26 from rdflib import term
29class AbstractEntity(ABC):
30 """
31 Abstract class which represents a generic entity from the OCDM. It sits
32 at the top of the entity class hierarchy.
33 """
35 short_name_to_type_iri: ClassVar[Dict[str, URIRef]] = {}
37 def __init__(self) -> None:
38 """
39 Constructor of the ``AbstractEntity`` class.
40 """
41 self.g: Graph = Graph()
42 self.res: URIRef = URIRef("")
43 self.short_name: str = ""
45 def remove_every_triple(self) -> None:
46 """
47 Remover method that removes every triple from the current entity.
49 **WARNING: the OCDM specification requires that every entity has at least
50 one triple that defines its type (through the** ``rdf:type`` **RDF predicate). If
51 such triple is not subsequently restored by the user, the entity will be considered
52 as to be deleted since it wouldn't be valid anymore.**
54 :return: None
55 """
56 self.g.remove((None, None, None))
58 # LABEL
59 def get_label(self) -> Optional[str]:
60 """
61 Getter method corresponding to the ``rdfs:label`` RDF predicate.
63 :return: The requested value if found, None otherwise
64 """
65 return self._get_literal(RDFS.label)
67 def create_label(self, string: str) -> None:
68 """
69 Setter method corresponding to the ``rdfs:label`` RDF predicate.
71 **WARNING: this is a functional property, hence any existing value will be overwritten!**
73 :param string: The value that will be set as the object of the property related to this method
74 :type string: str
75 :return: None
76 """
77 self.remove_label()
78 self._create_literal(RDFS.label, string)
80 def remove_label(self) -> None:
81 """
82 Remover method corresponding to the ``rdfs:label`` RDF predicate.
84 :return: None
85 """
86 self.g.remove((self.res, RDFS.label, None))
88 def _create_literal(self, p: URIRef, s: str, dt: URIRef = None, nor: bool = True) -> None:
89 """
90 Adds an RDF triple with a literal object inside the graph of the entity
92 :param p: The predicate
93 :type p: URIRef
94 :param s: The string to add as a literal value
95 :type s: str
96 :param dt: The object's datatype, if present
97 :type dt: URIRef, optional
98 :param nor: Whether to normalize the graph or not
99 :type nor: bool, optional
100 :return: None
101 """
102 create_literal(self.g, self.res, p, s, dt, nor)
104 # TYPE
105 def get_types(self) -> List[URIRef]:
106 """
107 Getter method corresponding to the ``rdf:type`` RDF predicate.
109 :return: A list containing the requested values if found, None otherwise
110 """
111 uri_list: List[URIRef] = self._get_multiple_uri_references(RDF.type)
112 return uri_list
114 def _create_type(self, res_type: URIRef) -> None:
115 """
116 Setter method corresponding to the ``rdf:type`` RDF predicate.
118 **WARNING: the OCDM specification admits at most two types for an entity.
119 The main type cannot be edited or removed. Any existing secondary type
120 will be overwritten!**
122 :param res_type: The value that will be set as the object of the property related to this method
123 :type res_type: URIRef
124 :return: None
125 """
126 self.remove_type() # <-- It doesn't remove the main type!
127 create_type(self.g, self.res, res_type)
129 def remove_type(self) -> None:
130 """
131 Remover method corresponding to the ``rdf:type`` RDF predicate.
133 **WARNING: the OCDM specification requires at least one type for an entity.
134 This method removes any existing secondary type, without removing the main type.**
136 :return: None
137 """
138 self.g.remove((self.res, RDF.type, None))
139 # Restore the main type IRI
140 iri_main_type: URIRef = self.short_name_to_type_iri[self.short_name]
141 create_type(self.g, self.res, iri_main_type)
143 # Overrides __str__ method
144 def __str__(self) -> str:
145 return str(self.res)
147 def add_triples(self, iterable_of_triples: Iterable[Tuple[term]]) -> None:
148 """
149 A utility method that allows to add a batch of triples into the graph of the entity.
151 **WARNING: Only triples that have this entity as their subject will be imported!**
153 :param iterable_of_triples: A collection of triples to be added to the entity
154 :type iterable_of_triples: Iterable[Tuple[term]]
155 :return: None
156 """
157 for s, p, o in iterable_of_triples:
158 if s == self.res: # This guarantees that only triples belonging to the resource will be added
159 self.g.add((s, p, o))
161 def _get_literal(self, predicate: URIRef) -> Optional[str]:
162 result: Optional[str] = None
163 for o in self.g.objects(self.res, predicate):
164 if type(o) == Literal:
165 result = str(o)
166 break
167 return result
169 def _get_multiple_literals(self, predicate: URIRef) -> List[str]:
170 result: List[str] = []
171 for o in self.g.objects(self.res, predicate):
172 if type(o) == Literal:
173 result.append(str(o))
174 return result
176 def _get_uri_reference(self, predicate: URIRef, short_name: str = None) -> Optional[URIRef]:
177 result: Optional[URIRef] = None
178 for o in self.g.objects(self.res, predicate):
179 if type(o) == URIRef:
180 if not is_string_empty(short_name):
181 # If a particular short_name is explicitly requested,
182 # then the following additional check must be performed:
183 if (short_name == '_dataset_' and is_dataset(o)) or get_short_name(o) == short_name:
184 result = o
185 break
186 else:
187 result = o
188 break
189 return result
191 def _get_multiple_uri_references(self, predicate: URIRef, short_name: str = None) -> List[URIRef]:
192 result: List[URIRef] = []
193 for o in self.g.objects(self.res, predicate):
194 if type(o) == URIRef:
195 if not is_string_empty(short_name):
196 # If a particular short_name is explicitly requested,
197 # then the following additional check must be performed:
198 if (short_name == '_dataset_' and is_dataset(o)) or get_short_name(o) == short_name:
199 result.append(o)
200 else:
201 result.append(o)
202 return result