Skip to content

Commit 919abfe

Browse files
stevenhsdstevenhsd
committed
Merge branch 'release_v010' of https://github.com/NHSDigital/data-validation-engine into docs/sr-ndsp-636_add_linkage_details
2 parents c019901 + 49290bb commit 919abfe

7 files changed

Lines changed: 201 additions & 123 deletions

File tree

‎.mise.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
11
[tools]
22
python="3.12"
33
poetry="2.4.1"
4-
java="liberica-1.8.0"
4+
java="zulu-17.60.17"

‎.tool-versions‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,3 @@
11
python 3.12.12
22
poetry 2.4.1
3-
java liberica-1.8.0
3+
java zulu-17.60.17

‎src/dve/core_engine/backends/base/rules.py‎

Lines changed: 84 additions & 97 deletions
Original file line numberDiff line numberDiff line change
@@ -387,119 +387,107 @@ def identify_and_remove_orphans(
387387
entities: Entities,
388388
entity_hierarchy: EntityHierarchy,
389389
key_fields: Optional[dict[str, list[str]]] = None,
390-
) -> tuple[Messages, bool]:
390+
) -> tuple[Messages, dict[EntityName, bool]]:
391391
"""
392392
Identifies and removes orphan records by traversing the EntityHierarchy object.
393393
An orphan is a child record whose parent FK does not exist in the parent entity.
394394
Processes recursively: removes orphans at each level, then processes children.
395395
"""
396396

397-
def process_node(
398-
node: HierarchyNode,
399-
orph_messages: Messages | None = None,
400-
processed: bool = False,
401-
):
397+
def process_node(node: HierarchyNode):
402398
"""Identify orphans and remove in a given node"""
399+
issues_found: bool = False
400+
if node.parent_entity is None:
401+
return issues_found
403402

404-
if orph_messages is None:
405-
orph_messages = []
403+
self.logger.info(f"Identifying orphans in {node.entity_name}")
406404

407-
if node.parent_entity is not None:
408-
self.logger.info(f"Identifying orphans in {node.entity_name}")
409-
410-
join_expr = " AND ".join(
411-
f"{node.parent_entity}.{k} = {node.entity_name}.{v}"
412-
for k, v in node.join_fields.items()
413-
)
405+
join_expr = " AND ".join(
406+
f"{node.parent_entity}.{k} = {node.entity_name}.{v}"
407+
for k, v in node.join_fields.items()
408+
)
414409

415-
_, no_orphs = self.identify_orphans(
416-
entities=entities,
417-
config=OrphanIdentification(
418-
id=list(node.join_fields.values())[0],
419-
entity_name=node.entity_name,
420-
target_name=node.parent_entity,
421-
join_condition=join_expr,
422-
),
423-
)
410+
_, no_orphs = self.identify_orphans(
411+
entities=entities,
412+
config=OrphanIdentification(
413+
id=list(node.join_fields.values())[0],
414+
entity_name=node.entity_name,
415+
target_name=node.parent_entity,
416+
join_condition=join_expr,
417+
),
418+
)
424419

425-
if no_orphs > 0:
426-
self.logger.info(
427-
f"Removing records with missing parent from {node.entity_name}"
428-
)
429-
processed = True
430-
location = list(node.join_fields.values())[0]
431-
with BackgroundMessageWriter(
432-
working_directory=working_directory,
433-
dve_stage=self.__stage_name__,
434-
key_fields=key_fields,
435-
logger=self.logger,
436-
) as msg_writer:
437-
_orph_records = self.remove_orphans(
438-
entities=entities,
439-
config=OrphanRemoval(
440-
entity_name=node.entity_name,
441-
reporting=ReportingConfig(
442-
emit="record_failure",
443-
code=node.missing_parent_id_error_code,
444-
message=node.missing_parent_id_error_message,
445-
location=location,
446-
),
420+
if no_orphs > 0:
421+
self.logger.info(f"Removing records with missing parent from {node.entity_name}")
422+
issues_found = True
423+
location = list(node.join_fields.values())[0]
424+
with BackgroundMessageWriter(
425+
working_directory=working_directory,
426+
dve_stage=self.__stage_name__,
427+
key_fields=key_fields,
428+
logger=self.logger,
429+
) as msg_writer:
430+
_orph_records = self.remove_orphans(
431+
entities=entities,
432+
config=OrphanRemoval(
433+
entity_name=node.entity_name,
434+
reporting=ReportingConfig(
435+
emit="record_failure",
436+
code=node.missing_parent_id_error_code,
437+
message=node.missing_parent_id_error_message,
438+
location=location,
447439
),
448-
)
449-
# moved to batch the write - risky if large number of
450-
msg_writer.write_queue.put(
451-
[
452-
FeedbackMessage(
453-
entity=node.entity_name,
454-
record=record, # type: ignore
455-
error_location=location,
456-
error_message=node.missing_parent_id_error_message,
457-
failure_type="record",
458-
error_type="record",
459-
error_code=node.missing_parent_id_error_code,
460-
reporting_field=location,
461-
category="Parent Missing",
462-
)
463-
for record in _orph_records
464-
]
465-
)
440+
),
441+
)
442+
# moved to batch the write - risky if large number of
443+
msg_writer.write_queue.put(
444+
[
445+
FeedbackMessage(
446+
entity=node.entity_name,
447+
record=record, # type: ignore
448+
error_location=location,
449+
error_message=node.missing_parent_id_error_message,
450+
failure_type="record",
451+
error_type="record",
452+
error_code=node.missing_parent_id_error_code,
453+
reporting_field=location,
454+
category="Parent Missing",
455+
)
456+
for record in _orph_records
457+
]
458+
)
466459

467-
return processed
460+
return issues_found
468461

469-
processed = False
462+
entity_issues_found: dict[EntityName, bool] = {}
470463

471464
for tree in entity_hierarchy.entity_trees.values():
472465
for node in tree.iterate_root_down():
473-
processed = process_node(node)
466+
entity_issues_found[node.entity_name] = process_node(node)
474467

475468
_orph_rel = entities.get(ORPHANED_RECORD_ENTITY_NAME)
476469
if _orph_rel is not None:
477470
del entities[ORPHANED_RECORD_ENTITY_NAME]
478471

479472
entities.update(entities)
480473

481-
return [], processed
474+
return [], entity_issues_found
482475

483476
def identify_and_remove_missing_mandatory_groups(
484477
self,
485478
working_directory: URI,
486479
entities: Entities,
487480
entity_hierarchy: EntityHierarchy,
488481
key_fields: Optional[dict[str, list[str]]] = None,
489-
) -> tuple[Messages, bool]:
482+
) -> tuple[Messages, dict[EntityName, bool]]:
490483
"""
491484
Identify that an entity with a mandatory key has at least one valid child record.
492485
"""
493486

494-
def process_node(
495-
node: HierarchyNode,
496-
processed: bool = False,
497-
) -> bool:
487+
def process_node(node: HierarchyNode) -> bool:
498488
"""Identify at least one valid child for a mandatory entity at a given node."""
499489
if node.parent_entity is None or not node.mandatory:
500-
return processed
501-
502-
processed = True
490+
return False
503491

504492
self.logger.info(
505493
f"Identifying that mandatory entity `{node.parent_entity}` has at least 1 valid child record" # pylint: disable=C0301
@@ -525,34 +513,33 @@ def process_node(
525513
join_condition=join_expr,
526514
),
527515
)
528-
for record in missing_children_records:
529-
msg_writer.write_queue.put(
530-
[
531-
FeedbackMessage(
532-
entity=node.parent_entity,
533-
record=record, # type: ignore
534-
error_location=location,
535-
error_message=node.no_valid_records_error_message,
536-
failure_type="record",
537-
error_type="record",
538-
error_code=node.no_valid_records_error_code,
539-
reporting_field=location,
540-
category="Children missing",
541-
)
542-
]
516+
_messages = [
517+
FeedbackMessage(
518+
entity=node.parent_entity,
519+
record=record, # type: ignore
520+
error_location=location,
521+
error_message=node.no_valid_records_error_message,
522+
failure_type="record",
523+
error_type="record",
524+
error_code=node.no_valid_records_error_code,
525+
reporting_field=location,
526+
category="Children missing",
543527
)
528+
for record in missing_children_records
529+
]
530+
msg_writer.write_queue.put(_messages)
531+
return len(_messages) > 0
544532

545-
return processed
546-
547-
processed = False
533+
entity_issues_found: dict[EntityName, bool] = {}
548534

549535
for tree in entity_hierarchy.entity_trees.values():
550536
for node in tree.iterate_lowest_descendent_up():
551-
processed = process_node(node, processed)
537+
if node.parent_entity and node.mandatory:
538+
entity_issues_found[node.parent_entity] = process_node(node)
552539

553-
entities.update(entities)
540+
# entities.update(entities)
554541

555-
return [], processed
542+
return [], entity_issues_found
556543

557544
# pylint: disable=R0912,R0914
558545
def apply_sync_filters(

‎src/dve/core_engine/backends/implementations/duckdb/rules.py‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -434,7 +434,11 @@ def identify_orphans(
434434

435435
def remove_orphans(self, entities: DuckDBEntities, *, config: OrphanRemoval) -> Iterator:
436436
"""Method to remove identified orphans in the orphan tracker entity."""
437-
orphan_rel = entities[ORPHANED_RECORD_ENTITY_NAME].set_alias("orphan")
437+
orphan_rel = (
438+
entities[ORPHANED_RECORD_ENTITY_NAME]
439+
.filter(f"entity_name = '{config.entity_name}'")
440+
.set_alias("orphan")
441+
)
438442
filtered_rel = (
439443
entities[config.entity_name]
440444
.set_alias(config.entity_name)
@@ -447,9 +451,7 @@ def remove_orphans(self, entities: DuckDBEntities, *, config: OrphanRemoval) ->
447451

448452
entities[config.entity_name] = filtered_rel
449453

450-
return duckdb_rel_to_dictionaries(
451-
orphan_rel.filter(f"entity_name = '{config.entity_name}'")
452-
)
454+
return duckdb_rel_to_dictionaries(orphan_rel)
453455

454456
def check_mandatory_group(
455457
self, entities: DuckDBEntities, *, config: GroupIdentification

‎src/dve/pipeline/pipeline.py‎

Lines changed: 46 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -643,30 +643,43 @@ def apply_business_rules( # pylint: disable=R0914,R0915
643643
projected
644644
)
645645

646-
_, orph_or_group = self.step_implementations.identify_and_remove_orphans( # type: ignore
646+
_, orph_issues_1 = self.step_implementations.identify_and_remove_orphans( # type: ignore
647647
working_directory,
648648
entity_manager.entities,
649649
entity_hierarchy,
650650
key_fields,
651651
)
652652

653-
_, orph_or_group = self.step_implementations.identify_and_remove_missing_mandatory_groups( # type: ignore
653+
_, grp_issues_1 = self.step_implementations.identify_and_remove_missing_mandatory_groups( # type: ignore
654654
working_directory,
655655
entity_manager.entities,
656656
entity_hierarchy,
657657
key_fields,
658658
)
659659

660660
# Perform a second time incase the mandatory groups result in new orphans
661-
_, orph_or_group = self.step_implementations.identify_and_remove_orphans( # type: ignore
661+
_, orph_issues_2 = self.step_implementations.identify_and_remove_orphans( # type: ignore
662662
working_directory,
663663
entity_manager.entities,
664664
entity_hierarchy,
665665
key_fields,
666666
)
667667

668+
entity_issues: dict[EntityName, bool] = {
669+
entity: any(
670+
val
671+
for val in (
672+
orph_issues_1.get(entity, False),
673+
grp_issues_1.get(entity, False),
674+
orph_issues_2.get(entity, False),
675+
)
676+
)
677+
for entity in orph_issues_1.keys()
678+
}
679+
680+
unchanged_entities: list[EntityName] = []
668681
for entity_name, entity in entity_manager.entities.items():
669-
if orph_or_group:
682+
if entity_issues.get(entity_name, False):
670683
self._logger.info(f"Writing {entity_name} out to disk.")
671684
final_projection = self._step_implementations.write_parquet( # type: ignore
672685
entity,
@@ -677,27 +690,41 @@ def apply_business_rules( # pylint: disable=R0914,R0915
677690
entity_name,
678691
),
679692
)
680-
else:
681-
self._logger.info(f"Moving {entity_name} from temp_business_rules to business_rules")
682-
final_projection = fh.move_resource(
683-
source_uri=fh.joinuri(
684-
self.processed_files_path,
685-
submission_info.submission_id,
686-
"temp_business_rules",
687-
entity_name
688-
),
689-
target_uri=fh.joinuri(
690-
self.processed_files_path,
691-
submission_info.submission_id,
692-
"business_rules",
693-
entity_name
694-
)
693+
694+
entity_manager.entities[entity_name] = self.step_implementations.read_parquet( # type: ignore
695+
final_projection
695696
)
697+
else:
698+
unchanged_entities.append(entity_name)
699+
700+
for entity_name in unchanged_entities:
701+
self._logger.info(f"Moving {entity_name} from temp_business_rules to business_rules")
702+
final_projection = fh.move_resource(
703+
source_uri=fh.joinuri(
704+
self.processed_files_path,
705+
submission_info.submission_id,
706+
"temp_business_rules",
707+
entity_name,
708+
),
709+
target_uri=fh.joinuri(
710+
self.processed_files_path,
711+
submission_info.submission_id,
712+
"business_rules",
713+
entity_name,
714+
),
715+
overwrite=True,
716+
)
696717

697718
entity_manager.entities[entity_name] = self.step_implementations.read_parquet( # type: ignore
698719
final_projection
699720
)
700721

722+
fh.remove_prefix(
723+
fh.joinuri(
724+
self.processed_files_path, submission_info.submission_id, "temp_business_rules"
725+
)
726+
)
727+
701728
submission_status.number_of_records = self.get_entity_count(
702729
entity=entity_manager.entities[f"""Original{rules.global_variables.get(
703730
'entity',

0 commit comments

Comments
 (0)