From debe04ed7ce51d00895044ba587e1e8313a342a5 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Mon, 20 Jul 2026 15:55:51 +0100 Subject: [PATCH 1/6] Add support for selected entities and redirects in add-data workflow - Introduced `selected_entities` and `selected_redirects` parameters in relevant functions and methods across the pipeline, workflow, and tasks modules. - Enhanced entity assignment logic to prioritize selected entities. - Updated tests to validate the new functionality and ensure correct behavior with selected entities and redirects. --- .../src/application/core/pipeline.py | 263 ++++++++- .../src/application/core/workflow.py | 6 + request-processor/src/tasks.py | 2 + .../src/application/core/test_pipeline.py | 501 +++++++++++++++++- .../src/application/core/test_workflow.py | 6 + .../tests/unit/src/test_tasks.py | 3 + request_model/schemas.py | 3 + 7 files changed, 778 insertions(+), 6 deletions(-) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index b3ac744..7eb5620 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -2,6 +2,7 @@ import csv import json import yaml +from datetime import date from application.logging.logger import get_logger from digital_land.organisation import Organisation from digital_land.api import API @@ -16,6 +17,197 @@ logger = get_logger(__name__) +def _selected_entity_keys(selected_entities): + """Return the stable keys used to match selected add-data rows.""" + return { + ( + str(entity.get("reference", "")).strip(), + str(entity.get("organisation", "")).strip(), + ) + for entity in selected_entities or [] + if entity.get("reference") and entity.get("organisation") + } + + +def _filter_selected_entities(new_entities, selected_entities): + """ + Return selected entities, treating a missing or empty selection as all entities. + + This mirrors the add-data request contract: selected_entities=None or [] + means process every new entity. A non-empty but invalid selection matches none. + """ + if selected_entities is None or selected_entities == []: + return new_entities + + selected_keys = _selected_entity_keys(selected_entities) + if not selected_keys: + return [] + + return [ + entity + for entity in new_entities + if ( + str(entity.get("reference", "")).strip(), + str(entity.get("organisation", "")).strip(), + ) + in selected_keys + ] + + +def _normalise_selection_list(selection): + """Normalise optional single-object/list request selections to a list.""" + if selection is None or selection == []: + return [] + if isinstance(selection, dict): + return [selection] + if not isinstance(selection, list): + return [] + return selection + + +def _create_old_entity_redirects( + new_entities, + selected_redirects, + selected_entities=None, + duplicate_candidates=None, +): + """ + Build old-entity rows from explicit redirect selections. + + Redirects are only created for assigned entities matching the selected + reference and organisation. When duplicate evidence exists for the same + old/new entity pair, it is copied into the notes field. + """ + selected_redirects = _normalise_selection_list(selected_redirects) + if not selected_redirects: + return [] + + selected_entity_keys = _selected_entity_keys(selected_entities) + new_entity_by_key = { + ( + str(entity.get("reference", "")).strip(), + str(entity.get("organisation", "")).strip(), + ): entity + for entity in new_entities + } + evidence_by_redirect = { + ( + str(candidate.get("old_entity", "")).strip(), + str(candidate.get("entity", "")).strip(), + ): str(candidate.get("evidence", "") or "") + for candidate in duplicate_candidates or [] + } + + old_entity_rows = [] + seen = set() + for redirect in selected_redirects: + old_entity_number = str(redirect.get("old_entity_number", "")).strip() + key = ( + str(redirect.get("reference", "")).strip(), + str(redirect.get("organisation", "")).strip(), + ) + if selected_entity_keys and key not in selected_entity_keys: + continue + + new_entity = new_entity_by_key.get(key) + if not old_entity_number or not new_entity or not new_entity.get("entity"): + continue + + new_entity_number = new_entity.get("entity") + notes = evidence_by_redirect.get( + (old_entity_number, str(new_entity_number).strip()), "" + ) + row = _old_entity_row(old_entity_number, new_entity_number, notes=notes) + if not row: + continue + + row_key = (row["old-entity"], row["entity"]) + if row_key in seen: + continue + seen.add(row_key) + old_entity_rows.append(row) + + return old_entity_rows + + +def _old_entity_row(old_entity, entity, notes=""): + """Return a valid old-entity redirect row, or None for incomplete ids.""" + old_entity = str(old_entity or "").strip() + entity = str(entity or "").strip() + if not old_entity or not entity: + return None + + return { + "old-entity": old_entity, + "status": "301", + "entity": entity, + "entry-date": date.today().isoformat(), + "notes": str(notes or ""), + } + + +def _name_similarity_score(candidate): + """Parse a duplicate candidate name similarity value like '86%'.""" + similarity = str(candidate.get("name_similarity", "") or "").strip() + if similarity.endswith("%"): + similarity = similarity[:-1] + try: + return int(float(similarity)) + except ValueError: + return 0 + + +def _should_auto_redirect(candidate): + """Return True for duplicate candidates safe enough to redirect automatically.""" + match_type = candidate.get("match_type") + if match_type == "complete_match": + return True + return match_type == "single_match" and _name_similarity_score(candidate) > 85 + + +def _create_auto_old_entity_redirects(duplicate_candidates, selected_entity_ids=None): + """ + Build old-entity rows for duplicate matches that meet the auto-redirect rules. + + Complete matches always qualify. Single matches qualify only above 85%. + Existing redirects are skipped, and selected_entity_ids limits redirects to + the user's selected new entities when a non-empty selection was supplied. + """ + old_entity_rows = [] + for candidate in duplicate_candidates: + entity = str(candidate.get("entity", "")).strip() + if selected_entity_ids is not None and entity not in selected_entity_ids: + continue + if candidate.get("old_entity_redirects"): + continue + if not _should_auto_redirect(candidate): + continue + + row = _old_entity_row( + candidate.get("old_entity"), + entity, + notes=candidate.get("evidence", ""), + ) + if row: + old_entity_rows.append(row) + + return old_entity_rows + + +def _merge_old_entity_rows(*row_groups): + """Merge old-entity row groups, keeping the first row for each old/new pair.""" + rows = [] + seen = set() + for row_group in row_groups: + for row in row_group: + key = (row.get("old-entity"), row.get("entity")) + if key in seen: + continue + seen.add(key) + rows.append(row) + return rows + + def load_mappings(): mappings_file_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), @@ -195,7 +387,16 @@ def _assign_entries( specification, cache_dir, endpoints=None, + selected_entities=None, ): + selected_entity_order = { + ( + str(entity.get("reference", "")).strip(), + str(entity.get("organisation", "")).strip(), + ): idx + for idx, entity in enumerate(selected_entities or []) + if entity.get("reference") and entity.get("organisation") + } pipeline = Pipeline(pipeline_dir, dataset) resource_lookups = get_resource_unidentified_lookups( resource_path, @@ -223,10 +424,25 @@ def _assign_entries( # Track which entries are new by checking before adding new_entries_added = [] + selected_entries = [] + other_entries = [] for new_lookup in unassigned_entries: - for idx, entry in enumerate(new_lookup): - lookups.add_entry(entry[0]) - new_entries_added.append(entry[0]) + for entry in new_lookup: + entry_key = ( + str(entry[0].get("reference", "")).strip(), + str(entry[0].get("organisation", "")).strip(), + ) + if entry_key in selected_entity_order: + selected_entries.append((selected_entity_order[entry_key], entry[0])) + else: + other_entries.append(entry[0]) + + for _, entry in sorted(selected_entries, key=lambda selected: selected[0]): + lookups.add_entry(entry) + new_entries_added.append(entry) + for entry in other_entries: + lookups.add_entry(entry) + new_entries_added.append(entry) # save edited csvs max_entity_num = lookups.get_max_entity(pipeline.name, specification) @@ -270,7 +486,15 @@ def _transform_add_data_resource( pipeline_dir, specification, cache_dir, + selected_entities, ): + """ + Transform one add-data resource and assign entities when unknown rows exist. + + All new rows are assigned entity numbers, but selected rows are prioritised + and only selected rows are used for the entity-organisation summary when a + non-empty selection is supplied. + """ issues_log = pipeline.transform( input_path=resource_file_path, output_path=output_path, @@ -305,9 +529,13 @@ def _transform_add_data_resource( specification=specification, cache_dir=cache_dir, endpoints=endpoints if endpoints else None, + selected_entities=selected_entities, ) entity_org_mapping = _create_entity_organisation( - new_lookups, dataset, organisations[0], pipeline_dir + _filter_selected_entities(new_lookups, selected_entities), + dataset, + organisations[0], + pipeline_dir, ) # Reload pipeline to pick up newly saved lookups before rerunning transform. @@ -367,6 +595,8 @@ def fetch_add_data_response( cache_dir, endpoint, converted_path=None, + selected_entities=None, + selected_redirects=None, ): """ Run the add-data pipeline transform and build the pipeline summary response. @@ -426,6 +656,7 @@ def fetch_add_data_response( pipeline_dir=pipeline_dir, specification=specification, cache_dir=cache_dir, + selected_entities=selected_entities, ) existing_entities.extend(transformed_entities) @@ -438,7 +669,11 @@ def fetch_add_data_response( else: logger.info(f"No unidentified lookups found in {resource_file}") - new_entities_breakdown = _get_entities_breakdown(new_entities) + selected_new_entities = _filter_selected_entities( + new_entities, selected_entities + ) + new_entities_breakdown = _get_entities_breakdown(selected_new_entities) + all_entities_breakdown = _get_entities_breakdown(new_entities) existing_entities_breakdown = _get_existing_entities_breakdown( existing_entities ) @@ -450,6 +685,22 @@ def fetch_add_data_response( organisations[0], organisation, ) + selected_entity_ids = ( + {str(entity.get("entity")) for entity in selected_new_entities} + if selected_entities is not None and selected_entities != [] + else None + ) + old_entity_rows = _merge_old_entity_rows( + _create_auto_old_entity_redirects( + duplicate_candidates, selected_entity_ids=selected_entity_ids + ), + _create_old_entity_redirects( + new_entities, + selected_redirects, + selected_entities, + duplicate_candidates=duplicate_candidates, + ), + ) if issues_log: issues_log.add_severity_column(severity_mapping=specification.issue_type) @@ -458,8 +709,10 @@ def fetch_add_data_response( "new-in-resource": len(new_entities), "existing-in-resource": len(existing_entities), "new-entities": new_entities_breakdown, + "all-entities": all_entities_breakdown, "existing-entities": existing_entities_breakdown, "entity-organisation": entity_org_mapping, + "old-entity": old_entity_rows, "duplicate-candidates": duplicate_candidates, "pipeline-issues": ( [dict(issue) for issue in issues_log.rows] if issues_log else [] diff --git a/request-processor/src/application/core/workflow.py b/request-processor/src/application/core/workflow.py index 00ae761..fba3f0b 100644 --- a/request-processor/src/application/core/workflow.py +++ b/request-processor/src/application/core/workflow.py @@ -535,6 +535,8 @@ def add_data_workflow( github_branch=None, endpoint_parameters=None, endpoints=None, + selected_entities=None, + selected_redirects=None, ): """ Setup directories and download required CSVs to manage add-data pipeline @@ -556,6 +558,8 @@ def add_data_workflow( github_branch (str): Optional branch name to indicate if the data should be appended to a specific branch endpoint_parameters: Optional opaque value stored as the parameters field in the endpoint entry endpoints: Optional list of existing endpoint hashes associated with a resource + selected_entities: Optional list of reference/organisation pairs to assign entity numbers first + selected_redirects: Optional list of reference/organisation/old entity number values to redirect """ response_data = {} @@ -611,6 +615,8 @@ def add_data_workflow( cache_dir=directories.CACHE_DIR, endpoint=pipeline_endpoint, converted_path=converted_path, + selected_entities=selected_entities, + selected_redirects=selected_redirects, ) # Create endpoint and source summaries in workflow diff --git a/request-processor/src/tasks.py b/request-processor/src/tasks.py index f2b736a..98c9fc8 100644 --- a/request-processor/src/tasks.py +++ b/request-processor/src/tasks.py @@ -432,6 +432,8 @@ def add_data_task(request: Dict, directories=None): github_branch=request_data.github_branch, endpoint_parameters=request_data.endpoint_parameters, endpoints=endpoints, + selected_entities=getattr(request_data, "selected_entities", None), + selected_redirects=getattr(request_data, "selected_redirects", None), ) if "plugin" in log: response["plugin"] = log["plugin"] diff --git a/request-processor/tests/unit/src/application/core/test_pipeline.py b/request-processor/tests/unit/src/application/core/test_pipeline.py index 8bc740b..8b49bf2 100644 --- a/request-processor/tests/unit/src/application/core/test_pipeline.py +++ b/request-processor/tests/unit/src/application/core/test_pipeline.py @@ -1,9 +1,14 @@ import json import pytest -from unittest.mock import MagicMock +from datetime import date +from unittest.mock import MagicMock, call from src.application.core.pipeline import ( fetch_response_data, fetch_add_data_response, + _assign_entries, + _create_auto_old_entity_redirects, + _create_old_entity_redirects, + _filter_selected_entities, _get_entities_breakdown, _get_existing_entities_breakdown, _create_entity_organisation, @@ -279,6 +284,142 @@ def test_fetch_add_data_response_success(monkeypatch, tmp_path): assert "existing-in-resource" in result +def test_assign_entries_prioritises_selected_entities(monkeypatch, tmp_path): + pipeline_dir = tmp_path / "pipeline" + pipeline_dir.mkdir() + resource_path = tmp_path / "resource.csv" + resource_path.write_text("reference\nREF001\nREF002\n") + discovered_lookups = [ + [ + { + "prefix": "test-prefix", + "resource": "resource", + "organisation": "test-org", + "reference": "REF001", + "entity": "", + } + ], + [ + { + "prefix": "test-prefix", + "resource": "resource", + "organisation": "test-org", + "reference": "REF002", + "entity": "", + } + ], + ] + assigned_lookups = [ + {**discovered_lookups[1][0], "entity": "1000001"}, + {**discovered_lookups[0][0], "entity": "1000002"}, + ] + + mock_pipeline = MagicMock() + mock_pipeline.name = "test-dataset" + mock_lookups = MagicMock() + mock_lookups.lookups_path = str(pipeline_dir / "lookup.csv") + mock_lookups.get_max_entity.return_value = 1000000 + mock_lookups.save_csv.return_value = assigned_lookups + mock_lookups.entity_num_gen.state = {} + mock_specification = MagicMock() + mock_specification.get_dataset_entity_min.return_value = 1000000 + mock_specification.get_dataset_entity_max.return_value = 1999999 + + monkeypatch.setattr( + "src.application.core.pipeline.Pipeline", MagicMock(return_value=mock_pipeline) + ) + monkeypatch.setattr("src.application.core.pipeline.Lookups", lambda _: mock_lookups) + monkeypatch.setattr( + "src.application.core.pipeline.get_resource_unidentified_lookups", + MagicMock(return_value=discovered_lookups), + ) + + result = _assign_entries( + resource_path=str(resource_path), + dataset="test-dataset", + organisation="test-org", + pipeline_dir=str(pipeline_dir), + specification=mock_specification, + cache_dir=str(tmp_path), + selected_entities=[ + {"reference": "REF002", "organisation": "test-org"}, + ], + ) + + assert mock_lookups.add_entry.call_args_list == [ + call(discovered_lookups[1][0]), + call(discovered_lookups[0][0]), + ] + assert result == assigned_lookups + + +@pytest.mark.parametrize("selected_entities", [None, []]) +def test_assign_entries_assigns_all_when_selected_entities_empty_or_null( + monkeypatch, tmp_path, selected_entities +): + pipeline_dir = tmp_path / "pipeline" + pipeline_dir.mkdir() + resource_path = tmp_path / "resource.csv" + resource_path.write_text("reference\nREF001\nREF002\n") + discovered_lookups = [ + [ + { + "prefix": "test-prefix", + "resource": "resource", + "organisation": "test-org", + "reference": "REF001", + "entity": "", + } + ], + [ + { + "prefix": "test-prefix", + "resource": "resource", + "organisation": "test-org", + "reference": "REF002", + "entity": "", + } + ], + ] + assigned_lookups = [ + {**discovered_lookups[0][0], "entity": "1000001"}, + {**discovered_lookups[1][0], "entity": "1000002"}, + ] + + mock_pipeline = MagicMock() + mock_pipeline.name = "test-dataset" + mock_lookups = MagicMock() + mock_lookups.lookups_path = str(pipeline_dir / "lookup.csv") + mock_lookups.get_max_entity.return_value = 1000000 + mock_lookups.save_csv.return_value = assigned_lookups + mock_lookups.entity_num_gen.state = {} + mock_specification = MagicMock() + mock_specification.get_dataset_entity_min.return_value = 1000000 + mock_specification.get_dataset_entity_max.return_value = 1999999 + + monkeypatch.setattr( + "src.application.core.pipeline.Pipeline", MagicMock(return_value=mock_pipeline) + ) + monkeypatch.setattr("src.application.core.pipeline.Lookups", lambda _: mock_lookups) + monkeypatch.setattr( + "src.application.core.pipeline.get_resource_unidentified_lookups", + MagicMock(return_value=discovered_lookups), + ) + + result = _assign_entries( + resource_path=str(resource_path), + dataset="test-dataset", + organisation="test-org", + pipeline_dir=str(pipeline_dir), + specification=mock_specification, + cache_dir=str(tmp_path), + selected_entities=selected_entities, + ) + + assert mock_lookups.add_entry.call_count == 2 + assert result == assigned_lookups + + def test_fetch_add_data_response_no_files(monkeypatch, tmp_path): """Test when input directory has no files""" dataset = "test-dataset" @@ -310,6 +451,136 @@ def test_fetch_add_data_response_no_files(monkeypatch, tmp_path): assert result["new-in-resource"] == 0 +def test_fetch_add_data_response_includes_selected_old_entity_redirects( + monkeypatch, tmp_path +): + dataset = "test-dataset" + organisation = "test-org" + pipeline_dir = tmp_path / "pipeline" + input_path = tmp_path / "resource" + cache_dir = tmp_path / "cache" + endpoint = "abc123hash" + + input_path.mkdir(parents=True) + pipeline_dir.mkdir(parents=True) + (input_path / "test.csv").write_text("reference\nREF001\nREF002") + + mock_pipeline = MagicMock() + mock_pipeline.path = str(pipeline_dir) + mock_pipeline.redirect_lookups.return_value = {} + monkeypatch.setattr( + "src.application.core.pipeline.Pipeline", MagicMock(return_value=mock_pipeline) + ) + monkeypatch.setattr("src.application.core.pipeline.Organisation", MagicMock()) + mock_api = MagicMock() + mock_api.get_valid_category_values.return_value = {} + monkeypatch.setattr( + "src.application.core.pipeline.API", MagicMock(return_value=mock_api) + ) + monkeypatch.setattr( + "src.application.core.pipeline._find_duplicate_candidates", + MagicMock( + return_value=[ + { + "old_entity": "900002", + "entity": "1000002", + "match_type": "complete_match", + "name_similarity": "", + "evidence": "geometry complete_match, name similarity 100%", + "old_entity_redirects": [], + }, + { + "old_entity": "900001", + "entity": "1000002", + "match_type": "single_match", + "name_similarity": "85%", + "evidence": "geometry single_match, name similarity 85%", + "old_entity_redirects": [], + }, + ] + ), + ) + issue_log = MagicMock() + issue_log.rows = [] + monkeypatch.setattr( + "src.application.core.pipeline._process_add_data_resource", + MagicMock( + return_value=( + mock_pipeline, + issue_log, + [], + [ + { + "entity": "1000001", + "reference": "REF001", + "organisation": "test-org", + }, + { + "entity": "1000002", + "reference": "REF002", + "organisation": "test-org", + }, + ], + [], + ) + ), + ) + + result = fetch_add_data_response( + dataset=dataset, + organisation_provider=organisation, + pipeline_dir=str(pipeline_dir), + input_dir=str(input_path), + output_path=str(input_path / "output.csv"), + specification=MagicMock(), + cache_dir=str(cache_dir), + endpoint=endpoint, + selected_entities=[{"reference": "REF002", "organisation": "test-org"}], + selected_redirects=[ + { + "reference": "REF002", + "organisation": "test-org", + "old_entity_number": "900001", + } + ], + ) + + assert result["old-entity"] == [ + { + "old-entity": "900002", + "status": "301", + "entity": "1000002", + "entry-date": date.today().isoformat(), + "notes": "geometry complete_match, name similarity 100%", + }, + { + "old-entity": "900001", + "status": "301", + "entity": "1000002", + "entry-date": date.today().isoformat(), + "notes": "geometry single_match, name similarity 85%", + }, + ] + assert result["new-entities"] == [ + { + "entity": "1000002", + "prefix": "", + "end-date": "", + "endpoint": "", + "resource": "", + "reference": "REF002", + "entry-date": "", + "start-date": "", + "entry-number": "", + "organisation": "test-org", + } + ] + assert [entity["reference"] for entity in result["all-entities"]] == [ + "REF001", + "REF002", + ] + + def test_fetch_add_data_response_file_not_found(monkeypatch, tmp_path): """Test when input path does not exist""" dataset = "test-dataset" @@ -490,6 +761,234 @@ def test_get_entities_breakdown_missing_fields(): # --- _create_entity_organisation --- +def test_filter_selected_entities_returns_only_selected_matches(): + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, + ] + + result = _filter_selected_entities( + new_entities, + [{"reference": "REF002", "organisation": "test-org"}], + ) + + assert result == [new_entities[1]] + + +@pytest.mark.parametrize("selected_entities", [None, []]) +def test_filter_selected_entities_returns_all_when_selection_empty_or_null( + selected_entities, +): + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, + ] + + assert _filter_selected_entities(new_entities, selected_entities) == new_entities + + +def test_filter_selected_entities_returns_empty_for_invalid_non_empty_selection(): + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + ] + + assert _filter_selected_entities(new_entities, [{"reference": "REF001"}]) == [] + + +def test_create_entity_organisation_uses_selected_entity_subset(tmp_path): + pipeline_dir = tmp_path / "pipeline" + pipeline_dir.mkdir() + (pipeline_dir / "entity-organisation.csv").write_text( + "dataset,entity-minimum,entity-maximum,organisation\n" + ) + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, + ] + + result = _create_entity_organisation( + _filter_selected_entities( + new_entities, + [{"reference": "REF002", "organisation": "test-org"}], + ), + "test-dataset", + "test-org", + str(pipeline_dir), + ) + + assert result[0]["entity-minimum"] == 1000002 + assert result[0]["entity-maximum"] == 1000002 + + +def test_create_old_entity_redirects_from_selected_redirects(): + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, + ] + + result = _create_old_entity_redirects( + new_entities, + [ + { + "reference": "REF002", + "organisation": "test-org", + "old_entity_number": "900001", + } + ], + duplicate_candidates=[ + { + "old_entity": "900001", + "entity": "1000002", + "evidence": "geometry single_match, name similarity 85%", + } + ], + ) + + assert result == [ + { + "old-entity": "900001", + "status": "301", + "entity": "1000002", + "entry-date": date.today().isoformat(), + "notes": "geometry single_match, name similarity 85%", + } + ] + + +def test_create_old_entity_redirects_ignores_redirects_for_unselected_entities(): + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, + ] + + result = _create_old_entity_redirects( + new_entities, + [ + { + "reference": "REF001", + "organisation": "test-org", + "old_entity_number": "900001", + } + ], + selected_entities=[{"reference": "REF002", "organisation": "test-org"}], + ) + + assert result == [] + + +def test_create_auto_old_entity_redirects_for_complete_matches(): + result = _create_auto_old_entity_redirects( + [ + { + "old_entity": "900001", + "entity": "1000001", + "match_type": "complete_match", + "name_similarity": "", + "evidence": "geometry complete_match", + "old_entity_redirects": [], + } + ], + ) + + assert result == [ + { + "old-entity": "900001", + "status": "301", + "entity": "1000001", + "entry-date": date.today().isoformat(), + "notes": "geometry complete_match", + } + ] + + +def test_create_auto_old_entity_redirects_for_single_matches_above_85_percent(): + result = _create_auto_old_entity_redirects( + [ + { + "old_entity": "900001", + "entity": "1000001", + "match_type": "single_match", + "name_similarity": "86%", + "evidence": "geometry single_match, name similarity 86%", + "old_entity_redirects": [], + }, + { + "old_entity": "900002", + "entity": "1000002", + "match_type": "single_match", + "name_similarity": "85%", + "evidence": "geometry single_match, name similarity 85%", + "old_entity_redirects": [], + }, + ], + ) + + assert result == [ + { + "old-entity": "900001", + "status": "301", + "entity": "1000001", + "entry-date": date.today().isoformat(), + "notes": "geometry single_match, name similarity 86%", + } + ] + + +def test_create_auto_old_entity_redirects_ignores_existing_redirects_and_unselected_entities(): + result = _create_auto_old_entity_redirects( + [ + { + "old_entity": "900001", + "entity": "1000001", + "match_type": "complete_match", + "name_similarity": "", + "old_entity_redirects": [], + }, + { + "old_entity": "900002", + "entity": "1000002", + "match_type": "complete_match", + "name_similarity": "", + "old_entity_redirects": [{"old-entity": "900002"}], + }, + ], + selected_entity_ids={"1000002"}, + ) + + assert result == [] + + +@pytest.mark.parametrize("selected_redirects", [None, []]) +def test_create_old_entity_redirects_returns_empty_when_selection_empty_or_null( + selected_redirects, +): + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + ] + + assert _create_old_entity_redirects(new_entities, selected_redirects) == [] + + +def test_create_old_entity_redirects_ignores_unassigned_or_invalid_redirects(): + new_entities = [ + {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, + ] + + result = _create_old_entity_redirects( + new_entities, + [ + { + "reference": "REF001", + "organisation": "other-org", + "old_entity_number": "900001", + }, + {"reference": "REF001", "organisation": "test-org"}, + ], + ) + + assert result == [] + + def test_create_entity_organisation_sets_overlap_true_when_range_already_present( tmp_path, ): diff --git a/request-processor/tests/unit/src/application/core/test_workflow.py b/request-processor/tests/unit/src/application/core/test_workflow.py index 6d4f553..134be6a 100644 --- a/request-processor/tests/unit/src/application/core/test_workflow.py +++ b/request-processor/tests/unit/src/application/core/test_workflow.py @@ -414,6 +414,8 @@ def fake_fetch_add_data_response( cache_dir, endpoint, converted_path=None, + selected_entities=None, + selected_redirects=None, ): called["fetch_add_data_response"] = { "dataset": dataset, @@ -424,6 +426,8 @@ def fake_fetch_add_data_response( "specification": specification, "cache_dir": cache_dir, "endpoint": endpoint, + "selected_entities": selected_entities, + "selected_redirects": selected_redirects, } return {"result": "ok"} @@ -476,6 +480,8 @@ def fake_fetch_add_data_response( assert called["fetch_add_data_response"]["cache_dir"] == directories.CACHE_DIR expected_endpoint_hash = hashlib.sha256(url.encode("utf-8")).hexdigest() assert called["fetch_add_data_response"]["endpoint"] == expected_endpoint_hash + assert called["fetch_add_data_response"]["selected_entities"] is None + assert called["fetch_add_data_response"]["selected_redirects"] is None def test_add_data_workflow_with_resource_endpoints(monkeypatch): diff --git a/request-processor/tests/unit/src/test_tasks.py b/request-processor/tests/unit/src/test_tasks.py index 72a0b62..1e77e78 100644 --- a/request-processor/tests/unit/src/test_tasks.py +++ b/request-processor/tests/unit/src/test_tasks.py @@ -272,6 +272,8 @@ def test_add_data_task_success_with_resource(monkeypatch): request_schema.params.column_mapping = None request_schema.params.github_branch = None request_schema.params.endpoint_parameters = None + request_schema.params.selected_entities = None + request_schema.params.selected_redirects = None workflow_call = {} @@ -314,6 +316,7 @@ def fake_add_data_workflow(*args, **kwargs): assert workflow_call["args"][8] == "ogl3" assert workflow_call["args"][9] == "2024-01-01" assert workflow_call["kwargs"]["endpoints"] == ["endpoint-1"] + assert workflow_call["kwargs"]["selected_redirects"] is None def test_add_data_task_fail(monkeypatch): diff --git a/request_model/schemas.py b/request_model/schemas.py index 7330211..31be553 100644 --- a/request_model/schemas.py +++ b/request_model/schemas.py @@ -35,6 +35,7 @@ class CheckUrlParams(Params): type: Literal[RequestTypeEnum.check_url] = RequestTypeEnum.check_url url: str + class AddDataParams(Params): type: Literal[RequestTypeEnum.add_data] = RequestTypeEnum.add_data url: Optional[str] = None @@ -46,6 +47,8 @@ class AddDataParams(Params): organisation: Optional[str] = None plugin: Optional[PluginTypeEnum] = None github_branch: Optional[str] = None + selected_entities: Optional[List[Dict[str, str]]] = None + selected_redirects: Optional[Union[List[Dict[str, str]], Dict[str, str]]] = None class RequestBase(BaseModel): From a96405c69822fa895e5e17c357ed4cbe2bde2f2f Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Mon, 20 Jul 2026 16:16:27 +0100 Subject: [PATCH 2/6] Add docstrings to several functions in pipeline.py for improved clarity --- .../src/application/core/pipeline.py | 32 +++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index 7eb5620..356fc03 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -209,6 +209,7 @@ def _merge_old_entity_rows(*row_groups): def load_mappings(): + """Load task summary templates keyed by field and issue type.""" mappings_file_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "../application/configs/mapping.yaml", @@ -220,6 +221,7 @@ def load_mappings(): def _format_task_summary(details_str, task_source, mappings): + """Create a human-readable task summary from task log detail JSON.""" try: details = json.loads(details_str) except (json.JSONDecodeError, TypeError): @@ -253,6 +255,7 @@ def run_task_pipeline( column_field_path=None, mandatory_fields=None, ): + """Run the task pipeline and attach generated summary text to each task row.""" task_pipeline = TaskPipeline() status = task_pipeline.run( output_path=task_log_path, @@ -294,6 +297,13 @@ def fetch_response_data( additional_col_mappings, additional_concats, ): + """ + Run the standard request pipeline and write transformed data plus issue logs. + + This path is used by non-add-data workflows. It assigns unknown entities + before transforming, then saves issue, column-field, and dataset-resource + logs for each uploaded resource. + """ pipeline = Pipeline(pipeline_dir, dataset, specification=specification) api = API(specification=specification) @@ -376,6 +386,7 @@ def fetch_response_data( def resource_from_path(path): + """Return the resource hash/name from an uploaded resource file path.""" return Path(path).stem @@ -389,6 +400,14 @@ def _assign_entries( endpoints=None, selected_entities=None, ): + """ + Assign entity numbers for unidentified lookups in a resource. + + When selected_entities is supplied, matching rows are added to lookup.csv + first so they receive the lowest new entity numbers. Unselected rows are + still added afterwards; selection affects ordering, not whether a row is + assigned. + """ selected_entity_order = { ( str(entity.get("reference", "")).strip(), @@ -556,6 +575,7 @@ def _transform_add_data_resource( def _process_add_data_resource(resource_file, **kwargs): + """Wrap one add-data resource transform with resource-specific logging.""" try: return _transform_add_data_resource(**kwargs) except Exception as err: @@ -571,6 +591,12 @@ def _find_duplicate_candidates( organisation_provider, organisation_index, ): + """ + Find possible redirects for the transformed add-data output. + + Duplicate analysis should not block add-data processing, so failures are + logged and returned as an empty candidate list. + """ try: return find_duplicate_redirect_candidates( dataset=dataset, @@ -604,6 +630,12 @@ def fetch_add_data_response( This is reached via POST /requests with type "add_data" through the AddDataTask and add_data_workflow. Processing exceptions are re-raised so add_data_workflow can return them in the standard async error response. + + selected_entities controls summary filtering and assignment order: + None or [] means all new entities are reported as new-entities, while a + non-empty list reports only those selected rows there and keeps the full + assignment list in all-entities. selected_redirects is only used to create + explicit old-entity redirect rows; None or [] means no manual redirects. """ try: pipeline = Pipeline(pipeline_dir, dataset, specification=specification) From 9d2bbaa520b20462268755259310d9a30e2d44b7 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Tue, 21 Jul 2026 20:20:20 +0100 Subject: [PATCH 3/6] Refactor entity selection logic --- .../src/application/core/pipeline.py | 180 ++++++++---------- .../src/application/core/workflow.py | 8 +- request-processor/src/tasks.py | 4 +- .../src/application/core/test_pipeline.py | 82 ++++---- .../src/application/core/test_workflow.py | 6 +- .../tests/unit/src/test_tasks.py | 2 +- request_model/schemas.py | 4 +- 7 files changed, 130 insertions(+), 156 deletions(-) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index 356fc03..2c4ea30 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -17,107 +17,102 @@ logger = get_logger(__name__) -def _selected_entity_keys(selected_entities): - """Return the stable keys used to match selected add-data rows.""" +def _reference_set(references): + """Return normalised reference values from an optional request list.""" + if isinstance(references, str): + references = [references] return { - ( - str(entity.get("reference", "")).strip(), - str(entity.get("organisation", "")).strip(), - ) - for entity in selected_entities or [] - if entity.get("reference") and entity.get("organisation") + str( + reference.get("reference") if isinstance(reference, dict) else reference + ).strip() + for reference in references or [] + if reference } -def _filter_selected_entities(new_entities, selected_entities): +def _redirect_reference(redirect): + """Return the reference from a selected redirect object.""" + if isinstance(redirect, dict): + return str(redirect.get("reference", "")).strip() + return str(redirect or "").strip() + + +def _redirect_old_entity(redirect): + """Return the old entity number from a selected redirect object.""" + if not isinstance(redirect, dict): + return "" + return str(redirect.get("old_entity_number", "")).strip() + + +def _filter_selected_entities(new_entities, excluded_references): """ - Return selected entities, treating a missing or empty selection as all entities. + Return entities not listed in excluded_references. - This mirrors the add-data request contract: selected_entities=None or [] - means process every new entity. A non-empty but invalid selection matches none. + This mirrors the add-data request contract: excluded_references=None + or [] means every new entity is selected. A non-empty list removes only + entities with matching references from selected outputs. """ - if selected_entities is None or selected_entities == []: + excluded_references = _reference_set(excluded_references) + if not excluded_references: return new_entities - selected_keys = _selected_entity_keys(selected_entities) - if not selected_keys: - return [] - return [ entity for entity in new_entities - if ( - str(entity.get("reference", "")).strip(), - str(entity.get("organisation", "")).strip(), - ) - in selected_keys + if str(entity.get("reference", "")).strip() not in excluded_references ] -def _normalise_selection_list(selection): - """Normalise optional single-object/list request selections to a list.""" - if selection is None or selection == []: - return [] - if isinstance(selection, dict): - return [selection] - if not isinstance(selection, list): - return [] - return selection - - def _create_old_entity_redirects( new_entities, selected_redirects, - selected_entities=None, + excluded_references=None, duplicate_candidates=None, ): """ Build old-entity rows from explicit redirect selections. - Redirects are only created for assigned entities matching the selected - reference and organisation. When duplicate evidence exists for the same - old/new entity pair, it is copied into the notes field. + selected_redirects is a list of objects containing a new entity reference + and old entity number. Duplicate candidates provide the evidence note. + References excluded by excluded_references cannot be redirected. """ - selected_redirects = _normalise_selection_list(selected_redirects) if not selected_redirects: return [] - selected_entity_keys = _selected_entity_keys(selected_entities) - new_entity_by_key = { - ( - str(entity.get("reference", "")).strip(), - str(entity.get("organisation", "")).strip(), - ): entity - for entity in new_entities + selected_new_entities = _filter_selected_entities(new_entities, excluded_references) + selected_entity_ids = { + str(entity.get("entity", "")).strip() for entity in selected_new_entities } + + old_entity_rows = [] + seen = set() evidence_by_redirect = { ( + str(candidate.get("new_reference", "")).strip(), str(candidate.get("old_entity", "")).strip(), str(candidate.get("entity", "")).strip(), ): str(candidate.get("evidence", "") or "") for candidate in duplicate_candidates or [] } - old_entity_rows = [] - seen = set() - for redirect in selected_redirects: - old_entity_number = str(redirect.get("old_entity_number", "")).strip() - key = ( - str(redirect.get("reference", "")).strip(), - str(redirect.get("organisation", "")).strip(), - ) - if selected_entity_keys and key not in selected_entity_keys: - continue + entity_by_reference = { + str(entity.get("reference", "")).strip(): str(entity.get("entity", "")).strip() + for entity in selected_new_entities + } - new_entity = new_entity_by_key.get(key) - if not old_entity_number or not new_entity or not new_entity.get("entity"): + for redirect in selected_redirects: + reference = _redirect_reference(redirect) + old_entity = _redirect_old_entity(redirect) + entity = entity_by_reference.get(reference, "") + if not old_entity or not entity: continue - new_entity_number = new_entity.get("entity") - notes = evidence_by_redirect.get( - (old_entity_number, str(new_entity_number).strip()), "" + notes = evidence_by_redirect.get((reference, old_entity, entity), "") + row = _old_entity_row( + old_entity, + entity, + notes=notes, ) - row = _old_entity_row(old_entity_number, new_entity_number, notes=notes) if not row: continue @@ -398,24 +393,17 @@ def _assign_entries( specification, cache_dir, endpoints=None, - selected_entities=None, + excluded_references=None, ): """ Assign entity numbers for unidentified lookups in a resource. - When selected_entities is supplied, matching rows are added to lookup.csv - first so they receive the lowest new entity numbers. Unselected rows are + Rows not listed in excluded_references are added to lookup.csv first so + selected rows receive the lowest new entity numbers. Excluded rows are still added afterwards; selection affects ordering, not whether a row is assigned. """ - selected_entity_order = { - ( - str(entity.get("reference", "")).strip(), - str(entity.get("organisation", "")).strip(), - ): idx - for idx, entity in enumerate(selected_entities or []) - if entity.get("reference") and entity.get("organisation") - } + excluded_references = _reference_set(excluded_references) pipeline = Pipeline(pipeline_dir, dataset) resource_lookups = get_resource_unidentified_lookups( resource_path, @@ -447,16 +435,13 @@ def _assign_entries( other_entries = [] for new_lookup in unassigned_entries: for entry in new_lookup: - entry_key = ( - str(entry[0].get("reference", "")).strip(), - str(entry[0].get("organisation", "")).strip(), - ) - if entry_key in selected_entity_order: - selected_entries.append((selected_entity_order[entry_key], entry[0])) - else: + entry_reference = str(entry[0].get("reference", "")).strip() + if entry_reference in excluded_references: other_entries.append(entry[0]) + else: + selected_entries.append(entry[0]) - for _, entry in sorted(selected_entries, key=lambda selected: selected[0]): + for entry in selected_entries: lookups.add_entry(entry) new_entries_added.append(entry) for entry in other_entries: @@ -505,14 +490,14 @@ def _transform_add_data_resource( pipeline_dir, specification, cache_dir, - selected_entities, + excluded_references, ): """ Transform one add-data resource and assign entities when unknown rows exist. - All new rows are assigned entity numbers, but selected rows are prioritised - and only selected rows are used for the entity-organisation summary when a - non-empty selection is supplied. + All new rows are assigned entity numbers, but rows not listed in + excluded_references are prioritised and used for the + entity-organisation summary. """ issues_log = pipeline.transform( input_path=resource_file_path, @@ -548,10 +533,10 @@ def _transform_add_data_resource( specification=specification, cache_dir=cache_dir, endpoints=endpoints if endpoints else None, - selected_entities=selected_entities, + excluded_references=excluded_references, ) entity_org_mapping = _create_entity_organisation( - _filter_selected_entities(new_lookups, selected_entities), + _filter_selected_entities(new_lookups, excluded_references), dataset, organisations[0], pipeline_dir, @@ -621,7 +606,7 @@ def fetch_add_data_response( cache_dir, endpoint, converted_path=None, - selected_entities=None, + excluded_references=None, selected_redirects=None, ): """ @@ -631,11 +616,12 @@ def fetch_add_data_response( AddDataTask and add_data_workflow. Processing exceptions are re-raised so add_data_workflow can return them in the standard async error response. - selected_entities controls summary filtering and assignment order: - None or [] means all new entities are reported as new-entities, while a - non-empty list reports only those selected rows there and keeps the full - assignment list in all-entities. selected_redirects is only used to create - explicit old-entity redirect rows; None or [] means no manual redirects. + excluded_references controls summary filtering and assignment + order: None or [] means all new entities are reported as new-entities, + while a non-empty list excludes those references from selected outputs. + selected_redirects is a list of reference/old entity number objects used + to create explicit old-entity redirect rows; None or [] means no manual + redirects. """ try: pipeline = Pipeline(pipeline_dir, dataset, specification=specification) @@ -688,7 +674,7 @@ def fetch_add_data_response( pipeline_dir=pipeline_dir, specification=specification, cache_dir=cache_dir, - selected_entities=selected_entities, + excluded_references=excluded_references, ) existing_entities.extend(transformed_entities) @@ -702,10 +688,9 @@ def fetch_add_data_response( logger.info(f"No unidentified lookups found in {resource_file}") selected_new_entities = _filter_selected_entities( - new_entities, selected_entities + new_entities, excluded_references ) new_entities_breakdown = _get_entities_breakdown(selected_new_entities) - all_entities_breakdown = _get_entities_breakdown(new_entities) existing_entities_breakdown = _get_existing_entities_breakdown( existing_entities ) @@ -719,7 +704,7 @@ def fetch_add_data_response( ) selected_entity_ids = ( {str(entity.get("entity")) for entity in selected_new_entities} - if selected_entities is not None and selected_entities != [] + if excluded_references is not None and excluded_references != [] else None ) old_entity_rows = _merge_old_entity_rows( @@ -729,7 +714,7 @@ def fetch_add_data_response( _create_old_entity_redirects( new_entities, selected_redirects, - selected_entities, + excluded_references, duplicate_candidates=duplicate_candidates, ), ) @@ -741,7 +726,6 @@ def fetch_add_data_response( "new-in-resource": len(new_entities), "existing-in-resource": len(existing_entities), "new-entities": new_entities_breakdown, - "all-entities": all_entities_breakdown, "existing-entities": existing_entities_breakdown, "entity-organisation": entity_org_mapping, "old-entity": old_entity_rows, diff --git a/request-processor/src/application/core/workflow.py b/request-processor/src/application/core/workflow.py index fba3f0b..c4721c3 100644 --- a/request-processor/src/application/core/workflow.py +++ b/request-processor/src/application/core/workflow.py @@ -535,7 +535,7 @@ def add_data_workflow( github_branch=None, endpoint_parameters=None, endpoints=None, - selected_entities=None, + excluded_references=None, selected_redirects=None, ): """ @@ -558,8 +558,8 @@ def add_data_workflow( github_branch (str): Optional branch name to indicate if the data should be appended to a specific branch endpoint_parameters: Optional opaque value stored as the parameters field in the endpoint entry endpoints: Optional list of existing endpoint hashes associated with a resource - selected_entities: Optional list of reference/organisation pairs to assign entity numbers first - selected_redirects: Optional list of reference/organisation/old entity number values to redirect + excluded_references: Optional list of references to exclude from selected entity outputs + selected_redirects: Optional list of reference/old entity number values to redirect """ response_data = {} @@ -615,7 +615,7 @@ def add_data_workflow( cache_dir=directories.CACHE_DIR, endpoint=pipeline_endpoint, converted_path=converted_path, - selected_entities=selected_entities, + excluded_references=excluded_references, selected_redirects=selected_redirects, ) diff --git a/request-processor/src/tasks.py b/request-processor/src/tasks.py index 32a053e..89c1468 100644 --- a/request-processor/src/tasks.py +++ b/request-processor/src/tasks.py @@ -432,7 +432,9 @@ def add_data_task(request: Dict, directories=None): github_branch=request_data.github_branch, endpoint_parameters=request_data.endpoint_parameters, endpoints=endpoints, - selected_entities=getattr(request_data, "selected_entities", None), + excluded_references=getattr( + request_data, "excluded_references", None + ), selected_redirects=getattr(request_data, "selected_redirects", None), ) if "plugin" in log: diff --git a/request-processor/tests/unit/src/application/core/test_pipeline.py b/request-processor/tests/unit/src/application/core/test_pipeline.py index 8b49bf2..55a022b 100644 --- a/request-processor/tests/unit/src/application/core/test_pipeline.py +++ b/request-processor/tests/unit/src/application/core/test_pipeline.py @@ -284,7 +284,9 @@ def test_fetch_add_data_response_success(monkeypatch, tmp_path): assert "existing-in-resource" in result -def test_assign_entries_prioritises_selected_entities(monkeypatch, tmp_path): +def test_assign_entries_prioritises_rows_not_in_excluded_references( + monkeypatch, tmp_path +): pipeline_dir = tmp_path / "pipeline" pipeline_dir.mkdir() resource_path = tmp_path / "resource.csv" @@ -341,9 +343,7 @@ def test_assign_entries_prioritises_selected_entities(monkeypatch, tmp_path): pipeline_dir=str(pipeline_dir), specification=mock_specification, cache_dir=str(tmp_path), - selected_entities=[ - {"reference": "REF002", "organisation": "test-org"}, - ], + excluded_references=["REF001"], ) assert mock_lookups.add_entry.call_args_list == [ @@ -353,9 +353,9 @@ def test_assign_entries_prioritises_selected_entities(monkeypatch, tmp_path): assert result == assigned_lookups -@pytest.mark.parametrize("selected_entities", [None, []]) -def test_assign_entries_assigns_all_when_selected_entities_empty_or_null( - monkeypatch, tmp_path, selected_entities +@pytest.mark.parametrize("excluded_references", [None, []]) +def test_assign_entries_keeps_original_order_when_excluded_references_empty_or_null( + monkeypatch, tmp_path, excluded_references ): pipeline_dir = tmp_path / "pipeline" pipeline_dir.mkdir() @@ -413,7 +413,7 @@ def test_assign_entries_assigns_all_when_selected_entities_empty_or_null( pipeline_dir=str(pipeline_dir), specification=mock_specification, cache_dir=str(tmp_path), - selected_entities=selected_entities, + excluded_references=excluded_references, ) assert mock_lookups.add_entry.call_count == 2 @@ -485,6 +485,7 @@ def test_fetch_add_data_response_includes_selected_old_entity_redirects( "old_entity": "900002", "entity": "1000002", "match_type": "complete_match", + "new_reference": "REF002", "name_similarity": "", "evidence": "geometry complete_match, name similarity 100%", "old_entity_redirects": [], @@ -493,6 +494,7 @@ def test_fetch_add_data_response_includes_selected_old_entity_redirects( "old_entity": "900001", "entity": "1000002", "match_type": "single_match", + "new_reference": "REF002", "name_similarity": "85%", "evidence": "geometry single_match, name similarity 85%", "old_entity_redirects": [], @@ -535,13 +537,9 @@ def test_fetch_add_data_response_includes_selected_old_entity_redirects( specification=MagicMock(), cache_dir=str(cache_dir), endpoint=endpoint, - selected_entities=[{"reference": "REF002", "organisation": "test-org"}], + excluded_references=["REF001"], selected_redirects=[ - { - "reference": "REF002", - "organisation": "test-org", - "old_entity_number": "900001", - } + {"reference": "REF002", "old_entity_number": "900001"}, ], ) @@ -575,10 +573,7 @@ def test_fetch_add_data_response_includes_selected_old_entity_redirects( "organisation": "test-org", } ] - assert [entity["reference"] for entity in result["all-entities"]] == [ - "REF001", - "REF002", - ] + assert "all-entities" not in result def test_fetch_add_data_response_file_not_found(monkeypatch, tmp_path): @@ -761,7 +756,7 @@ def test_get_entities_breakdown_missing_fields(): # --- _create_entity_organisation --- -def test_filter_selected_entities_returns_only_selected_matches(): +def test_filter_selected_entities_excludes_requested_references(): new_entities = [ {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, @@ -769,30 +764,30 @@ def test_filter_selected_entities_returns_only_selected_matches(): result = _filter_selected_entities( new_entities, - [{"reference": "REF002", "organisation": "test-org"}], + ["REF001"], ) assert result == [new_entities[1]] -@pytest.mark.parametrize("selected_entities", [None, []]) -def test_filter_selected_entities_returns_all_when_selection_empty_or_null( - selected_entities, +@pytest.mark.parametrize("excluded_references", [None, []]) +def test_filter_selected_entities_returns_all_when_excluded_references_empty_or_null( + excluded_references, ): new_entities = [ {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, ] - assert _filter_selected_entities(new_entities, selected_entities) == new_entities + assert _filter_selected_entities(new_entities, excluded_references) == new_entities -def test_filter_selected_entities_returns_empty_for_invalid_non_empty_selection(): +def test_filter_selected_entities_returns_all_for_non_matching_excluded_reference(): new_entities = [ {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, ] - assert _filter_selected_entities(new_entities, [{"reference": "REF001"}]) == [] + assert _filter_selected_entities(new_entities, ["REF999"]) == new_entities def test_create_entity_organisation_uses_selected_entity_subset(tmp_path): @@ -809,7 +804,7 @@ def test_create_entity_organisation_uses_selected_entity_subset(tmp_path): result = _create_entity_organisation( _filter_selected_entities( new_entities, - [{"reference": "REF002", "organisation": "test-org"}], + ["REF001"], ), "test-dataset", "test-org", @@ -828,17 +823,12 @@ def test_create_old_entity_redirects_from_selected_redirects(): result = _create_old_entity_redirects( new_entities, - [ - { - "reference": "REF002", - "organisation": "test-org", - "old_entity_number": "900001", - } - ], + [{"reference": "REF002", "old_entity_number": "900001"}], duplicate_candidates=[ { "old_entity": "900001", "entity": "1000002", + "new_reference": "REF002", "evidence": "geometry single_match, name similarity 85%", } ], @@ -855,7 +845,7 @@ def test_create_old_entity_redirects_from_selected_redirects(): ] -def test_create_old_entity_redirects_ignores_redirects_for_unselected_entities(): +def test_create_old_entity_redirects_ignores_redirects_for_excluded_references(): new_entities = [ {"entity": "1000001", "reference": "REF001", "organisation": "test-org"}, {"entity": "1000002", "reference": "REF002", "organisation": "test-org"}, @@ -863,14 +853,15 @@ def test_create_old_entity_redirects_ignores_redirects_for_unselected_entities() result = _create_old_entity_redirects( new_entities, - [ + [{"reference": "REF001", "old_entity_number": "900001"}], + excluded_references=["REF001"], + duplicate_candidates=[ { - "reference": "REF001", - "organisation": "test-org", - "old_entity_number": "900001", + "old_entity": "900001", + "entity": "1000001", + "new_reference": "REF001", } ], - selected_entities=[{"reference": "REF002", "organisation": "test-org"}], ) assert result == [] @@ -934,7 +925,7 @@ def test_create_auto_old_entity_redirects_for_single_matches_above_85_percent(): ] -def test_create_auto_old_entity_redirects_ignores_existing_redirects_and_unselected_entities(): +def test_create_auto_old_entity_redirects_ignores_existing_redirects_and_unselected_ids(): result = _create_auto_old_entity_redirects( [ { @@ -977,13 +968,10 @@ def test_create_old_entity_redirects_ignores_unassigned_or_invalid_redirects(): result = _create_old_entity_redirects( new_entities, [ - { - "reference": "REF001", - "organisation": "other-org", - "old_entity_number": "900001", - }, - {"reference": "REF001", "organisation": "test-org"}, + {"reference": "REF001"}, + {"old_entity_number": "900001"}, ], + duplicate_candidates=[], ) assert result == [] diff --git a/request-processor/tests/unit/src/application/core/test_workflow.py b/request-processor/tests/unit/src/application/core/test_workflow.py index 134be6a..e44554d 100644 --- a/request-processor/tests/unit/src/application/core/test_workflow.py +++ b/request-processor/tests/unit/src/application/core/test_workflow.py @@ -414,7 +414,7 @@ def fake_fetch_add_data_response( cache_dir, endpoint, converted_path=None, - selected_entities=None, + excluded_references=None, selected_redirects=None, ): called["fetch_add_data_response"] = { @@ -426,7 +426,7 @@ def fake_fetch_add_data_response( "specification": specification, "cache_dir": cache_dir, "endpoint": endpoint, - "selected_entities": selected_entities, + "excluded_references": excluded_references, "selected_redirects": selected_redirects, } return {"result": "ok"} @@ -480,7 +480,7 @@ def fake_fetch_add_data_response( assert called["fetch_add_data_response"]["cache_dir"] == directories.CACHE_DIR expected_endpoint_hash = hashlib.sha256(url.encode("utf-8")).hexdigest() assert called["fetch_add_data_response"]["endpoint"] == expected_endpoint_hash - assert called["fetch_add_data_response"]["selected_entities"] is None + assert called["fetch_add_data_response"]["excluded_references"] is None assert called["fetch_add_data_response"]["selected_redirects"] is None diff --git a/request-processor/tests/unit/src/test_tasks.py b/request-processor/tests/unit/src/test_tasks.py index 1e77e78..242c448 100644 --- a/request-processor/tests/unit/src/test_tasks.py +++ b/request-processor/tests/unit/src/test_tasks.py @@ -272,7 +272,7 @@ def test_add_data_task_success_with_resource(monkeypatch): request_schema.params.column_mapping = None request_schema.params.github_branch = None request_schema.params.endpoint_parameters = None - request_schema.params.selected_entities = None + request_schema.params.excluded_references = None request_schema.params.selected_redirects = None workflow_call = {} diff --git a/request_model/schemas.py b/request_model/schemas.py index 31be553..37f8ae6 100644 --- a/request_model/schemas.py +++ b/request_model/schemas.py @@ -47,8 +47,8 @@ class AddDataParams(Params): organisation: Optional[str] = None plugin: Optional[PluginTypeEnum] = None github_branch: Optional[str] = None - selected_entities: Optional[List[Dict[str, str]]] = None - selected_redirects: Optional[Union[List[Dict[str, str]], Dict[str, str]]] = None + excluded_references: Optional[List[str]] = None + selected_redirects: Optional[List[Dict[str, str]]] = None class RequestBase(BaseModel): From 6f8b97cf568c48620b63aea4afbe5413518e8525 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Wed, 22 Jul 2026 08:54:46 +0100 Subject: [PATCH 4/6] Refactor formatting of excluded_references assignment in add_data_task --- request-processor/src/tasks.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/request-processor/src/tasks.py b/request-processor/src/tasks.py index 89c1468..2b276aa 100644 --- a/request-processor/src/tasks.py +++ b/request-processor/src/tasks.py @@ -432,9 +432,7 @@ def add_data_task(request: Dict, directories=None): github_branch=request_data.github_branch, endpoint_parameters=request_data.endpoint_parameters, endpoints=endpoints, - excluded_references=getattr( - request_data, "excluded_references", None - ), + excluded_references=getattr(request_data, "excluded_references", None), selected_redirects=getattr(request_data, "selected_redirects", None), ) if "plugin" in log: From 3821c79d029dd9ac8ce6e97ab5b35a46f0372472 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Wed, 22 Jul 2026 09:02:21 +0100 Subject: [PATCH 5/6] Remove redundant redirect reference functions and streamline old entity retrieval in _create_old_entity_redirects --- .../src/application/core/pipeline.py | 18 ++---------------- 1 file changed, 2 insertions(+), 16 deletions(-) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index 2c4ea30..4c7afbc 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -30,20 +30,6 @@ def _reference_set(references): } -def _redirect_reference(redirect): - """Return the reference from a selected redirect object.""" - if isinstance(redirect, dict): - return str(redirect.get("reference", "")).strip() - return str(redirect or "").strip() - - -def _redirect_old_entity(redirect): - """Return the old entity number from a selected redirect object.""" - if not isinstance(redirect, dict): - return "" - return str(redirect.get("old_entity_number", "")).strip() - - def _filter_selected_entities(new_entities, excluded_references): """ Return entities not listed in excluded_references. @@ -101,8 +87,8 @@ def _create_old_entity_redirects( } for redirect in selected_redirects: - reference = _redirect_reference(redirect) - old_entity = _redirect_old_entity(redirect) + reference = str(redirect.get("reference", "")).strip() + old_entity = str(redirect.get("old_entity_number", "")).strip() entity = entity_by_reference.get(reference, "") if not old_entity or not entity: continue From 7432667b7f309e830373d8180bbe0b56c75e8178 Mon Sep 17 00:00:00 2001 From: Gibah Joseph Date: Wed, 22 Jul 2026 09:05:30 +0100 Subject: [PATCH 6/6] Remove redundant selection of entity IDs in _create_old_entity_redirects --- request-processor/src/application/core/pipeline.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/request-processor/src/application/core/pipeline.py b/request-processor/src/application/core/pipeline.py index 4c7afbc..5013e7e 100644 --- a/request-processor/src/application/core/pipeline.py +++ b/request-processor/src/application/core/pipeline.py @@ -66,9 +66,6 @@ def _create_old_entity_redirects( return [] selected_new_entities = _filter_selected_entities(new_entities, excluded_references) - selected_entity_ids = { - str(entity.get("entity", "")).strip() for entity in selected_new_entities - } old_entity_rows = [] seen = set()