From 107f8228f459707611c11f3feb5b8a3f198460de Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Fri, 25 Apr 2025 12:46:21 -0300 Subject: [PATCH 1/9] Adding contribute app to voyages-api. --- src/api/contrib/__init__.py | 0 src/api/contrib/samples.http | 103 ++++++++++++++ src/api/contrib/urls.py | 6 + src/api/contrib/views.py | 127 ++++++++++++++++++ src/api/voyages3/settings.py | 2 +- src/api/voyages3/urls.py | 1 + ...py-default.py => localsettings.py-default} | 0 7 files changed, 238 insertions(+), 1 deletion(-) create mode 100644 src/api/contrib/__init__.py create mode 100644 src/api/contrib/samples.http create mode 100644 src/api/contrib/urls.py create mode 100644 src/api/contrib/views.py rename src/geo-networks/{localsettings.py-default.py => localsettings.py-default} (100%) diff --git a/src/api/contrib/__init__.py b/src/api/contrib/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/api/contrib/samples.http b/src/api/contrib/samples.http new file mode 100644 index 00000000..76cbc932 --- /dev/null +++ b/src/api/contrib/samples.http @@ -0,0 +1,103 @@ +@CurrentHost = http://localhost:8000 + +# @name location-fetch +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "single": { + "query": { + "model": "geo.Location", + "filter": [] + }, + "fields": [ + "name", + "value" + ] + } +} + +### + +# @name nationality +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "single": { + "query": { + "model": "voyage_nationality", + "filter": [] + }, + "fields": [ + "name" + ] + } +} + +### + +# @name single-voyage +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "q1": { + "query": { + "model": "voyage_voyage", + "filter": [ { "field": "voyage_id", "value": "1" } ] + }, + "fields": [ "voyage_id", "dataset" ] + } +} + +### + +# @name multiple-queries +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "q47": { + "query": { + "model": "past_enslaveralias", + "filter": [ + { + "field": "id", + "value": 1035662 + } + ] + }, + "fields": [ + "alias", + "identity_id", + "id" + ] + }, + "q48": { + "query": { + "model": "past_enslaverinrelation_roles", + "filter": [ + { + "field": "enslaverinrelation_id", + "value": 1080 + } + ] + }, + "fields": [ + "enslaverinrelation_id", + "enslaverrole_id", + "id" + ] + } +} + +### \ No newline at end of file diff --git a/src/api/contrib/urls.py b/src/api/contrib/urls.py new file mode 100644 index 00000000..6cd77d68 --- /dev/null +++ b/src/api/contrib/urls.py @@ -0,0 +1,6 @@ +from django.urls import path +from .views import batch_data_api + +urlpatterns = [ + path('data', batch_data_api, name='contrib_data_api'), +] \ No newline at end of file diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py new file mode 100644 index 00000000..6f10f20e --- /dev/null +++ b/src/api/contrib/views.py @@ -0,0 +1,127 @@ +""" +Django implementation of the BatchDataResolver interface. +This module provides a way to execute multiple database queries in a single request. +""" +from typing import Dict, List, Any, Optional, Union, Tuple +import json +from django.http import JsonResponse +from django.views.decorators.http import require_POST +from django.views.decorators.csrf import csrf_exempt +from django.db.models import Q, Model +from django.apps import apps + +# Type definitions to match the TypeScript interface +NonNullFieldValue = Union[str, int, float, bool] +BaseFieldValue = Optional[NonNullFieldValue] +ResolvedEntityData = Dict[str, BaseFieldValue] + +class DataFilter: + """Represents a filter condition on a field.""" + + def __init__(self, field: str, value: Union[NonNullFieldValue, List[NonNullFieldValue]], operator: str = "equals"): + self.field = field + self.value = value + self.operator = operator + + def to_q(self) -> Q: + """Convert the filter to a Django Q object.""" + if self.operator == "in": + return Q(**{f"{self.field}__in": self.value}) + # Default to equals + return Q(**{self.field: self.value}) + + +class DataQuery: + """Represents a query on a specific model with filters.""" + + def __init__(self, model: str, filters: List[DataFilter]): + self.model = model.replace("_", ".", 1) # Convert to Django model path + self.filters = filters + + def get_model_class(self) -> Model: + """Get the Django model class from the model name.""" + try: + return apps.get_model(self.model) + except LookupError: + raise ValueError(f"Model {self.model} not found") + + def execute(self, fields: List[str]) -> List[ResolvedEntityData]: + """Execute the query and return the results.""" + model_class = self.get_model_class() + + # Build the filter query + query = Q() + for filter_obj in self.filters: + query &= filter_obj.to_q() + + # Execute the query and return the results + queryset = model_class.objects.filter(query) + # Convert to list of dictionaries with only the requested fields + result = [] + for instance in queryset: + entity_data = {} + for field in fields: + try: + value = getattr(instance, field) + # Ensure the value is of a compatible type + if value is not None and not isinstance(value, (str, int, float, bool)): + value = str(value) + entity_data[field] = value + except AttributeError: + entity_data[field] = None + result.append(entity_data) + + return result + + +class DataResolver: + """Resolver for a single data query.""" + + @staticmethod + def fetch(input_data: Dict[str, Any]) -> List[ResolvedEntityData]: + """Fetch data based on the input query.""" + query = DataQuery( + model=input_data["query"]["model"], + filters=[ + DataFilter( + field=filter_obj["field"], + value=filter_obj["value"], + operator=filter_obj.get("operator", "equals") + ) + for filter_obj in input_data["query"]["filter"] + ] + ) + return query.execute(input_data["fields"]) + + +class BatchDataResolver: + """Resolver for multiple data queries in one batch.""" + + @staticmethod + def fetchBatch(batch: Dict[str, Dict[str, Any]]) -> Dict[str, List[ResolvedEntityData]]: + """Fetch multiple queries serially and return results keyed by query ID.""" + results = {} + for key, input_data in batch.items(): + results[key] = DataResolver.fetch(input_data) + return results + + +@csrf_exempt +@require_POST +def batch_data_api(request): + """API endpoint that handles batch data requests.""" + try: + # Parse the request body + batch_data = json.loads(request.body) + # Validate the input + if not isinstance(batch_data, dict): + return JsonResponse({"error": "Input must be a dictionary"}, status=400) + # Execute the batch query + results = BatchDataResolver.fetchBatch(batch_data) + # Return the results + return JsonResponse(results, safe=False) + + except json.JSONDecodeError: + return JsonResponse({"error": "Invalid JSON"}, status=400) + except Exception as e: + return JsonResponse({"error": str(e)}, status=500) diff --git a/src/api/voyages3/settings.py b/src/api/voyages3/settings.py index d4536754..fa92869b 100644 --- a/src/api/voyages3/settings.py +++ b/src/api/voyages3/settings.py @@ -45,7 +45,7 @@ 'blog', 'geo', 'document', -# 'contribute', + 'contrib', 'tinymce', 'rest_framework', 'rest_framework.authtoken', diff --git a/src/api/voyages3/urls.py b/src/api/voyages3/urls.py index d19fcfbc..98d687b8 100644 --- a/src/api/voyages3/urls.py +++ b/src/api/voyages3/urls.py @@ -38,6 +38,7 @@ path('captcha/', include('captcha.urls')), path('filebrowser/', site.urls), path('tinymce/', site.urls), + path('contrib/', include('contrib.urls')), re_path(r'^_nested_admin/', include('nested_admin.urls')), path('schema/', SpectacularAPIView.as_view(), name='schema'), path('', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui') diff --git a/src/geo-networks/localsettings.py-default.py b/src/geo-networks/localsettings.py-default similarity index 100% rename from src/geo-networks/localsettings.py-default.py rename to src/geo-networks/localsettings.py-default From 34c5a4cad4a72490097b450df4072c83e48f64ee Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Thu, 1 May 2025 09:18:55 -0300 Subject: [PATCH 2/9] Adding an un-filtered geo tree endpoint for contrib. --- src/api/contrib/samples.http | 5 +++++ src/api/contrib/urls.py | 3 ++- src/api/contrib/views.py | 6 ++++++ 3 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/api/contrib/samples.http b/src/api/contrib/samples.http index 76cbc932..355ba1c4 100644 --- a/src/api/contrib/samples.http +++ b/src/api/contrib/samples.http @@ -100,4 +100,9 @@ Accept-Language: en-US,en;q=0.5 } } +### + +# @name location_tree +GET {{CurrentHost}}/contrib/location_tree + ### \ No newline at end of file diff --git a/src/api/contrib/urls.py b/src/api/contrib/urls.py index 6cd77d68..c10e158d 100644 --- a/src/api/contrib/urls.py +++ b/src/api/contrib/urls.py @@ -1,6 +1,7 @@ from django.urls import path -from .views import batch_data_api +from .views import batch_data_api, location_tree urlpatterns = [ path('data', batch_data_api, name='contrib_data_api'), + path('location_tree', location_tree, name='location_tree'), ] \ No newline at end of file diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py index 6f10f20e..9401a436 100644 --- a/src/api/contrib/views.py +++ b/src/api/contrib/views.py @@ -10,6 +10,8 @@ from django.db.models import Q, Model from django.apps import apps +from geo.common import GeoTreeFilter + # Type definitions to match the TypeScript interface NonNullFieldValue = Union[str, int, float, bool] BaseFieldValue = Optional[NonNullFieldValue] @@ -125,3 +127,7 @@ def batch_data_api(request): return JsonResponse({"error": "Invalid JSON"}, status=400) except Exception as e: return JsonResponse({"error": str(e)}, status=500) + +@csrf_exempt +def location_tree(_): + return JsonResponse(GeoTreeFilter(select_all=True), safe=False) \ No newline at end of file From 9f45e22d7762126c4da11742f71054ddaf24b0f5 Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Tue, 20 May 2025 17:32:59 -0300 Subject: [PATCH 3/9] Much faster query times (for Source) by _not_ manipulating Django ORM objects and instead using raw dictionaries. --- src/api/contrib/samples.http | 23 +++++++++++++++++++++ src/api/contrib/views.py | 40 +++++++++++++++++++++--------------- 2 files changed, 46 insertions(+), 17 deletions(-) diff --git a/src/api/contrib/samples.http b/src/api/contrib/samples.http index 355ba1c4..0b34c429 100644 --- a/src/api/contrib/samples.http +++ b/src/api/contrib/samples.http @@ -102,6 +102,29 @@ Accept-Language: en-US,en;q=0.5 ### +# @name VoyageSources +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "q1": { + "query": { + "model": "document_source", + "filter": [] + }, + "fields": [ + "title", + "short_ref", + "bib", + "notes" + ] + } +} + +### + # @name location_tree GET {{CurrentHost}}/contrib/location_tree diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py index 9401a436..2ec435a8 100644 --- a/src/api/contrib/views.py +++ b/src/api/contrib/views.py @@ -32,6 +32,14 @@ def to_q(self) -> Q: # Default to equals return Q(**{self.field: self.value}) +def remap(r: dict[str, any], remapped: list[str]): + for f in remapped: + key = f.replace("_id", "") + if key in r: + r[f] = r[key] + del r[key] + return r + class DataQuery: """Represents a query on a specific model with filters.""" @@ -57,25 +65,23 @@ def execute(self, fields: List[str]) -> List[ResolvedEntityData]: query &= filter_obj.to_q() # Execute the query and return the results - queryset = model_class.objects.filter(query) - # Convert to list of dictionaries with only the requested fields - result = [] - for instance in queryset: - entity_data = {} - for field in fields: - try: - value = getattr(instance, field) - # Ensure the value is of a compatible type - if value is not None and not isinstance(value, (str, int, float, bool)): - value = str(value) - entity_data[field] = value - except AttributeError: - entity_data[field] = None - result.append(entity_data) - + # Get all valid field names for the model + valid_field_names = {f.name for f in model_class._meta.fields} + # Filter out invalid field names. + # First detect fields with _id suffix that should be remapped. + remapped = [f for f in fields if "_id" in f and f not in valid_field_names] + fields = [f.replace("_id", "") if f in remapped else f for f in fields] + valid_fields = [f for f in fields if f in valid_field_names] + if len(valid_fields) < len(fields): + print("Invalid fields found in query: ") + print([f for f in fields if f not in valid_fields]) + print("Expected:") + print(valid_field_names) + result = list(model_class.objects.filter(query).values(*valid_fields)) + if len(remapped) > 0: + result = [remap(r, remapped) for r in result] return result - class DataResolver: """Resolver for a single data query.""" From 48a41dbe6ea7e2631f7e072d99742da75f6e8ea9 Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Mon, 1 Sep 2025 13:59:31 -0300 Subject: [PATCH 4/9] Implementing a publication endpoint for folded contribution updates. --- src/api/contrib/urls.py | 7 +- src/api/contrib/views.py | 528 ++++++++++++++++++++++++++++++++++- src/api/voyages3/settings.py | 3 + 3 files changed, 525 insertions(+), 13 deletions(-) diff --git a/src/api/contrib/urls.py b/src/api/contrib/urls.py index c10e158d..cadd8c77 100644 --- a/src/api/contrib/urls.py +++ b/src/api/contrib/urls.py @@ -1,7 +1,10 @@ from django.urls import path -from .views import batch_data_api, location_tree +from .views import batch_data_api, location_tree, publish_batch, publication_status urlpatterns = [ path('data', batch_data_api, name='contrib_data_api'), path('location_tree', location_tree, name='location_tree'), -] \ No newline at end of file + path('publish_batch', publish_batch, name='publish_batch'), + # Updated to capture publication_key from the URL: /publication_status/ + path('publication_status/', publication_status, name='publication_status'), +] diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py index 2ec435a8..52b7afac 100644 --- a/src/api/contrib/views.py +++ b/src/api/contrib/views.py @@ -1,21 +1,32 @@ """ -Django implementation of the BatchDataResolver interface. -This module provides a way to execute multiple database queries in a single request. +This module provides APIs that are used by the Contribute Application (backend +and frontend). """ -from typing import Dict, List, Any, Optional, Union, Tuple -import json +from dataclasses import dataclass, field +from datetime import datetime, timedelta +import threading +import traceback +from typing import Dict, List, Any, Optional, Type, Union, Set +from collections import defaultdict, deque +from django.apps import apps +from django.db import connection, transaction +from django.db.models import Q, Model from django.http import JsonResponse +from django.views import View from django.views.decorators.http import require_POST from django.views.decorators.csrf import csrf_exempt -from django.db.models import Q, Model -from django.apps import apps - +from django.db import models +from voyage.models import Voyage from geo.common import GeoTreeFilter +import json +import logging # Type definitions to match the TypeScript interface NonNullFieldValue = Union[str, int, float, bool] BaseFieldValue = Optional[NonNullFieldValue] ResolvedEntityData = Dict[str, BaseFieldValue] + +logger = logging.getLogger(__name__) class DataFilter: """Represents a filter condition on a field.""" @@ -32,7 +43,7 @@ def to_q(self) -> Q: # Default to equals return Q(**{self.field: self.value}) -def remap(r: dict[str, any], remapped: list[str]): +def remap(r: Dict[str, Any], remapped: List[str]): for f in remapped: key = f.replace("_id", "") if key in r: @@ -40,7 +51,6 @@ def remap(r: dict[str, any], remapped: list[str]): del r[key] return r - class DataQuery: """Represents a query on a specific model with filters.""" @@ -113,7 +123,6 @@ def fetchBatch(batch: Dict[str, Dict[str, Any]]) -> Dict[str, List[ResolvedEntit results[key] = DataResolver.fetch(input_data) return results - @csrf_exempt @require_POST def batch_data_api(request): @@ -136,4 +145,501 @@ def batch_data_api(request): @csrf_exempt def location_tree(_): - return JsonResponse(GeoTreeFilter(select_all=True), safe=False) \ No newline at end of file + return JsonResponse(GeoTreeFilter(select_all=True), safe=False) + +_schemaToDbTable: Dict[str, str] = { + 'VoyageSparseDate': 'voyage_voyagesparsedate', + 'Nationality': 'voyage_nationality', + 'TonType': 'voyage_tontype', + 'RigOfVessel': 'voyage_rigofvessel', + 'Location': 'geo_location', + 'VoyageShip': 'voyage_voyageship', + 'VoyageItinerary': 'voyage_voyageitinerary', + 'VoyageSlaveNumbers': 'voyage_voyageslavesnumbers', + 'VoyageDates': 'voyage_voyagedates', + 'VoyageGrouping': 'voyage_voyagegroupings', + 'AfricanInfo': 'voyage_africaninfo', + 'CargoUnit': 'voyage_cargounit', + 'CargoType': 'voyage_cargotype', + 'VoyageCargoConnectionSchema': 'voyage_voyagecargoconnection', + 'ParticularOutcome': 'voyage_particularoutcome', + 'SlavesOutcome': 'voyage_slavesoutcome', + 'VesselOutcomeSchema': 'voyage_vesselcapturedoutcome', + 'OwnerOutcome': 'voyage_owneroutcome', + 'Resistance': 'voyage_resistance', + 'VoyageOutcome': 'voyage_voyageoutcome', + 'VoyageCrew': 'voyage_voyagecrew', + 'EnslaverAlias': 'past_enslaveralias', + 'Enslaver': 'past_enslaveridentity', + 'EnslaverAliasWithFK': 'past_enslaveralias', + 'EnslaverAliasWithIdentity': 'past_enslaveralias', + 'EnslavementRelationType': 'past_enslavementrelationtype', + 'EnslaverRole': 'past_enslaverrole', + 'EnslaverRelationRoleConn': 'past_enslaverinrelation_roles', + 'EnslaverInRelation': 'past_enslaverinrelation', + 'Enslaved': 'past_enslaved', + 'EnslavedInRelation': 'past_enslavedinrelation', + 'EnslavementRelation': 'past_enslavementrelation', + 'Voyage Source Type': 'document_sourcetype', + 'Voyage Source Short Reference': 'document_shortref', + 'Voyage Source': 'document_source', + 'Voyage Source Connection': 'document_sourcevoyageconnection', + 'Voyage': 'voyage_voyage' +} + +_schemaToModel: Dict[str, Type[models.Model]] = {} +for m in apps.get_models(include_auto_created=True): + for k, v in _schemaToDbTable.items(): + if m._meta.db_table == v: + _schemaToModel[k] = m +_missing = set() +for k in _schemaToDbTable.keys(): + if k not in _schemaToModel: + _missing.add(k) +if _missing: + logger.error(f"Missing model for schema: {list(_missing)}") +else: + logger.info("All models mapped successfully") + +class ChangeSetProcessor: + """Processes CombinedChangeSet and applies changes to Django ORM.""" + + def __init__(self): + self.temp_id_to_db_id: Dict[str, Any] = {} + self.new_entity_ids: Set[str] = set() + + def process_changeset(self, changeset: Dict) -> Dict[str, Any]: + """ + Process the entire changeset in a transaction. + Returns a mapping of temporary IDs to actual database IDs. + """ + deletions = changeset.get('deletions', []) + updates = changeset.get('updates', []) + + # Validate no duplicate entity updates and no updates to deleted entities + self._validate_no_duplicates(deletions, updates) + + # Collect all new entity IDs first + self.new_entity_ids = { + update['entityRef']['id'] + for update in updates + if update['entityRef'].get('type') == 'new' + } + + # Separate new and existing entity updates + new_updates = [u for u in updates if u['entityRef'].get('type') == 'new'] + existing_updates = [u for u in updates if u['entityRef'].get('type') != 'new'] + + # Sort new entities by FK dependencies + sorted_new_updates = self._topological_sort(new_updates) + + with transaction.atomic(): + # Process deletions first + for deletion in deletions: + self._process_deletion(deletion) + + # Process new entities in dependency order + for update in sorted_new_updates: + logger.debug(json.dumps(update)) + self._process_new_entity(update) + + # Process existing entity updates + for update in existing_updates: + self._process_existing_entity(update) + + return self.temp_id_to_db_id + + def _validate_no_duplicates(self, deletions: List[Dict], updates: List[Dict]): + """Validate that each entity is updated at most once and deleted entities aren't updated.""" + updated_entities = set() + deleted_entities = set() + + # Collect deleted entity references + for deletion in deletions: + entity_key = self._get_entity_key(deletion['entityRef']) + deleted_entities.add(entity_key) + + # Check for duplicate updates and updates to deleted entities + for update in updates: + entity_key = self._get_entity_key(update['entityRef']) + + if entity_key in deleted_entities: + raise ValueError(f"Cannot update deleted entity: {entity_key}") + + if entity_key in updated_entities: + raise ValueError(f"Duplicate update for entity: {entity_key}") + + updated_entities.add(entity_key) + + def _get_entity_key(self, entity_ref: Dict) -> str: + """Create a unique key for an entity reference.""" + return f"{entity_ref['schema']}:{entity_ref['id']}" + + def _topological_sort(self, new_updates: List[Dict]) -> List[Dict]: + """ + Sort new entities based on foreign key dependencies. + Returns sorted list or raises ValueError if cycle detected. + """ + # Build dependency graph + graph = defaultdict(list) # entity_id -> list of entities that depend on it + in_degree = defaultdict(int) # entity_id -> number of dependencies + entity_map = {} # entity_id -> update dict + + for update in new_updates: + entity_id = update['entityRef']['id'] + entity_map[entity_id] = update + + # Find FK dependencies to other new entities + for change in update['changes']: + if change.get('isForeignKey') and change.get('changed'): + fk_value = change['changed'] + # Check if FK points to a new entity (using set membership) + if fk_value in self.new_entity_ids: + graph[fk_value].append(entity_id) + in_degree[entity_id] += 1 + + # Initialize queue with entities that have no dependencies + queue = deque() + for update in new_updates: + entity_id = update['entityRef']['id'] + if in_degree[entity_id] == 0: + queue.append(entity_id) + + # Process queue (Kahn's algorithm) + sorted_ids = [] + while queue: + current_id = queue.popleft() + sorted_ids.append(current_id) + + # Reduce in-degree for dependent entities + for dependent_id in graph[current_id]: + in_degree[dependent_id] -= 1 + if in_degree[dependent_id] == 0: + queue.append(dependent_id) + + # Check for cycles + if len(sorted_ids) != len(new_updates): + raise ValueError("Circular dependency detected in new entities") + + # Return updates in sorted order + return [entity_map[entity_id] for entity_id in sorted_ids] + + def _process_deletion(self, deletion: Dict): + """Process a deletion operation.""" + entity_ref = deletion['entityRef'] + model_class = self._get_model_class(entity_ref['schema']) + + try: + obj = model_class.objects.get(pk=entity_ref['id']) + obj.delete() + logger.info(f"Deleted {entity_ref['schema']} with id {entity_ref['id']}") + except model_class.DoesNotExist: + logger.warning(f"Entity {entity_ref['schema']} with id {entity_ref['id']} not found for deletion") + + def _process_new_entity(self, update: Dict): + """Process creation of a new entity.""" + entity_ref = update['entityRef'] + model_class = self._get_model_class(entity_ref['schema']) + + # Prepare field values + field_values = {} + for change in update['changes']: + field_name = change['property'] + value = change['changed'] + + # Replace temporary FK references with actual IDs + if change.get('isForeignKey') and value and value in self.new_entity_ids: + if value not in self.temp_id_to_db_id: + raise ValueError(f"Foreign key references unprocessed entity: {value}") + value = self.temp_id_to_db_id[value] + + field_values[field_name] = value + + # Create the entity - let Django handle field validation + obj = model_class(**field_values) + # Special cases: + if (model_class == Voyage): + # (domingos): I don't know why we have both id and voyage_id in the + # model, but at this point it is better to make sure they're equal. + obj.id = obj.voyage_id + obj.save() + + # Store the mapping from temporary ID to actual database ID + self.temp_id_to_db_id[entity_ref['id']] = obj.pk + + logger.info(f"Created {entity_ref['schema']} with temp id {entity_ref['id']} -> db id {obj.pk}") + + def _process_existing_entity(self, update: Dict): + """Process update of an existing entity.""" + entity_ref = update['entityRef'] + model_class = self._get_model_class(entity_ref['schema']) + + try: + obj = model_class.objects.get(pk=entity_ref['id']) + + # Apply changes + for change in update['changes']: + field_name = change['property'] + value = change['changed'] + + # Replace temporary FK references with actual IDs + if change.get('isForeignKey') and value and value in self.new_entity_ids: + if value not in self.temp_id_to_db_id: + raise ValueError(f"Foreign key references unprocessed entity: {value}") + value = self.temp_id_to_db_id[value] + + setattr(obj, field_name, value) + + obj.save() + logger.info(f"Updated {entity_ref['schema']} with id {entity_ref['id']}") + + except model_class.DoesNotExist: + raise ValueError(f"Entity {entity_ref['schema']} with id {entity_ref['id']} not found for update") + + def _get_model_class(self, schema_name: str): + """Get Django model class from schema name.""" + # Try to get model from any installed app + if schema_name not in _schemaToModel: + raise ValueError(f"Model {schema_name} not found in any app") + return _schemaToModel[schema_name] + + +class PublicationStatus: + PENDING = "pending" + PROCESSING = "processing" + COMPLETED = "completed" + FAILED = "failed" + + +@dataclass +class PublicationTask: + """Represents a changeset publication task.""" + publication_key: str + status: PublicationStatus + created_at: datetime + started_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + changeset: Dict = field(default_factory=dict) + result: Optional[Dict] = None + error: Optional[str] = None + id_mappings: Dict[str, Any] = field(default_factory=dict) + + +# TODO: move the TaskStore to the database so that it can be shared across +# workers. + +class TaskStore: + """In-memory storage for publication tasks.""" + + def __init__(self, max_age_hours: int = 24): + self._tasks: Dict[str, PublicationTask] = {} + self._lock = threading.Lock() + self.max_age_hours = max_age_hours + + def get(self, publication_key: str) -> Optional[PublicationTask]: + """Get a task by publication key.""" + with self._lock: + self._cleanup_old_tasks() + return self._tasks.get(publication_key) + + def create(self, publication_key: str, changeset: Dict) -> PublicationTask: + """Create or return existing task.""" + with self._lock: + # Check if task already exists (idempotency) + if publication_key in self._tasks: + return self._tasks[publication_key] + + # Create new task + task = PublicationTask( + publication_key=publication_key, + status=PublicationStatus.PENDING, + created_at=datetime.now(), + changeset=changeset + ) + self._tasks[publication_key] = task + return task + + def update_status(self, publication_key: str, status: PublicationStatus, + result: Optional[Dict] = None, error: Optional[str] = None): + """Update task status.""" + with self._lock: + task = self._tasks.get(publication_key) + if task: + task.status = status + if status == PublicationStatus.PROCESSING: + task.started_at = datetime.now() + elif status in (PublicationStatus.COMPLETED, PublicationStatus.FAILED): + task.completed_at = datetime.now() + if result: + task.result = result + if error: + task.error = error + + def _cleanup_old_tasks(self): + """Remove tasks older than max_age_hours.""" + cutoff = datetime.now() - timedelta(hours=self.max_age_hours) + expired_keys = [ + key for key, task in self._tasks.items() + if task.created_at < cutoff + ] + for key in expired_keys: + del self._tasks[key] + logger.info(f"Cleaned up expired task: {key}") + + +# Global task store (singleton for the application instance) +task_store = TaskStore() + +def process_changeset_background(publication_key: str): + """Background task to process a changeset.""" + task = task_store.get(publication_key) + if not task: + logger.error(f"Task not found for publication key: {publication_key}") + return + + try: + # Update status to processing + task_store.update_status(publication_key, PublicationStatus.PROCESSING) + logger.info(f"Starting processing for publication: {publication_key}") + + # Process the changeset + processor = ChangeSetProcessor() + + # Close any existing DB connections to ensure thread safety + connection.close() + + id_mappings = processor.process_changeset(task.changeset) + + # Update task with success + task_store.update_status( + publication_key, + PublicationStatus.COMPLETED, + result={ + 'success': True, + 'id_mappings': id_mappings, + 'message': 'Changeset processed successfully' + } + ) + logger.info(f"Successfully processed publication: {publication_key}") + + except ValueError as e: + logger.error(f"Validation error for publication {publication_key}: {e}") + task_store.update_status( + publication_key, + PublicationStatus.FAILED, + error=traceback.format_exc() + ) + except Exception as e: + logger.error(f"Unexpected error processing publication {publication_key}: {e}", exc_info=True) + task_store.update_status( + publication_key, + PublicationStatus.FAILED, + error=f"Internal error: {traceback.format_exc()}" + ) + finally: + # Ensure DB connection is closed + connection.close() + +@csrf_exempt +@require_POST +def publish_batch(request): + """ + Initiate processing of a CombinedChangeSet. + Expects JSON body with 'publication_key' and 'changeset'. + """ + try: + # Parse JSON body + if not request.body: + return JsonResponse({'error': 'Empty request body'}, status=400) + + data = json.loads(request.body) + + # Extract publication key and changeset + publication_key = data.get('key') + if not publication_key: + return JsonResponse({'error': 'Missing idempotency key'}, status=400) + + changeset = data.get('changeset') + if not changeset: + return JsonResponse({'error': 'Missing changeset'}, status=400) + + # Create or get existing task (idempotency) + task = task_store.create(publication_key, changeset) + + # If task is already completed or failed, return immediately + if task.status == PublicationStatus.COMPLETED: + return JsonResponse(task.result) + elif task.status == PublicationStatus.FAILED: + return JsonResponse({'error': task.error}, status=400) + elif task.status == PublicationStatus.PROCESSING: + return JsonResponse({ + 'status': 'processing', + 'publication_key': publication_key, + 'message': 'Publication is already being processed' + }) + + # Start background processing for new task + thread = threading.Thread( + target=process_changeset_background, + args=(publication_key,), + daemon=True + ) + thread.start() + + return JsonResponse({ + 'status': 'accepted', + 'publication_key': publication_key, + 'message': 'Publication queued for processing' + }, status=202) + + except json.JSONDecodeError as e: + logger.error(f"JSON parse error: {e}") + return JsonResponse({'error': 'Invalid JSON format'}, status=400) + except Exception as e: + logger.error(f"Unexpected error in publish_batch: {e}", exc_info=True) + return JsonResponse({'error': str(e)}, status=500) + +def publication_status(request, publication_key): + """ + Check the status of a publication task. + Returns current status and result if completed. + """ + try: + task = task_store.get(publication_key) + + if not task: + return JsonResponse({ + 'error': 'Publication not found', + 'publication_key': publication_key + }, status=404) + + # Build response based on task status + response = { + 'publication_key': publication_key, + 'status': task.status, + 'created_at': task.created_at.isoformat(), + } + + if task.started_at: + response['started_at'] = task.started_at.isoformat() + + if task.completed_at: + response['completed_at'] = task.completed_at.isoformat() + duration = (task.completed_at - task.started_at).total_seconds() + response['duration_seconds'] = duration + + if task.status == PublicationStatus.COMPLETED: + response.update(task.result) + elif task.status == PublicationStatus.FAILED: + response['error'] = task.error + return JsonResponse(response, status=400) + elif task.status == PublicationStatus.PROCESSING: + response['message'] = 'Publication is being processed' + elif task.status == PublicationStatus.PENDING: + response['message'] = 'Publication is queued for processing' + + return JsonResponse(response) + + except Exception as e: + logger.error(f"Error checking publication status: {e}", exc_info=True) + return JsonResponse({'error': str(e)}, status=500) diff --git a/src/api/voyages3/settings.py b/src/api/voyages3/settings.py index fa92869b..8a431733 100644 --- a/src/api/voyages3/settings.py +++ b/src/api/voyages3/settings.py @@ -202,3 +202,6 @@ # https://docs.djangoproject.com/en/3.2/ref/settings/#default-auto-field DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField' + +# Setting a large upload size for publication to work. +DATA_UPLOAD_MAX_MEMORY_SIZE = 50 * 1024 * 1024 # 50MB From 0519216260ab5a327701f0a04877552fcfbd4d50 Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Fri, 25 Apr 2025 12:46:21 -0300 Subject: [PATCH 5/9] Adding contribute app to voyages-api. --- src/api/contrib/__init__.py | 0 src/api/contrib/samples.http | 103 ++++++++++++++ src/api/contrib/urls.py | 6 + src/api/contrib/views.py | 127 ++++++++++++++++++ src/api/voyages3/settings.py | 1 + src/api/voyages3/urls.py | 1 + ...py-default.py => localsettings.py-default} | 0 7 files changed, 238 insertions(+) create mode 100644 src/api/contrib/__init__.py create mode 100644 src/api/contrib/samples.http create mode 100644 src/api/contrib/urls.py create mode 100644 src/api/contrib/views.py rename src/geo-networks/{localsettings.py-default.py => localsettings.py-default} (100%) diff --git a/src/api/contrib/__init__.py b/src/api/contrib/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/api/contrib/samples.http b/src/api/contrib/samples.http new file mode 100644 index 00000000..76cbc932 --- /dev/null +++ b/src/api/contrib/samples.http @@ -0,0 +1,103 @@ +@CurrentHost = http://localhost:8000 + +# @name location-fetch +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "single": { + "query": { + "model": "geo.Location", + "filter": [] + }, + "fields": [ + "name", + "value" + ] + } +} + +### + +# @name nationality +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "single": { + "query": { + "model": "voyage_nationality", + "filter": [] + }, + "fields": [ + "name" + ] + } +} + +### + +# @name single-voyage +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "q1": { + "query": { + "model": "voyage_voyage", + "filter": [ { "field": "voyage_id", "value": "1" } ] + }, + "fields": [ "voyage_id", "dataset" ] + } +} + +### + +# @name multiple-queries +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "q47": { + "query": { + "model": "past_enslaveralias", + "filter": [ + { + "field": "id", + "value": 1035662 + } + ] + }, + "fields": [ + "alias", + "identity_id", + "id" + ] + }, + "q48": { + "query": { + "model": "past_enslaverinrelation_roles", + "filter": [ + { + "field": "enslaverinrelation_id", + "value": 1080 + } + ] + }, + "fields": [ + "enslaverinrelation_id", + "enslaverrole_id", + "id" + ] + } +} + +### \ No newline at end of file diff --git a/src/api/contrib/urls.py b/src/api/contrib/urls.py new file mode 100644 index 00000000..6cd77d68 --- /dev/null +++ b/src/api/contrib/urls.py @@ -0,0 +1,6 @@ +from django.urls import path +from .views import batch_data_api + +urlpatterns = [ + path('data', batch_data_api, name='contrib_data_api'), +] \ No newline at end of file diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py new file mode 100644 index 00000000..6f10f20e --- /dev/null +++ b/src/api/contrib/views.py @@ -0,0 +1,127 @@ +""" +Django implementation of the BatchDataResolver interface. +This module provides a way to execute multiple database queries in a single request. +""" +from typing import Dict, List, Any, Optional, Union, Tuple +import json +from django.http import JsonResponse +from django.views.decorators.http import require_POST +from django.views.decorators.csrf import csrf_exempt +from django.db.models import Q, Model +from django.apps import apps + +# Type definitions to match the TypeScript interface +NonNullFieldValue = Union[str, int, float, bool] +BaseFieldValue = Optional[NonNullFieldValue] +ResolvedEntityData = Dict[str, BaseFieldValue] + +class DataFilter: + """Represents a filter condition on a field.""" + + def __init__(self, field: str, value: Union[NonNullFieldValue, List[NonNullFieldValue]], operator: str = "equals"): + self.field = field + self.value = value + self.operator = operator + + def to_q(self) -> Q: + """Convert the filter to a Django Q object.""" + if self.operator == "in": + return Q(**{f"{self.field}__in": self.value}) + # Default to equals + return Q(**{self.field: self.value}) + + +class DataQuery: + """Represents a query on a specific model with filters.""" + + def __init__(self, model: str, filters: List[DataFilter]): + self.model = model.replace("_", ".", 1) # Convert to Django model path + self.filters = filters + + def get_model_class(self) -> Model: + """Get the Django model class from the model name.""" + try: + return apps.get_model(self.model) + except LookupError: + raise ValueError(f"Model {self.model} not found") + + def execute(self, fields: List[str]) -> List[ResolvedEntityData]: + """Execute the query and return the results.""" + model_class = self.get_model_class() + + # Build the filter query + query = Q() + for filter_obj in self.filters: + query &= filter_obj.to_q() + + # Execute the query and return the results + queryset = model_class.objects.filter(query) + # Convert to list of dictionaries with only the requested fields + result = [] + for instance in queryset: + entity_data = {} + for field in fields: + try: + value = getattr(instance, field) + # Ensure the value is of a compatible type + if value is not None and not isinstance(value, (str, int, float, bool)): + value = str(value) + entity_data[field] = value + except AttributeError: + entity_data[field] = None + result.append(entity_data) + + return result + + +class DataResolver: + """Resolver for a single data query.""" + + @staticmethod + def fetch(input_data: Dict[str, Any]) -> List[ResolvedEntityData]: + """Fetch data based on the input query.""" + query = DataQuery( + model=input_data["query"]["model"], + filters=[ + DataFilter( + field=filter_obj["field"], + value=filter_obj["value"], + operator=filter_obj.get("operator", "equals") + ) + for filter_obj in input_data["query"]["filter"] + ] + ) + return query.execute(input_data["fields"]) + + +class BatchDataResolver: + """Resolver for multiple data queries in one batch.""" + + @staticmethod + def fetchBatch(batch: Dict[str, Dict[str, Any]]) -> Dict[str, List[ResolvedEntityData]]: + """Fetch multiple queries serially and return results keyed by query ID.""" + results = {} + for key, input_data in batch.items(): + results[key] = DataResolver.fetch(input_data) + return results + + +@csrf_exempt +@require_POST +def batch_data_api(request): + """API endpoint that handles batch data requests.""" + try: + # Parse the request body + batch_data = json.loads(request.body) + # Validate the input + if not isinstance(batch_data, dict): + return JsonResponse({"error": "Input must be a dictionary"}, status=400) + # Execute the batch query + results = BatchDataResolver.fetchBatch(batch_data) + # Return the results + return JsonResponse(results, safe=False) + + except json.JSONDecodeError: + return JsonResponse({"error": "Invalid JSON"}, status=400) + except Exception as e: + return JsonResponse({"error": str(e)}, status=500) diff --git a/src/api/voyages3/settings.py b/src/api/voyages3/settings.py index 19373bfd..5c1a28e7 100644 --- a/src/api/voyages3/settings.py +++ b/src/api/voyages3/settings.py @@ -45,6 +45,7 @@ 'blog', 'geo', 'document', + 'contrib', 'tinymce', 'rest_framework', 'rest_framework.authtoken', diff --git a/src/api/voyages3/urls.py b/src/api/voyages3/urls.py index d19fcfbc..98d687b8 100644 --- a/src/api/voyages3/urls.py +++ b/src/api/voyages3/urls.py @@ -38,6 +38,7 @@ path('captcha/', include('captcha.urls')), path('filebrowser/', site.urls), path('tinymce/', site.urls), + path('contrib/', include('contrib.urls')), re_path(r'^_nested_admin/', include('nested_admin.urls')), path('schema/', SpectacularAPIView.as_view(), name='schema'), path('', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui') diff --git a/src/geo-networks/localsettings.py-default.py b/src/geo-networks/localsettings.py-default similarity index 100% rename from src/geo-networks/localsettings.py-default.py rename to src/geo-networks/localsettings.py-default From e90d3666ded7b6119d13f0169c51c887027203ad Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Thu, 1 May 2025 09:18:55 -0300 Subject: [PATCH 6/9] Adding an un-filtered geo tree endpoint for contrib. --- src/api/contrib/samples.http | 5 +++++ src/api/contrib/urls.py | 3 ++- src/api/contrib/views.py | 6 ++++++ 3 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/api/contrib/samples.http b/src/api/contrib/samples.http index 76cbc932..355ba1c4 100644 --- a/src/api/contrib/samples.http +++ b/src/api/contrib/samples.http @@ -100,4 +100,9 @@ Accept-Language: en-US,en;q=0.5 } } +### + +# @name location_tree +GET {{CurrentHost}}/contrib/location_tree + ### \ No newline at end of file diff --git a/src/api/contrib/urls.py b/src/api/contrib/urls.py index 6cd77d68..c10e158d 100644 --- a/src/api/contrib/urls.py +++ b/src/api/contrib/urls.py @@ -1,6 +1,7 @@ from django.urls import path -from .views import batch_data_api +from .views import batch_data_api, location_tree urlpatterns = [ path('data', batch_data_api, name='contrib_data_api'), + path('location_tree', location_tree, name='location_tree'), ] \ No newline at end of file diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py index 6f10f20e..9401a436 100644 --- a/src/api/contrib/views.py +++ b/src/api/contrib/views.py @@ -10,6 +10,8 @@ from django.db.models import Q, Model from django.apps import apps +from geo.common import GeoTreeFilter + # Type definitions to match the TypeScript interface NonNullFieldValue = Union[str, int, float, bool] BaseFieldValue = Optional[NonNullFieldValue] @@ -125,3 +127,7 @@ def batch_data_api(request): return JsonResponse({"error": "Invalid JSON"}, status=400) except Exception as e: return JsonResponse({"error": str(e)}, status=500) + +@csrf_exempt +def location_tree(_): + return JsonResponse(GeoTreeFilter(select_all=True), safe=False) \ No newline at end of file From b27a1f22be145c8d66aa7d6caf099d61e81e51b1 Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Tue, 20 May 2025 17:32:59 -0300 Subject: [PATCH 7/9] Much faster query times (for Source) by _not_ manipulating Django ORM objects and instead using raw dictionaries. --- src/api/contrib/samples.http | 23 +++++++++++++++++++++ src/api/contrib/views.py | 40 +++++++++++++++++++++--------------- 2 files changed, 46 insertions(+), 17 deletions(-) diff --git a/src/api/contrib/samples.http b/src/api/contrib/samples.http index 355ba1c4..0b34c429 100644 --- a/src/api/contrib/samples.http +++ b/src/api/contrib/samples.http @@ -102,6 +102,29 @@ Accept-Language: en-US,en;q=0.5 ### +# @name VoyageSources +POST {{CurrentHost}}/contrib/data +Accept: application/json +Content-Type: application/json +Accept-Language: en-US,en;q=0.5 + +{ + "q1": { + "query": { + "model": "document_source", + "filter": [] + }, + "fields": [ + "title", + "short_ref", + "bib", + "notes" + ] + } +} + +### + # @name location_tree GET {{CurrentHost}}/contrib/location_tree diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py index 9401a436..2ec435a8 100644 --- a/src/api/contrib/views.py +++ b/src/api/contrib/views.py @@ -32,6 +32,14 @@ def to_q(self) -> Q: # Default to equals return Q(**{self.field: self.value}) +def remap(r: dict[str, any], remapped: list[str]): + for f in remapped: + key = f.replace("_id", "") + if key in r: + r[f] = r[key] + del r[key] + return r + class DataQuery: """Represents a query on a specific model with filters.""" @@ -57,25 +65,23 @@ def execute(self, fields: List[str]) -> List[ResolvedEntityData]: query &= filter_obj.to_q() # Execute the query and return the results - queryset = model_class.objects.filter(query) - # Convert to list of dictionaries with only the requested fields - result = [] - for instance in queryset: - entity_data = {} - for field in fields: - try: - value = getattr(instance, field) - # Ensure the value is of a compatible type - if value is not None and not isinstance(value, (str, int, float, bool)): - value = str(value) - entity_data[field] = value - except AttributeError: - entity_data[field] = None - result.append(entity_data) - + # Get all valid field names for the model + valid_field_names = {f.name for f in model_class._meta.fields} + # Filter out invalid field names. + # First detect fields with _id suffix that should be remapped. + remapped = [f for f in fields if "_id" in f and f not in valid_field_names] + fields = [f.replace("_id", "") if f in remapped else f for f in fields] + valid_fields = [f for f in fields if f in valid_field_names] + if len(valid_fields) < len(fields): + print("Invalid fields found in query: ") + print([f for f in fields if f not in valid_fields]) + print("Expected:") + print(valid_field_names) + result = list(model_class.objects.filter(query).values(*valid_fields)) + if len(remapped) > 0: + result = [remap(r, remapped) for r in result] return result - class DataResolver: """Resolver for a single data query.""" From 5d0ba299b36a5120432abb578bf2585a2323a138 Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Mon, 1 Sep 2025 13:59:31 -0300 Subject: [PATCH 8/9] Implementing a publication endpoint for folded contribution updates. --- src/api/contrib/urls.py | 7 +- src/api/contrib/views.py | 528 ++++++++++++++++++++++++++++++++++- src/api/voyages3/settings.py | 3 + 3 files changed, 525 insertions(+), 13 deletions(-) diff --git a/src/api/contrib/urls.py b/src/api/contrib/urls.py index c10e158d..cadd8c77 100644 --- a/src/api/contrib/urls.py +++ b/src/api/contrib/urls.py @@ -1,7 +1,10 @@ from django.urls import path -from .views import batch_data_api, location_tree +from .views import batch_data_api, location_tree, publish_batch, publication_status urlpatterns = [ path('data', batch_data_api, name='contrib_data_api'), path('location_tree', location_tree, name='location_tree'), -] \ No newline at end of file + path('publish_batch', publish_batch, name='publish_batch'), + # Updated to capture publication_key from the URL: /publication_status/ + path('publication_status/', publication_status, name='publication_status'), +] diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py index 2ec435a8..52b7afac 100644 --- a/src/api/contrib/views.py +++ b/src/api/contrib/views.py @@ -1,21 +1,32 @@ """ -Django implementation of the BatchDataResolver interface. -This module provides a way to execute multiple database queries in a single request. +This module provides APIs that are used by the Contribute Application (backend +and frontend). """ -from typing import Dict, List, Any, Optional, Union, Tuple -import json +from dataclasses import dataclass, field +from datetime import datetime, timedelta +import threading +import traceback +from typing import Dict, List, Any, Optional, Type, Union, Set +from collections import defaultdict, deque +from django.apps import apps +from django.db import connection, transaction +from django.db.models import Q, Model from django.http import JsonResponse +from django.views import View from django.views.decorators.http import require_POST from django.views.decorators.csrf import csrf_exempt -from django.db.models import Q, Model -from django.apps import apps - +from django.db import models +from voyage.models import Voyage from geo.common import GeoTreeFilter +import json +import logging # Type definitions to match the TypeScript interface NonNullFieldValue = Union[str, int, float, bool] BaseFieldValue = Optional[NonNullFieldValue] ResolvedEntityData = Dict[str, BaseFieldValue] + +logger = logging.getLogger(__name__) class DataFilter: """Represents a filter condition on a field.""" @@ -32,7 +43,7 @@ def to_q(self) -> Q: # Default to equals return Q(**{self.field: self.value}) -def remap(r: dict[str, any], remapped: list[str]): +def remap(r: Dict[str, Any], remapped: List[str]): for f in remapped: key = f.replace("_id", "") if key in r: @@ -40,7 +51,6 @@ def remap(r: dict[str, any], remapped: list[str]): del r[key] return r - class DataQuery: """Represents a query on a specific model with filters.""" @@ -113,7 +123,6 @@ def fetchBatch(batch: Dict[str, Dict[str, Any]]) -> Dict[str, List[ResolvedEntit results[key] = DataResolver.fetch(input_data) return results - @csrf_exempt @require_POST def batch_data_api(request): @@ -136,4 +145,501 @@ def batch_data_api(request): @csrf_exempt def location_tree(_): - return JsonResponse(GeoTreeFilter(select_all=True), safe=False) \ No newline at end of file + return JsonResponse(GeoTreeFilter(select_all=True), safe=False) + +_schemaToDbTable: Dict[str, str] = { + 'VoyageSparseDate': 'voyage_voyagesparsedate', + 'Nationality': 'voyage_nationality', + 'TonType': 'voyage_tontype', + 'RigOfVessel': 'voyage_rigofvessel', + 'Location': 'geo_location', + 'VoyageShip': 'voyage_voyageship', + 'VoyageItinerary': 'voyage_voyageitinerary', + 'VoyageSlaveNumbers': 'voyage_voyageslavesnumbers', + 'VoyageDates': 'voyage_voyagedates', + 'VoyageGrouping': 'voyage_voyagegroupings', + 'AfricanInfo': 'voyage_africaninfo', + 'CargoUnit': 'voyage_cargounit', + 'CargoType': 'voyage_cargotype', + 'VoyageCargoConnectionSchema': 'voyage_voyagecargoconnection', + 'ParticularOutcome': 'voyage_particularoutcome', + 'SlavesOutcome': 'voyage_slavesoutcome', + 'VesselOutcomeSchema': 'voyage_vesselcapturedoutcome', + 'OwnerOutcome': 'voyage_owneroutcome', + 'Resistance': 'voyage_resistance', + 'VoyageOutcome': 'voyage_voyageoutcome', + 'VoyageCrew': 'voyage_voyagecrew', + 'EnslaverAlias': 'past_enslaveralias', + 'Enslaver': 'past_enslaveridentity', + 'EnslaverAliasWithFK': 'past_enslaveralias', + 'EnslaverAliasWithIdentity': 'past_enslaveralias', + 'EnslavementRelationType': 'past_enslavementrelationtype', + 'EnslaverRole': 'past_enslaverrole', + 'EnslaverRelationRoleConn': 'past_enslaverinrelation_roles', + 'EnslaverInRelation': 'past_enslaverinrelation', + 'Enslaved': 'past_enslaved', + 'EnslavedInRelation': 'past_enslavedinrelation', + 'EnslavementRelation': 'past_enslavementrelation', + 'Voyage Source Type': 'document_sourcetype', + 'Voyage Source Short Reference': 'document_shortref', + 'Voyage Source': 'document_source', + 'Voyage Source Connection': 'document_sourcevoyageconnection', + 'Voyage': 'voyage_voyage' +} + +_schemaToModel: Dict[str, Type[models.Model]] = {} +for m in apps.get_models(include_auto_created=True): + for k, v in _schemaToDbTable.items(): + if m._meta.db_table == v: + _schemaToModel[k] = m +_missing = set() +for k in _schemaToDbTable.keys(): + if k not in _schemaToModel: + _missing.add(k) +if _missing: + logger.error(f"Missing model for schema: {list(_missing)}") +else: + logger.info("All models mapped successfully") + +class ChangeSetProcessor: + """Processes CombinedChangeSet and applies changes to Django ORM.""" + + def __init__(self): + self.temp_id_to_db_id: Dict[str, Any] = {} + self.new_entity_ids: Set[str] = set() + + def process_changeset(self, changeset: Dict) -> Dict[str, Any]: + """ + Process the entire changeset in a transaction. + Returns a mapping of temporary IDs to actual database IDs. + """ + deletions = changeset.get('deletions', []) + updates = changeset.get('updates', []) + + # Validate no duplicate entity updates and no updates to deleted entities + self._validate_no_duplicates(deletions, updates) + + # Collect all new entity IDs first + self.new_entity_ids = { + update['entityRef']['id'] + for update in updates + if update['entityRef'].get('type') == 'new' + } + + # Separate new and existing entity updates + new_updates = [u for u in updates if u['entityRef'].get('type') == 'new'] + existing_updates = [u for u in updates if u['entityRef'].get('type') != 'new'] + + # Sort new entities by FK dependencies + sorted_new_updates = self._topological_sort(new_updates) + + with transaction.atomic(): + # Process deletions first + for deletion in deletions: + self._process_deletion(deletion) + + # Process new entities in dependency order + for update in sorted_new_updates: + logger.debug(json.dumps(update)) + self._process_new_entity(update) + + # Process existing entity updates + for update in existing_updates: + self._process_existing_entity(update) + + return self.temp_id_to_db_id + + def _validate_no_duplicates(self, deletions: List[Dict], updates: List[Dict]): + """Validate that each entity is updated at most once and deleted entities aren't updated.""" + updated_entities = set() + deleted_entities = set() + + # Collect deleted entity references + for deletion in deletions: + entity_key = self._get_entity_key(deletion['entityRef']) + deleted_entities.add(entity_key) + + # Check for duplicate updates and updates to deleted entities + for update in updates: + entity_key = self._get_entity_key(update['entityRef']) + + if entity_key in deleted_entities: + raise ValueError(f"Cannot update deleted entity: {entity_key}") + + if entity_key in updated_entities: + raise ValueError(f"Duplicate update for entity: {entity_key}") + + updated_entities.add(entity_key) + + def _get_entity_key(self, entity_ref: Dict) -> str: + """Create a unique key for an entity reference.""" + return f"{entity_ref['schema']}:{entity_ref['id']}" + + def _topological_sort(self, new_updates: List[Dict]) -> List[Dict]: + """ + Sort new entities based on foreign key dependencies. + Returns sorted list or raises ValueError if cycle detected. + """ + # Build dependency graph + graph = defaultdict(list) # entity_id -> list of entities that depend on it + in_degree = defaultdict(int) # entity_id -> number of dependencies + entity_map = {} # entity_id -> update dict + + for update in new_updates: + entity_id = update['entityRef']['id'] + entity_map[entity_id] = update + + # Find FK dependencies to other new entities + for change in update['changes']: + if change.get('isForeignKey') and change.get('changed'): + fk_value = change['changed'] + # Check if FK points to a new entity (using set membership) + if fk_value in self.new_entity_ids: + graph[fk_value].append(entity_id) + in_degree[entity_id] += 1 + + # Initialize queue with entities that have no dependencies + queue = deque() + for update in new_updates: + entity_id = update['entityRef']['id'] + if in_degree[entity_id] == 0: + queue.append(entity_id) + + # Process queue (Kahn's algorithm) + sorted_ids = [] + while queue: + current_id = queue.popleft() + sorted_ids.append(current_id) + + # Reduce in-degree for dependent entities + for dependent_id in graph[current_id]: + in_degree[dependent_id] -= 1 + if in_degree[dependent_id] == 0: + queue.append(dependent_id) + + # Check for cycles + if len(sorted_ids) != len(new_updates): + raise ValueError("Circular dependency detected in new entities") + + # Return updates in sorted order + return [entity_map[entity_id] for entity_id in sorted_ids] + + def _process_deletion(self, deletion: Dict): + """Process a deletion operation.""" + entity_ref = deletion['entityRef'] + model_class = self._get_model_class(entity_ref['schema']) + + try: + obj = model_class.objects.get(pk=entity_ref['id']) + obj.delete() + logger.info(f"Deleted {entity_ref['schema']} with id {entity_ref['id']}") + except model_class.DoesNotExist: + logger.warning(f"Entity {entity_ref['schema']} with id {entity_ref['id']} not found for deletion") + + def _process_new_entity(self, update: Dict): + """Process creation of a new entity.""" + entity_ref = update['entityRef'] + model_class = self._get_model_class(entity_ref['schema']) + + # Prepare field values + field_values = {} + for change in update['changes']: + field_name = change['property'] + value = change['changed'] + + # Replace temporary FK references with actual IDs + if change.get('isForeignKey') and value and value in self.new_entity_ids: + if value not in self.temp_id_to_db_id: + raise ValueError(f"Foreign key references unprocessed entity: {value}") + value = self.temp_id_to_db_id[value] + + field_values[field_name] = value + + # Create the entity - let Django handle field validation + obj = model_class(**field_values) + # Special cases: + if (model_class == Voyage): + # (domingos): I don't know why we have both id and voyage_id in the + # model, but at this point it is better to make sure they're equal. + obj.id = obj.voyage_id + obj.save() + + # Store the mapping from temporary ID to actual database ID + self.temp_id_to_db_id[entity_ref['id']] = obj.pk + + logger.info(f"Created {entity_ref['schema']} with temp id {entity_ref['id']} -> db id {obj.pk}") + + def _process_existing_entity(self, update: Dict): + """Process update of an existing entity.""" + entity_ref = update['entityRef'] + model_class = self._get_model_class(entity_ref['schema']) + + try: + obj = model_class.objects.get(pk=entity_ref['id']) + + # Apply changes + for change in update['changes']: + field_name = change['property'] + value = change['changed'] + + # Replace temporary FK references with actual IDs + if change.get('isForeignKey') and value and value in self.new_entity_ids: + if value not in self.temp_id_to_db_id: + raise ValueError(f"Foreign key references unprocessed entity: {value}") + value = self.temp_id_to_db_id[value] + + setattr(obj, field_name, value) + + obj.save() + logger.info(f"Updated {entity_ref['schema']} with id {entity_ref['id']}") + + except model_class.DoesNotExist: + raise ValueError(f"Entity {entity_ref['schema']} with id {entity_ref['id']} not found for update") + + def _get_model_class(self, schema_name: str): + """Get Django model class from schema name.""" + # Try to get model from any installed app + if schema_name not in _schemaToModel: + raise ValueError(f"Model {schema_name} not found in any app") + return _schemaToModel[schema_name] + + +class PublicationStatus: + PENDING = "pending" + PROCESSING = "processing" + COMPLETED = "completed" + FAILED = "failed" + + +@dataclass +class PublicationTask: + """Represents a changeset publication task.""" + publication_key: str + status: PublicationStatus + created_at: datetime + started_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + changeset: Dict = field(default_factory=dict) + result: Optional[Dict] = None + error: Optional[str] = None + id_mappings: Dict[str, Any] = field(default_factory=dict) + + +# TODO: move the TaskStore to the database so that it can be shared across +# workers. + +class TaskStore: + """In-memory storage for publication tasks.""" + + def __init__(self, max_age_hours: int = 24): + self._tasks: Dict[str, PublicationTask] = {} + self._lock = threading.Lock() + self.max_age_hours = max_age_hours + + def get(self, publication_key: str) -> Optional[PublicationTask]: + """Get a task by publication key.""" + with self._lock: + self._cleanup_old_tasks() + return self._tasks.get(publication_key) + + def create(self, publication_key: str, changeset: Dict) -> PublicationTask: + """Create or return existing task.""" + with self._lock: + # Check if task already exists (idempotency) + if publication_key in self._tasks: + return self._tasks[publication_key] + + # Create new task + task = PublicationTask( + publication_key=publication_key, + status=PublicationStatus.PENDING, + created_at=datetime.now(), + changeset=changeset + ) + self._tasks[publication_key] = task + return task + + def update_status(self, publication_key: str, status: PublicationStatus, + result: Optional[Dict] = None, error: Optional[str] = None): + """Update task status.""" + with self._lock: + task = self._tasks.get(publication_key) + if task: + task.status = status + if status == PublicationStatus.PROCESSING: + task.started_at = datetime.now() + elif status in (PublicationStatus.COMPLETED, PublicationStatus.FAILED): + task.completed_at = datetime.now() + if result: + task.result = result + if error: + task.error = error + + def _cleanup_old_tasks(self): + """Remove tasks older than max_age_hours.""" + cutoff = datetime.now() - timedelta(hours=self.max_age_hours) + expired_keys = [ + key for key, task in self._tasks.items() + if task.created_at < cutoff + ] + for key in expired_keys: + del self._tasks[key] + logger.info(f"Cleaned up expired task: {key}") + + +# Global task store (singleton for the application instance) +task_store = TaskStore() + +def process_changeset_background(publication_key: str): + """Background task to process a changeset.""" + task = task_store.get(publication_key) + if not task: + logger.error(f"Task not found for publication key: {publication_key}") + return + + try: + # Update status to processing + task_store.update_status(publication_key, PublicationStatus.PROCESSING) + logger.info(f"Starting processing for publication: {publication_key}") + + # Process the changeset + processor = ChangeSetProcessor() + + # Close any existing DB connections to ensure thread safety + connection.close() + + id_mappings = processor.process_changeset(task.changeset) + + # Update task with success + task_store.update_status( + publication_key, + PublicationStatus.COMPLETED, + result={ + 'success': True, + 'id_mappings': id_mappings, + 'message': 'Changeset processed successfully' + } + ) + logger.info(f"Successfully processed publication: {publication_key}") + + except ValueError as e: + logger.error(f"Validation error for publication {publication_key}: {e}") + task_store.update_status( + publication_key, + PublicationStatus.FAILED, + error=traceback.format_exc() + ) + except Exception as e: + logger.error(f"Unexpected error processing publication {publication_key}: {e}", exc_info=True) + task_store.update_status( + publication_key, + PublicationStatus.FAILED, + error=f"Internal error: {traceback.format_exc()}" + ) + finally: + # Ensure DB connection is closed + connection.close() + +@csrf_exempt +@require_POST +def publish_batch(request): + """ + Initiate processing of a CombinedChangeSet. + Expects JSON body with 'publication_key' and 'changeset'. + """ + try: + # Parse JSON body + if not request.body: + return JsonResponse({'error': 'Empty request body'}, status=400) + + data = json.loads(request.body) + + # Extract publication key and changeset + publication_key = data.get('key') + if not publication_key: + return JsonResponse({'error': 'Missing idempotency key'}, status=400) + + changeset = data.get('changeset') + if not changeset: + return JsonResponse({'error': 'Missing changeset'}, status=400) + + # Create or get existing task (idempotency) + task = task_store.create(publication_key, changeset) + + # If task is already completed or failed, return immediately + if task.status == PublicationStatus.COMPLETED: + return JsonResponse(task.result) + elif task.status == PublicationStatus.FAILED: + return JsonResponse({'error': task.error}, status=400) + elif task.status == PublicationStatus.PROCESSING: + return JsonResponse({ + 'status': 'processing', + 'publication_key': publication_key, + 'message': 'Publication is already being processed' + }) + + # Start background processing for new task + thread = threading.Thread( + target=process_changeset_background, + args=(publication_key,), + daemon=True + ) + thread.start() + + return JsonResponse({ + 'status': 'accepted', + 'publication_key': publication_key, + 'message': 'Publication queued for processing' + }, status=202) + + except json.JSONDecodeError as e: + logger.error(f"JSON parse error: {e}") + return JsonResponse({'error': 'Invalid JSON format'}, status=400) + except Exception as e: + logger.error(f"Unexpected error in publish_batch: {e}", exc_info=True) + return JsonResponse({'error': str(e)}, status=500) + +def publication_status(request, publication_key): + """ + Check the status of a publication task. + Returns current status and result if completed. + """ + try: + task = task_store.get(publication_key) + + if not task: + return JsonResponse({ + 'error': 'Publication not found', + 'publication_key': publication_key + }, status=404) + + # Build response based on task status + response = { + 'publication_key': publication_key, + 'status': task.status, + 'created_at': task.created_at.isoformat(), + } + + if task.started_at: + response['started_at'] = task.started_at.isoformat() + + if task.completed_at: + response['completed_at'] = task.completed_at.isoformat() + duration = (task.completed_at - task.started_at).total_seconds() + response['duration_seconds'] = duration + + if task.status == PublicationStatus.COMPLETED: + response.update(task.result) + elif task.status == PublicationStatus.FAILED: + response['error'] = task.error + return JsonResponse(response, status=400) + elif task.status == PublicationStatus.PROCESSING: + response['message'] = 'Publication is being processed' + elif task.status == PublicationStatus.PENDING: + response['message'] = 'Publication is queued for processing' + + return JsonResponse(response) + + except Exception as e: + logger.error(f"Error checking publication status: {e}", exc_info=True) + return JsonResponse({'error': str(e)}, status=500) diff --git a/src/api/voyages3/settings.py b/src/api/voyages3/settings.py index 5c1a28e7..838dde0c 100644 --- a/src/api/voyages3/settings.py +++ b/src/api/voyages3/settings.py @@ -203,3 +203,6 @@ # https://docs.djangoproject.com/en/3.2/ref/settings/#default-auto-field DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField' + +# Setting a large upload size for publication to work. +DATA_UPLOAD_MAX_MEMORY_SIZE = 50 * 1024 * 1024 # 50MB From b046df217b73939d892b1f75421b3ad7d2bf86bf Mon Sep 17 00:00:00 2001 From: "Domingos Dellamonica Jr." Date: Mon, 27 Oct 2025 13:15:18 -0300 Subject: [PATCH 9/9] Publication task is now stored in Redis cache. This allows a worker different from the one actually running the publicaation to respond to a state query. --- src/api/contrib/views.py | 277 ++++++++++++++++++++++++++++----------- 1 file changed, 198 insertions(+), 79 deletions(-) diff --git a/src/api/contrib/views.py b/src/api/contrib/views.py index 52b7afac..195706bf 100644 --- a/src/api/contrib/views.py +++ b/src/api/contrib/views.py @@ -20,6 +20,9 @@ from geo.common import GeoTreeFilter import json import logging +import redis +import pickle +from voyages3.localsettings import REDIS_HOST, REDIS_PORT # Type definitions to match the TypeScript interface NonNullFieldValue = Union[str, int, float, bool] @@ -201,6 +204,30 @@ def location_tree(_): else: logger.info("All models mapped successfully") + +class PublicationStatus: + PENDING = "pending" + PROCESSING = "processing" + COMPLETED = "completed" + FAILED = "failed" + + +@dataclass +class PublicationTask: + """Represents a changeset publication task.""" + publication_key: str + status: PublicationStatus + contribution_ids: List[str] + created_at: datetime + updated_at: Optional[datetime] = None + completed_at: Optional[datetime] = None + changeset: Dict = field(default_factory=dict) + result: Optional[Dict] = None + error: Optional[str] = None + id_mappings: Dict[str, Any] = field(default_factory=dict) + processed: int = 0 + + class ChangeSetProcessor: """Processes CombinedChangeSet and applies changes to Django ORM.""" @@ -208,14 +235,23 @@ def __init__(self): self.temp_id_to_db_id: Dict[str, Any] = {} self.new_entity_ids: Set[str] = set() - def process_changeset(self, changeset: Dict) -> Dict[str, Any]: + def process_changeset(self, task: PublicationTask) -> Dict[str, Any]: """ Process the entire changeset in a transaction. Returns a mapping of temporary IDs to actual database IDs. """ + changeset = task.changeset deletions = changeset.get('deletions', []) updates = changeset.get('updates', []) - + processed = 0 + + def increment_processed(): + nonlocal processed + processed += 1 + if (processed % 100 == 0): + logger.info(f"Processed {processed} operations...") + task_store.update_status(task.publication_key, status=PublicationStatus.PROCESSING, processed=processed) + # Validate no duplicate entity updates and no updates to deleted entities self._validate_no_duplicates(deletions, updates) @@ -236,15 +272,18 @@ def process_changeset(self, changeset: Dict) -> Dict[str, Any]: with transaction.atomic(): # Process deletions first for deletion in deletions: + increment_processed() self._process_deletion(deletion) # Process new entities in dependency order for update in sorted_new_updates: logger.debug(json.dumps(update)) + increment_processed() self._process_new_entity(update) # Process existing entity updates for update in existing_updates: + increment_processed() self._process_existing_entity(update) return self.temp_id_to_db_id @@ -332,7 +371,7 @@ def _process_deletion(self, deletion: Dict): try: obj = model_class.objects.get(pk=entity_ref['id']) obj.delete() - logger.info(f"Deleted {entity_ref['schema']} with id {entity_ref['id']}") + logger.debug(f"Deleted {entity_ref['schema']} with id {entity_ref['id']}") except model_class.DoesNotExist: logger.warning(f"Entity {entity_ref['schema']} with id {entity_ref['id']} not found for deletion") @@ -367,7 +406,7 @@ def _process_new_entity(self, update: Dict): # Store the mapping from temporary ID to actual database ID self.temp_id_to_db_id[entity_ref['id']] = obj.pk - logger.info(f"Created {entity_ref['schema']} with temp id {entity_ref['id']} -> db id {obj.pk}") + logger.debug(f"Created {entity_ref['schema']} with temp id {entity_ref['id']} -> db id {obj.pk}") def _process_existing_entity(self, update: Dict): """Process update of an existing entity.""" @@ -391,7 +430,7 @@ def _process_existing_entity(self, update: Dict): setattr(obj, field_name, value) obj.save() - logger.info(f"Updated {entity_ref['schema']} with id {entity_ref['id']}") + logger.debug(f"Updated {entity_ref['schema']} with id {entity_ref['id']}") except model_class.DoesNotExist: raise ValueError(f"Entity {entity_ref['schema']} with id {entity_ref['id']} not found for update") @@ -404,87 +443,168 @@ def _get_model_class(self, schema_name: str): return _schemaToModel[schema_name] -class PublicationStatus: - PENDING = "pending" - PROCESSING = "processing" - COMPLETED = "completed" - FAILED = "failed" - - -@dataclass -class PublicationTask: - """Represents a changeset publication task.""" - publication_key: str - status: PublicationStatus - created_at: datetime - started_at: Optional[datetime] = None - completed_at: Optional[datetime] = None - changeset: Dict = field(default_factory=dict) - result: Optional[Dict] = None - error: Optional[str] = None - id_mappings: Dict[str, Any] = field(default_factory=dict) - - -# TODO: move the TaskStore to the database so that it can be shared across -# workers. - class TaskStore: - """In-memory storage for publication tasks.""" + """Redis-backed storage for publication tasks.""" - def __init__(self, max_age_hours: int = 24): - self._tasks: Dict[str, PublicationTask] = {} - self._lock = threading.Lock() + def __init__(self, max_age_hours: int = 1): + self.redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=False) self.max_age_hours = max_age_hours + self.key_prefix = "taskstore:publication:" + self.cleanup_interval = timedelta(hours=1) + self._last_cleanup = datetime.now() + + def _get_redis_key(self, publication_key: str) -> str: + """Generate Redis key for a publication task.""" + return f"{self.key_prefix}{publication_key}" + + def _serialize_task(self, task: PublicationTask) -> bytes: + """Serialize task to bytes for Redis storage.""" + # Convert dataclass to dict + task_dict = { + 'publication_key': task.publication_key, + 'status': task.status, + 'contribution_ids': task.contribution_ids, + 'created_at': task.created_at.isoformat() if task.created_at else None, + 'updated_at': task.updated_at.isoformat() if task.updated_at else None, + 'completed_at': task.completed_at.isoformat() if task.completed_at else None, + 'changeset': task.changeset, + 'result': task.result, + 'error': task.error, + 'id_mappings': task.id_mappings, + 'processed': task.processed + } + return pickle.dumps(task_dict) + + def _deserialize_task(self, data: bytes) -> PublicationTask: + """Deserialize task from Redis bytes.""" + task_dict = pickle.loads(data) + return PublicationTask( + publication_key=task_dict['publication_key'], + status=task_dict['status'], + contribution_ids=task_dict['contribution_ids'], + created_at=datetime.fromisoformat(task_dict['created_at']) if task_dict['created_at'] else None, + updated_at=datetime.fromisoformat(task_dict['updated_at']) if task_dict['updated_at'] else None, + completed_at=datetime.fromisoformat(task_dict['completed_at']) if task_dict['completed_at'] else None, + changeset=task_dict['changeset'], + result=task_dict['result'], + error=task_dict['error'], + id_mappings=task_dict['id_mappings'], + processed=task_dict['processed'] + ) def get(self, publication_key: str) -> Optional[PublicationTask]: """Get a task by publication key.""" - with self._lock: - self._cleanup_old_tasks() - return self._tasks.get(publication_key) + try: + # Periodic cleanup + self._cleanup_old_tasks_if_needed() + + redis_key = self._get_redis_key(publication_key) + data = self.redis_client.get(redis_key) + if data: + return self._deserialize_task(data) + return None + except Exception as e: + logger.error(f"Error getting task from Redis: {e}") + return None - def create(self, publication_key: str, changeset: Dict) -> PublicationTask: + def create(self, publication_key: str, changeset: Dict, contribution_ids: List[str]) -> PublicationTask: """Create or return existing task.""" - with self._lock: + try: + redis_key = self._get_redis_key(publication_key) + # Check if task already exists (idempotency) - if publication_key in self._tasks: - return self._tasks[publication_key] + existing_data = self.redis_client.get(redis_key) + if existing_data: + return self._deserialize_task(existing_data) # Create new task task = PublicationTask( publication_key=publication_key, status=PublicationStatus.PENDING, created_at=datetime.now(), - changeset=changeset + changeset=changeset, + contribution_ids=contribution_ids ) - self._tasks[publication_key] = task + + # Store in Redis with expiration + serialized_task = self._serialize_task(task) + expiration_seconds = self.max_age_hours * 3600 + self.redis_client.setex(redis_key, expiration_seconds, serialized_task) + return task + except Exception as e: + logger.error(f"Error creating task in Redis: {e}") + raise def update_status(self, publication_key: str, status: PublicationStatus, - result: Optional[Dict] = None, error: Optional[str] = None): + result: Optional[Dict] = None, error: Optional[str] = None, + processed: Optional[int] = None): """Update task status.""" - with self._lock: - task = self._tasks.get(publication_key) - if task: - task.status = status - if status == PublicationStatus.PROCESSING: - task.started_at = datetime.now() - elif status in (PublicationStatus.COMPLETED, PublicationStatus.FAILED): - task.completed_at = datetime.now() - if result: - task.result = result - if error: - task.error = error + try: + redis_key = self._get_redis_key(publication_key) + data = self.redis_client.get(redis_key) + + if not data: + logger.warning(f"Task not found for update: {publication_key}") + return + + task = self._deserialize_task(data) + + # Update task fields + task.status = status + if status == PublicationStatus.PROCESSING: + task.updated_at = datetime.now() + elif status in (PublicationStatus.COMPLETED, PublicationStatus.FAILED): + task.completed_at = datetime.now() + if result: + task.result = result + if error: + task.error = error + if processed is not None: + task.processed = processed + + # Save back to Redis + serialized_task = self._serialize_task(task) + expiration_seconds = self.max_age_hours * 3600 + self.redis_client.setex(redis_key, expiration_seconds, serialized_task) + + except Exception as e: + logger.error(f"Error updating task in Redis: {e}") + + def _cleanup_old_tasks_if_needed(self): + """Remove expired tasks if cleanup interval has passed.""" + now = datetime.now() + if now - self._last_cleanup > self.cleanup_interval: + self._cleanup_old_tasks() + self._last_cleanup = now def _cleanup_old_tasks(self): """Remove tasks older than max_age_hours.""" - cutoff = datetime.now() - timedelta(hours=self.max_age_hours) - expired_keys = [ - key for key, task in self._tasks.items() - if task.created_at < cutoff - ] - for key in expired_keys: - del self._tasks[key] - logger.info(f"Cleaned up expired task: {key}") + try: + pattern = f"{self.key_prefix}*" + keys = self.redis_client.keys(pattern) + + cutoff = datetime.now() - timedelta(hours=self.max_age_hours) + expired_count = 0 + + for key in keys: + try: + data = self.redis_client.get(key) + if data: + task = self._deserialize_task(data) + if task.created_at and task.created_at < cutoff: + self.redis_client.delete(key) + expired_count += 1 + except Exception as e: + logger.warning(f"Error checking task age for cleanup: {e}") + # Delete corrupted entries + self.redis_client.delete(key) + expired_count += 1 + + if expired_count > 0: + logger.info(f"Cleaned up {expired_count} expired publication tasks") + except Exception as e: + logger.error(f"Error during task cleanup: {e}") # Global task store (singleton for the application instance) @@ -499,16 +619,12 @@ def process_changeset_background(publication_key: str): try: # Update status to processing + processor = ChangeSetProcessor() task_store.update_status(publication_key, PublicationStatus.PROCESSING) logger.info(f"Starting processing for publication: {publication_key}") # Process the changeset - processor = ChangeSetProcessor() - - # Close any existing DB connections to ensure thread safety - connection.close() - - id_mappings = processor.process_changeset(task.changeset) + id_mappings = processor.process_changeset(task) # Update task with success task_store.update_status( @@ -536,9 +652,6 @@ def process_changeset_background(publication_key: str): PublicationStatus.FAILED, error=f"Internal error: {traceback.format_exc()}" ) - finally: - # Ensure DB connection is closed - connection.close() @csrf_exempt @require_POST @@ -562,10 +675,14 @@ def publish_batch(request): changeset = data.get('changeset') if not changeset: return JsonResponse({'error': 'Missing changeset'}, status=400) - + + contribution_ids = data.get('contribution_ids', []) + if not isinstance(contribution_ids, list): + return JsonResponse({'error': 'Invalid contribution_ids format'}, status=400) + # Create or get existing task (idempotency) - task = task_store.create(publication_key, changeset) - + task = task_store.create(publication_key, changeset, contribution_ids) + # If task is already completed or failed, return immediately if task.status == PublicationStatus.COMPLETED: return JsonResponse(task.result) @@ -617,15 +734,17 @@ def publication_status(request, publication_key): response = { 'publication_key': publication_key, 'status': task.status, - 'created_at': task.created_at.isoformat(), + 'contribution_ids': task.contribution_ids, + 'processed_operations': task.processed, + 'created_at': task.created_at.isoformat() if task.created_at else None, } - if task.started_at: - response['started_at'] = task.started_at.isoformat() + if task.updated_at: + response['updated_at'] = task.updated_at.isoformat() if task.completed_at: response['completed_at'] = task.completed_at.isoformat() - duration = (task.completed_at - task.started_at).total_seconds() + duration = (task.completed_at - task.updated_at).total_seconds() response['duration_seconds'] = duration if task.status == PublicationStatus.COMPLETED: