invenio-rdm-migrator 4.4.3

Development

pip install -e .

./run-tests.sh

How to run it

To run the migration you need:

• A running InvenioRDM instance.

• If your data contains references to other records (e.g. vocabularies), then it is also required to run the setup step.

invenio-cli services setup --force --no-demo-data

• Install Invenio-RDM-Migrator, any other dependencies must be handled in the Pipfile of your instance.

$ pip install invenio-rdm-migrator

• Create/edit the configuration file on your instance, for example streams.yaml:

data_dir: /path/to/data
tmp_dir: /path/to/tmp
state_dir: /path/to/state
log_dir: /path/to/logs
db_uri: postgresql+psycopg2://inveniordm:inveniordm@localhost:5432/inveniordm
old_secret_key: CHANGE_ME
new_secret_key: CHANGE_ME
records:
    extract:
        filename: /path/to/records.json

• You will need to create a small python script putting together the different blocks of the ETL. You can find an eample at my-site/site/my_site/migrator/__main__.py.

from invenio_rdm_migrator.streams import StreamDefinition
from invenio_rdm_migrator.streams.records import RDMRecordCopyLoad

if __name__ == "__main__":
    RecordStreamDefinition = StreamDefinition(
        name="records",
        extract_cls=JSONLExtract,
        transform_cls=ZenodoToRDMRecordTransform,
        load_cls=RDMRecordCopyLoad,
    )

    runner = Runner(
        stream_definitions=[
            RecordStreamDefinition,
        ],
        config_filepath="path/to/your/streams.yaml",
    )

    runner.run()

• Finally, you can execute the above code. Since it is in the __main__ file of the python package, you can run it as a module:

$ python -m my_site.migrator

• Once the migration has completed, in your instance you can reindex the data. Following the records example above, it would look like:

$ invenio-cli pyshell

In [1]: from invenio_access.permissions import system_identity
In [2]: from invenio_rdm_records.proxies import current_rdm_records_service
In [3]: current_rdm_records_service.rebuild_index(identity=system_identity)

ETL {Extract/Transform/Load} architecture

There are four packages in this module extract, transform, load, and streams. The first three correspond to the three steps of an ETL process. The streams package contains the logic to run the process and different stream-specific implementations of ETL classes (e.g. records).

class XMLExtract(Extract):
...

    def run(self):
        with open("file.xml") as file:
            for entry in file:
                yield xml.loads(entry)

{
    "record": self._record(entry),
    "draft": self._draft(entry),
    "parent": self._parent(entry),
}

$ /path/to/data/tables1668697280.943311
    |
    | - pidstore_pid.csv
    | - rdm_parents_metadata.csv
    | - rdm_records_metadata.csv
    | - rdm_versions_state.csv

{
    "tx_id": the actual transaction id, useful for debug and error handling
    "action": this information refers to the semantic meaning of the group
                   for example: record metadata update or file upload
    "operations": [
        {
            "op": c (create), u (update), d (delete)
            "table": the name of the table in the source system (e.g. Zenodo)
            "data": the transformed data, this can use any `Transform` implementation
        }
    ]
}

State

During a migration run, there is a need to share information across different streams or different generators on the same stream. For example, the records stream needs to access the UUID to slug map that was populated on the communities stream; or the drafts generator needs to know which parent records have been created on the records generator to keep the version state consistent.

All this information is persisted to a SQLite database. This state database is kept in memory during each stream processing, and it is persisted to disk if the stream finishes without errors. The state will be saved with the name of the stream (e.g. records.db) to avoid overwriting a previous state. Therefore, a migration can be restarted from any stream.

There are two ways to add more information to the state:

• Full entities, for example record or users, require their own DB table. Those must be defined at state.py:State._initialize_db. In addition, to abstract the access to that table, a state entity is required. It needs to be initialized in the Runner.py:Runner constructor and added the the state_entities dictionary.

• Independent value, for example the maximum value of generated primary keys. Those can be stored in the global_state. This state has two columns: key and value; adding information to it would look like {key: name_of_the_value, value: actual_value}.

Notes

Using python generators

Using generators instead of lists, allows us to iterate through the data only once and perform the E-T-L steps on them. Instead of loop for E, loop for T, loop for L. In addition, this allows us to have the csv files open during the writing and closing them at the end (open/close is an expensive op when done 3M times).

Changes

Version 4.4.1

• Fix default value for nullable model fields.

Version 4.4.0

• Add GitHub stream.

• Add verified_at and blocked_at for usesr models.

• Handle parent DOIs for records.

• Add media files to records and drafts.

• Add deletion_status to record models.

• Switch to orjson for JSON dumping/loading.

• Add multi-processing for transform.

• Refactor state to also use Python dict for caching.

Version 4.3.0

• Add community basic CRUD actions.

• Add DB session fixtures.

Version 4.2.0

• Rename FileUploadAction to DraftFileUploadAction.

Version 4.1.0

• Add file upload action.

• Add draft edit action.

Version 4.0.0

• Namespace actions by load and transform.

Version 3.1.0

• Add DatetimeMixin to transform timestamps into iso formatted date strings.

• Add JSONLoadMixin to load dictionaries from strings.

Version 3.0.0

• Operation instances have split the model and the data into two attributes.

• Add user actions.

• PostgreSQLTx resolve_references function has now a default behaviour (pass).

• Add nullable configuration to draft and user related models.

• Minor bug fixes.

Version 2.0.0

• Make state globally available.

• Refactor transactions into actions. Create transaction and load data classes.

• Removed empty kafka extract module.

• Improved error handling and created specialized classes.

• Move dict_set to utils.

• Remove Python 3.8 from test matrix.

Version 1.0.0

• Initial public release.