Coverage for oc_ocdm / abstract_entity.py: 96%
71 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-28 18:52 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2020-2022 Simone Persiani <iosonopersia@gmail.com>
4# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
5#
6# SPDX-License-Identifier: ISC
8# -*- coding: utf-8 -*-
9from __future__ import annotations
11from abc import ABC
12from typing import TYPE_CHECKING
14from oc_ocdm.support.support import create_type, create_literal, get_short_name, is_dataset, is_string_empty
15from rdflib import URIRef, RDFS, RDF, Literal, Graph
17if TYPE_CHECKING:
18 from typing import Optional, List, ClassVar, Dict, Iterable
19 from rdflib.term import IdentifiedNode, Node
22class AbstractEntity(ABC):
23 """
24 Abstract class which represents a generic entity from the OCDM. It sits
25 at the top of the entity class hierarchy.
26 """
28 short_name_to_type_iri: ClassVar[Dict[str, URIRef]] = {}
30 def __init__(self) -> None:
31 """
32 Constructor of the ``AbstractEntity`` class.
33 """
34 self.g: Graph = Graph()
35 self.res: URIRef = URIRef("")
36 self.short_name: str = ""
38 def remove_every_triple(self) -> None:
39 """
40 Remover method that removes every triple from the current entity.
42 **WARNING: the OCDM specification requires that every entity has at least
43 one triple that defines its type (through the** ``rdf:type`` **RDF predicate). If
44 such triple is not subsequently restored by the user, the entity will be considered
45 as to be deleted since it wouldn't be valid anymore.**
47 :return: None
48 """
49 self.g.remove((None, None, None)) # type: ignore[arg-type]
51 # LABEL
52 def get_label(self) -> Optional[str]:
53 """
54 Getter method corresponding to the ``rdfs:label`` RDF predicate.
56 :return: The requested value if found, None otherwise
57 """
58 return self._get_literal(RDFS.label)
60 def create_label(self, string: str) -> None:
61 """
62 Setter method corresponding to the ``rdfs:label`` RDF predicate.
64 **WARNING: this is a functional property, hence any existing value will be overwritten!**
66 :param string: The value that will be set as the object of the property related to this method
67 :type string: str
68 :return: None
69 """
70 self.remove_label()
71 self._create_literal(RDFS.label, string)
73 def remove_label(self) -> None:
74 """
75 Remover method corresponding to the ``rdfs:label`` RDF predicate.
77 :return: None
78 """
79 self.g.remove((self.res, RDFS.label, None))
81 def _create_literal(self, p: URIRef, s: str, dt: Optional[URIRef] = None, nor: bool = True) -> None:
82 """
83 Adds an RDF triple with a literal object inside the graph of the entity
85 :param p: The predicate
86 :type p: URIRef
87 :param s: The string to add as a literal value
88 :type s: str
89 :param dt: The object's datatype, if present
90 :type dt: URIRef, optional
91 :param nor: Whether to normalize the graph or not
92 :type nor: bool, optional
93 :return: None
94 """
95 create_literal(self.g, self.res, p, s, dt, nor)
97 # TYPE
98 def get_types(self) -> List[URIRef]:
99 """
100 Getter method corresponding to the ``rdf:type`` RDF predicate.
102 :return: A list containing the requested values if found, None otherwise
103 """
104 uri_list: List[URIRef] = self._get_multiple_uri_references(RDF.type)
105 return uri_list
107 def _create_type(self, res_type: URIRef) -> None:
108 """
109 Setter method corresponding to the ``rdf:type`` RDF predicate.
111 **WARNING: the OCDM specification admits at most two types for an entity.
112 The main type cannot be edited or removed. Any existing secondary type
113 will be overwritten!**
115 :param res_type: The value that will be set as the object of the property related to this method
116 :type res_type: URIRef
117 :return: None
118 """
119 self.remove_type() # <-- It doesn't remove the main type!
120 create_type(self.g, self.res, res_type)
122 def remove_type(self) -> None:
123 """
124 Remover method corresponding to the ``rdf:type`` RDF predicate.
126 **WARNING: the OCDM specification requires at least one type for an entity.
127 This method removes any existing secondary type, without removing the main type.**
129 :return: None
130 """
131 self.g.remove((self.res, RDF.type, None))
132 # Restore the main type IRI
133 iri_main_type: URIRef = self.short_name_to_type_iri[self.short_name]
134 create_type(self.g, self.res, iri_main_type)
136 # Overrides __str__ method
137 def __str__(self) -> str:
138 return str(self.res)
140 def add_triples(self, iterable_of_triples: Iterable[tuple[IdentifiedNode, URIRef, Node]]) -> None:
141 """
142 A utility method that allows to add a batch of triples into the graph of the entity.
144 **WARNING: Only triples that have this entity as their subject will be imported!**
146 :param iterable_of_triples: A collection of triples to be added to the entity
147 :type iterable_of_triples: Iterable[Tuple[term]]
148 :return: None
149 """
150 for s, p, o in iterable_of_triples:
151 if s == self.res: # This guarantees that only triples belonging to the resource will be added
152 self.g.add((s, p, o))
154 def _get_literal(self, predicate: URIRef) -> Optional[str]:
155 result: Optional[str] = None
156 for o in self.g.objects(self.res, predicate):
157 if type(o) == Literal:
158 result = str(o)
159 break
160 return result
162 def _get_multiple_literals(self, predicate: URIRef) -> List[str]:
163 result: List[str] = []
164 for o in self.g.objects(self.res, predicate):
165 if type(o) == Literal:
166 result.append(str(o))
167 return result
169 def _get_uri_reference(self, predicate: URIRef, short_name: Optional[str] = None) -> Optional[URIRef]:
170 result: Optional[URIRef] = None
171 for o in self.g.objects(self.res, predicate):
172 if type(o) == URIRef:
173 if not is_string_empty(short_name):
174 # If a particular short_name is explicitly requested,
175 # then the following additional check must be performed:
176 if (short_name == '_dataset_' and is_dataset(o)) or get_short_name(o) == short_name:
177 result = o
178 break
179 else:
180 result = o
181 break
182 return result
184 def _get_multiple_uri_references(self, predicate: URIRef, short_name: Optional[str] = None) -> List[URIRef]:
185 result: List[URIRef] = []
186 for o in self.g.objects(self.res, predicate):
187 if type(o) == URIRef:
188 if not is_string_empty(short_name):
189 # If a particular short_name is explicitly requested,
190 # then the following additional check must be performed:
191 if (short_name == '_dataset_' and is_dataset(o)) or get_short_name(o) == short_name:
192 result.append(o)
193 else:
194 result.append(o)
195 return result