Coverage for heritrace/routes/entity/_creation.py: 94%
226 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +0000
1# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import json
6from dataclasses import dataclass
8from flask import current_app, flash, jsonify, render_template, request, url_for
9from flask_babel import gettext
10from flask_login import current_user, login_required
11from rdflib import RDF, XSD, Literal, URIRef
12from werkzeug.wrappers import Response
14from heritrace.apis.orcid import get_responsible_agent_uri
15from heritrace.editor import Editor, EditorError, EndpointConfig
16from heritrace.extensions import (
17 get_dataset_endpoint,
18 get_form_fields,
19 get_provenance_endpoint,
20)
21from heritrace.routes.entity._blueprint import entity_bp
22from heritrace.routes.entity._validation import validate_entity_data
23from heritrace.utils.datatypes import DATATYPE_MAPPING, get_datatype_options
24from heritrace.utils.display_rules_utils import (
25 get_class_priority,
26 is_entity_type_visible,
27)
28from heritrace.utils.primary_source_utils import (
29 get_user_default_primary_source,
30 save_user_default_primary_source,
31)
32from heritrace.utils.shacl_utils import find_matching_form_field
33from heritrace.utils.sparql_utils import import_referenced_entities
34from heritrace.utils.uri_utils import generate_unique_uri, is_valid_url
35from heritrace.utils.virtual_properties import (
36 remove_virtual_properties_from_creation_data,
37 transform_entity_creation_with_virtual_properties,
38)
41def _prepare_entity_creation_data(
42 structured_data: dict,
43) -> tuple[dict, str]:
44 cleaned_structured_data = remove_virtual_properties_from_creation_data(
45 structured_data
46 )
47 entity_type: str = cleaned_structured_data["entity_type"]
49 return cleaned_structured_data, entity_type
52def _setup_editor_for_creation(editor: Editor, cleaned_structured_data: dict) -> None:
53 import_referenced_entities(editor, cleaned_structured_data)
54 editor.preexisting_finished()
57def _process_virtual_properties_after_creation(
58 editor: Editor,
59 structured_data: dict,
60 entity_uri: URIRef,
61 default_graph_uri: URIRef | None,
62) -> None:
63 virtual_entities = transform_entity_creation_with_virtual_properties(
64 structured_data, str(entity_uri)
65 )
67 if virtual_entities:
68 editor.begin_counter_transaction()
69 for virtual_entity in virtual_entities:
70 virtual_entity_uri = generate_unique_uri(virtual_entity["entity_type"])
71 create_nested_entity(
72 editor, virtual_entity_uri, virtual_entity, default_graph_uri
73 )
75 editor.save()
78def _create_entity_with_form_fields(
79 editor: Editor,
80 structured_data: dict,
81 entity_uri: URIRef,
82 default_graph_uri: URIRef | None,
83 form_fields: dict,
84) -> None:
85 cleaned_structured_data = remove_virtual_properties_from_creation_data(
86 structured_data
87 )
88 entity_type = cleaned_structured_data["entity_type"]
89 properties = cleaned_structured_data.get("properties", {})
91 for predicate, raw_values in properties.items():
92 predicate_uri = URIRef(predicate)
93 values = raw_values if isinstance(raw_values, list) else [raw_values]
95 entity_shape = cleaned_structured_data.get("entity_shape")
96 matching_key = find_matching_form_field(entity_type, entity_shape, form_fields)
98 field_definitions = (
99 form_fields.get(matching_key, {}).get(predicate, []) if matching_key else []
100 )
102 property_shape = None
103 if values and isinstance(values[0], dict):
104 property_shape = values[0].get("shape")
106 matching_field_def = None
107 for field_def in field_definitions:
108 if property_shape:
109 if field_def.get("subjectShape") == property_shape:
110 matching_field_def = field_def
111 break
112 elif not field_def.get("subjectShape"):
113 matching_field_def = field_def
114 break
116 if not matching_field_def and field_definitions:
117 matching_field_def = field_definitions[0]
119 ordered_by = matching_field_def.get("orderedBy") if matching_field_def else None
121 ctx = CreationContext(
122 editor=editor,
123 entity_uri=entity_uri,
124 predicate=predicate_uri,
125 default_graph_uri=default_graph_uri,
126 )
128 if ordered_by:
129 process_ordered_properties(ctx, values, URIRef(ordered_by))
130 else:
131 process_unordered_properties(ctx, values, matching_field_def)
134def _create_entity_without_form_fields(
135 editor: Editor,
136 structured_data: dict,
137 entity_uri: URIRef,
138 default_graph_uri: URIRef | None,
139) -> None:
140 cleaned_structured_data = remove_virtual_properties_from_creation_data(
141 structured_data
142 )
143 entity_type = cleaned_structured_data["entity_type"]
144 properties = cleaned_structured_data.get("properties", {})
146 editor.create(
147 entity_uri,
148 RDF.type,
149 URIRef(entity_type),
150 default_graph_uri,
151 )
153 for predicate, values in properties.items():
154 predicate_uri = URIRef(predicate)
155 for value_dict in values:
156 if value_dict["type"] == "uri":
157 editor.create(
158 entity_uri,
159 predicate_uri,
160 URIRef(value_dict["value"]),
161 default_graph_uri,
162 )
163 elif value_dict["type"] == "literal":
164 datatype = (
165 URIRef(value_dict["datatype"])
166 if "datatype" in value_dict
167 else XSD.string
168 )
169 editor.create(
170 entity_uri,
171 predicate_uri,
172 Literal(value_dict["value"], datatype=datatype),
173 default_graph_uri,
174 )
177def _handle_create_entity_post(
178 form_fields: dict,
179 structured_data: dict,
180 primary_source: str | None,
181 *,
182 save_default_source: bool,
183) -> tuple[Response, int]:
184 if primary_source and not is_valid_url(primary_source):
185 return jsonify(
186 {
187 "status": "error",
188 "errors": [gettext("Invalid primary source URL provided")],
189 }
190 ), 400
192 if save_default_source and primary_source and is_valid_url(primary_source):
193 save_user_default_primary_source(current_user.orcid, primary_source)
195 if not structured_data.get("entity_type"):
196 return jsonify(
197 {"status": "error", "errors": [gettext("Entity type is required")]}
198 ), 400
200 cleaned_structured_data, entity_type = _prepare_entity_creation_data(
201 structured_data
202 )
204 if form_fields:
205 validation_errors = validate_entity_data(cleaned_structured_data)
206 if validation_errors:
207 return jsonify({"status": "error", "errors": validation_errors}), 400
209 resp_agent = get_responsible_agent_uri(current_user.orcid)
210 editor = Editor(
211 EndpointConfig(
212 dataset=get_dataset_endpoint(),
213 provenance=get_provenance_endpoint(),
214 is_quadstore=current_app.config["DATASET_IS_QUADSTORE"],
215 ),
216 current_app.config["COUNTER_HANDLER"],
217 resp_agent,
218 URIRef(current_app.config["PRIMARY_SOURCE"]),
219 current_app.config["DATASET_GENERATION_TIME"],
220 save_plugin=current_app.config.get("SAVE_PLUGIN"),
221 )
222 entity_uri = generate_unique_uri(entity_type)
223 default_graph_uri = (
224 URIRef(f"{entity_uri}/graph") if editor.dataset_is_quadstore else None
225 )
226 if not form_fields:
227 editor.import_entity(entity_uri)
228 _setup_editor_for_creation(editor, cleaned_structured_data)
229 editor.set_primary_source(URIRef(primary_source) if primary_source else None)
231 if form_fields:
232 _create_entity_with_form_fields(
233 editor,
234 structured_data,
235 entity_uri,
236 default_graph_uri,
237 form_fields,
238 )
239 else:
240 _create_entity_without_form_fields(
241 editor,
242 structured_data,
243 entity_uri,
244 default_graph_uri,
245 )
247 try:
248 editor.save()
249 _process_virtual_properties_after_creation(
250 editor, structured_data, entity_uri, default_graph_uri
251 )
252 except (EditorError, OSError) as e:
253 error_message = gettext(
254 "An error occurred while creating the entity: %(error)s", error=str(e)
255 )
256 return jsonify({"status": "error", "errors": [error_message]}), 500
257 else:
258 response = jsonify(
259 {
260 "status": "success",
261 "redirect_url": url_for("entity.about", subject=str(entity_uri)),
262 }
263 )
264 flash(gettext("Entity created successfully"), "success")
265 return response, 200
268@entity_bp.route("/create-entity", methods=["GET", "POST"])
269@login_required
270def create_entity() -> str | tuple[Response, int]:
271 form_fields = get_form_fields()
273 default_primary_source = get_user_default_primary_source(current_user.orcid)
275 entity_class_shape_pairs = sorted(
276 [
277 entity_key
278 for entity_key in form_fields
279 if is_entity_type_visible(entity_key)
280 ],
281 key=get_class_priority,
282 reverse=True,
283 )
285 datatype_options = get_datatype_options()
287 if request.method == "POST":
288 structured_data = json.loads(request.form.get("structured_data", "{}"))
289 primary_source = request.form.get("primary_source") or None
290 save_default_source = request.form.get("save_default_source") == "true"
291 return _handle_create_entity_post(
292 form_fields,
293 structured_data,
294 primary_source,
295 save_default_source=save_default_source,
296 )
298 return render_template(
299 "create_entity.jinja",
300 datatype_options=datatype_options,
301 dataset_db_triplestore=current_app.config["DATASET_DB_TRIPLESTORE"],
302 dataset_db_text_index_enabled=current_app.config[
303 "DATASET_DB_TEXT_INDEX_ENABLED"
304 ],
305 default_primary_source=default_primary_source,
306 shacl=bool(form_fields),
307 entity_class_shape_pairs=entity_class_shape_pairs,
308 )
311def create_nested_entity(
312 editor: Editor,
313 entity_uri: URIRef,
314 entity_data: dict,
315 graph_uri: URIRef | None = None,
316) -> None:
317 form_fields = get_form_fields()
319 editor.create(
320 entity_uri,
321 RDF.type,
322 URIRef(entity_data["entity_type"]),
323 graph_uri,
324 )
326 entity_type = entity_data.get("entity_type")
327 entity_shape = entity_data.get("entity_shape")
328 properties = entity_data.get("properties", {})
330 matching_key = find_matching_form_field(entity_type, entity_shape, form_fields)
332 if not matching_key:
333 return
335 for predicate, raw_values in properties.items():
336 predicate_uri = URIRef(predicate)
337 values = raw_values if isinstance(raw_values, list) else [raw_values]
338 field_definitions = form_fields[matching_key].get(predicate, [])
340 for value in values:
341 if isinstance(value, dict) and "entity_type" in value:
342 if "intermediateRelation" in value:
343 intermediate_uri = generate_unique_uri(
344 value["intermediateRelation"]["class"]
345 )
346 target_uri = generate_unique_uri(value["entity_type"])
347 editor.create(
348 entity_uri, predicate_uri, intermediate_uri, graph_uri
349 )
350 editor.create(
351 intermediate_uri,
352 URIRef(value["intermediateRelation"]["property"]),
353 target_uri,
354 graph_uri,
355 )
356 create_nested_entity(editor, target_uri, value, graph_uri)
357 else:
358 nested_uri = generate_unique_uri(value["entity_type"])
359 editor.create(entity_uri, predicate_uri, nested_uri, graph_uri)
360 create_nested_entity(editor, nested_uri, value, graph_uri)
361 elif isinstance(value, dict) and value.get("is_existing_entity", False):
362 existing_entity_uri = value.get("entity_uri")
363 if existing_entity_uri:
364 editor.create(
365 entity_uri,
366 predicate_uri,
367 URIRef(existing_entity_uri),
368 graph_uri,
369 )
370 else:
371 str_value = str(value)
372 if is_valid_url(str_value):
373 object_value: URIRef | Literal = URIRef(str_value)
374 else:
375 datatype = XSD.string
376 datatype_uris = []
377 if field_definitions:
378 datatype_uris = field_definitions[0].get("datatypes", [])
379 datatype = determine_datatype(str_value, datatype_uris)
380 object_value = Literal(str_value, datatype=datatype)
381 editor.create(entity_uri, predicate_uri, object_value, graph_uri)
384@dataclass(frozen=True, slots=True)
385class CreationContext:
386 editor: Editor
387 entity_uri: URIRef
388 predicate: URIRef
389 default_graph_uri: URIRef | None
392def process_entity_value(
393 ctx: CreationContext,
394 value: dict | str,
395 matching_field_def: dict | None,
396) -> URIRef | Literal:
397 if isinstance(value, dict) and "entity_type" in value:
398 nested_uri = generate_unique_uri(value["entity_type"])
399 ctx.editor.create(
400 ctx.entity_uri,
401 ctx.predicate,
402 nested_uri,
403 ctx.default_graph_uri,
404 )
405 create_nested_entity(ctx.editor, nested_uri, value, ctx.default_graph_uri)
406 return nested_uri
407 if isinstance(value, dict) and value.get("is_existing_entity", False):
408 entity_ref_uri = value.get("entity_uri")
409 if entity_ref_uri:
410 object_value = URIRef(entity_ref_uri)
411 ctx.editor.create(
412 ctx.entity_uri,
413 ctx.predicate,
414 object_value,
415 ctx.default_graph_uri,
416 )
417 return object_value
418 msg = "Missing entity_uri in existing entity reference"
419 raise ValueError(msg)
420 str_value = str(value)
421 if is_valid_url(str_value):
422 object_value: URIRef | Literal = URIRef(str_value)
423 else:
424 datatype_uris = []
425 if matching_field_def:
426 datatype_uris = matching_field_def.get("datatypes", [])
427 datatype = determine_datatype(str_value, datatype_uris)
428 object_value = Literal(str_value, datatype=datatype)
429 ctx.editor.create(
430 ctx.entity_uri,
431 ctx.predicate,
432 object_value,
433 ctx.default_graph_uri,
434 )
435 return object_value
438def _process_ordered_entity_value(
439 ctx: CreationContext,
440 value: dict,
441) -> URIRef:
442 if isinstance(value, dict) and "entity_type" in value:
443 nested_uri = generate_unique_uri(value["entity_type"])
444 ctx.editor.create(
445 ctx.entity_uri,
446 ctx.predicate,
447 nested_uri,
448 ctx.default_graph_uri,
449 )
450 create_nested_entity(ctx.editor, nested_uri, value, ctx.default_graph_uri)
451 return nested_uri
452 if isinstance(value, dict) and value.get("is_existing_entity", False):
453 nested_uri = URIRef(value["entity_uri"])
454 ctx.editor.create(
455 ctx.entity_uri,
456 ctx.predicate,
457 nested_uri,
458 ctx.default_graph_uri,
459 )
460 return nested_uri
461 msg = "Unexpected value type for ordered property"
462 raise ValueError(msg)
465def process_ordered_properties(
466 ctx: CreationContext,
467 values: list[dict],
468 ordered_by: URIRef,
469) -> None:
470 values_by_shape = {}
471 for value in values:
472 shape = value.get("entity_shape")
473 if not shape:
474 shape = "default_shape"
475 if shape not in values_by_shape:
476 values_by_shape[shape] = []
477 values_by_shape[shape].append(value)
479 for shape_values in values_by_shape.values():
480 previous_entity = None
481 for value in shape_values:
482 nested_uri = _process_ordered_entity_value(ctx, value)
484 if previous_entity:
485 ctx.editor.create(
486 previous_entity,
487 ordered_by,
488 nested_uri,
489 ctx.default_graph_uri,
490 )
491 previous_entity = nested_uri
494def process_unordered_properties(
495 ctx: CreationContext,
496 values: list[dict | str],
497 matching_field_def: dict | None,
498) -> None:
499 for value in values:
500 process_entity_value(ctx, value, matching_field_def)
503def determine_datatype(value: str, datatype_uris: list[str]) -> URIRef:
504 for datatype_uri in datatype_uris:
505 validation_func = next(
506 (d[1] for d in DATATYPE_MAPPING if str(d[0]) == str(datatype_uri)), None
507 )
508 if validation_func and validation_func(value):
509 return URIRef(datatype_uri)
510 return XSD.string