diff --git a/cds_migrator_kit/rdm/records/transform/entities/parent.py b/cds_migrator_kit/rdm/records/transform/entities/parent.py index 9f966be3..adfd130a 100644 --- a/cds_migrator_kit/rdm/records/transform/entities/parent.py +++ b/cds_migrator_kit/rdm/records/transform/entities/parent.py @@ -19,6 +19,10 @@ EMAIL_PATTERN = re.compile(r"[^@]+@[^@]+\.[^@]+") +# Cache submitter email → user_id across records. Stable for one migration +# run since users are created before the records stream starts. +_submitter_id_by_email: dict = {} + class RecordParent: """The parent record for one migrated CDS record. @@ -77,9 +81,12 @@ def _build_access(self): email = self.dojson_entry.pop("submitter", None) if not email: owner = "system" + elif email in _submitter_id_by_email: + owner = _submitter_id_by_email[email] else: try: user = User.query.filter_by(email=email).one() + _submitter_id_by_email[email] = user.id owner = user.id except NoResultFound: raise UnexpectedValue( diff --git a/cds_migrator_kit/rdm/records/transform/entities/request.py b/cds_migrator_kit/rdm/records/transform/entities/request.py index 9a57f759..af5c6f64 100644 --- a/cds_migrator_kit/rdm/records/transform/entities/request.py +++ b/cds_migrator_kit/rdm/records/transform/entities/request.py @@ -12,6 +12,12 @@ from cds_migrator_kit.errors import ManualImportRequired, RecordFlaggedCuration +# Cache reviewer lookup results across records. Maps the raw reviewer string +# (email or "Family, Given" name) to the resolved user_id (int) on success, or +# to the RecordFlaggedCuration exception instance when no account was found. +# Stable for one migration run: user accounts don't change during records stream. +_reviewer_id_cache: dict = {} + class RecordRequest: """A community-inclusion request for one migrated CDS record. @@ -67,14 +73,24 @@ def _resolve_reviewers(self, reviewer_names): """ resolved = [] for reviewer_name in reviewer_names: - try: - user = self._find_reviewer(reviewer_name) - reviewer_entry = {"user": str(user.id)} - except RecordFlaggedCuration as exc: + cached = _reviewer_id_cache.get(reviewer_name) + if cached is None and reviewer_name not in _reviewer_id_cache: + try: + user = self._find_reviewer(reviewer_name) + _reviewer_id_cache[reviewer_name] = user.id + cached = user.id + except RecordFlaggedCuration as exc: + _reviewer_id_cache[reviewer_name] = exc + cached = exc + + if isinstance(cached, RecordFlaggedCuration): self.migration_logger.add_information( - self.recid, {"message": exc.message, "value": exc.value} + self.recid, {"message": cached.message, "value": cached.value} ) reviewer_entry = {"user": "-1"} + else: + reviewer_entry = {"user": str(cached)} + if reviewer_entry not in resolved: resolved.append(reviewer_entry) return resolved diff --git a/cds_migrator_kit/rdm/records/transform/mappers/contributors.py b/cds_migrator_kit/rdm/records/transform/mappers/contributors.py index 9cb52a71..9f8098fe 100644 --- a/cds_migrator_kit/rdm/records/transform/mappers/contributors.py +++ b/cds_migrator_kit/rdm/records/transform/mappers/contributors.py @@ -20,14 +20,32 @@ from cds_migrator_kit.rdm.migration_config import VOCABULARIES_NAMES_SCHEMES from cds_migrator_kit.rdm.records.transform.mappers.base import FieldMapper +# Sentinel distinguishing "not yet cached" from "cached as None (not found)". +_MISSING = object() + +# Per-process caches. Plain-dict row representations avoid SQLAlchemy object +# expiry issues that occur across UoW commits between records. +_affiliation_ror_cache: dict = ( + {} +) # ror_normalized → bool (exists in AffiliationsMetadata) +_affiliation_legacy_cache: dict = ( + {} +) # legacy_name → {"curated","exact","not_exact"} or None +_person_id_to_user_id: dict = {} # cern person_id str → user_id int or None + def match_affiliation(affiliation_name, ctx): """Match an affiliation against `CDSMigrationAffiliationMapping` db table.""" dojson_entry = ctx.dojson_entry if is_ror(affiliation_name): ror = normalize_ror(affiliation_name) - name = AffiliationsMetadata.query.filter_by(pid=ror).one_or_none() - if name is None: + exists = _affiliation_ror_cache.get(ror, _MISSING) + if exists is _MISSING: + exists = ( + AffiliationsMetadata.query.filter_by(pid=ror).one_or_none() is not None + ) + _affiliation_ror_cache[ror] = exists + if not exists: raise ManualImportRequired( message="Affiliation {ror} does not exist in the AffiliationMetadata table".format( ror=ror @@ -41,20 +59,35 @@ def match_affiliation(affiliation_name, ctx): subfield=None, ) return {"id": normalize_ror(affiliation_name)} - # Step 1: search in the affiliation mapping (ROR organizations) - match = ctx.affiliations_mapping.query.filter_by( - legacy_affiliation_input=affiliation_name - ).one_or_none() - if match: + + # Legacy name lookup — cache the row fields as a plain dict to avoid + # SQLAlchemy object expiry that occurs across UoW commits. + cached = _affiliation_legacy_cache.get(affiliation_name, _MISSING) + if cached is _MISSING: + match = ctx.affiliations_mapping.query.filter_by( + legacy_affiliation_input=affiliation_name + ).one_or_none() + cached = ( + { + "curated": match.curated_affiliation, + "exact": match.ror_exact_match, + "not_exact": match.ror_not_exact_match, + } + if match is not None + else None + ) + _affiliation_legacy_cache[affiliation_name] = cached + + if cached is not None: # Step 1: check if there is a curated input - if match.curated_affiliation: - return match.curated_affiliation + if cached["curated"]: + return cached["curated"] # Step 2: check if there is an exact match - if match.ror_exact_match: - return {"id": normalize_ror(match.ror_exact_match)} + if cached["exact"]: + return {"id": normalize_ror(cached["exact"])} # Step 3: check if there is not exact match - if match.ror_not_exact_match: - _affiliation_ror_id = normalize_ror(match.ror_not_exact_match) + if cached["not_exact"]: + _affiliation_ror_id = normalize_ror(cached["not_exact"]) raise RecordFlaggedCuration( subfield="u", value={"id": _affiliation_ror_id}, @@ -120,9 +153,12 @@ def _lookup_person_id(creator): {}, ).get("identifier") if person_id: - ui = UserIdentity.query.filter_by(id=person_id).one_or_none() - if ui: - user_id = ui.user.id + user_id = _person_id_to_user_id.get(person_id, _MISSING) + if user_id is _MISSING: + ui = UserIdentity.query.filter_by(id=person_id).one_or_none() + user_id = ui.user.id if ui else None + _person_id_to_user_id[person_id] = user_id + if user_id is not None: names = NamesMetadata.query.filter_by(internal_id=str(user_id)).all() name = next( (name for name in names if "unlisted" not in name.json.get("tags", [])), diff --git a/cds_migrator_kit/reports/log.py b/cds_migrator_kit/reports/log.py index 698f6e04..c8cf7a27 100644 --- a/cds_migrator_kit/reports/log.py +++ b/cds_migrator_kit/reports/log.py @@ -88,6 +88,8 @@ def __init__( ] self.log_writer = csv.DictWriter(self.error_file, fieldnames=columns) self._temp_state_cache = {} + self._flush_interval = 100 + self._rows_since_flush = 0 def start_log(self): """Initialize logging file descriptors.""" @@ -116,6 +118,7 @@ def read_log(self): def finalise(self): """Finalise logging files.""" + self.error_file.flush() self.error_file.close() def add_log(self, exc, record=None, key=None, value=None): @@ -145,7 +148,10 @@ def add_log(self, exc, record=None, key=None, value=None): } self.log_writer.writerow(error_format) logger_migrator.error(exc) - self.error_file.flush() + self._rows_since_flush += 1 + if self._rows_since_flush >= self._flush_interval: + self.error_file.flush() + self._rows_since_flush = 0 def add_information(self, recid, state): """Save a temporary success state for recid. @@ -163,7 +169,10 @@ def finalise_record(self, recid): """Log recid as success.""" _state = self._temp_state_cache.pop(recid, {}) self.log_writer.writerow({"recid": recid, "clean": True, **_state}) - self.error_file.flush() + self._rows_since_flush += 1 + if self._rows_since_flush >= self._flush_interval: + self.error_file.flush() + self._rows_since_flush = 0 class RecordStateLogger: diff --git a/cds_migrator_kit/runner/runner.py b/cds_migrator_kit/runner/runner.py index 1ffe142c..975bf0ce 100644 --- a/cds_migrator_kit/runner/runner.py +++ b/cds_migrator_kit/runner/runner.py @@ -7,6 +7,8 @@ """InvenioRDM migration streams runner.""" +import logging +import time from pathlib import Path import yaml @@ -120,15 +122,21 @@ def __init__( def run(self): """Run ETL streams.""" + perf_logger = logging.getLogger("migrator-perf") self.migration_logger.start_log() self.record_state_logger.start_log() for stream in self.streams: + t0 = time.perf_counter() try: stream.run(cleanup=True) except Exception as e: self.migration_logger.add_log(e) raise e finally: + elapsed = time.perf_counter() - t0 + perf_logger.info( + f"Stream '{stream.name}' finished in {elapsed:.1f}s" + ) self.migration_logger.finalise() self.record_state_logger.finalise()