Skip to content

oarepo_oaipmh_harvester.jobs

oarepo_oaipmh_harvester.jobs

invenio-jobs integration - one job type per harvester.

invenio-jobs supports run parameters only, not job parameters, so a single job type can not be scheduled separately for different harvesters. register_oaipmh_job therefore creates a subclass of OAIHarvestJob with the id harvest_oaipmh_records_<harvester id> for every harvester - when the harvester is created and for all stored harvesters at application start (register_current_harvesters). invenio-jobs passes the start of the last successful run as since, so scheduled runs are incremental.

OAIHarvestJob

Bases: JobType

OAI Harvester Job.

Source code in oarepo_oaipmh_harvester/jobs.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class OAIHarvestJob(JobType):
    """OAI Harvester Job."""

    id = "harvest_oaipmh_records"
    """ID is passed by the subclass when harvester is created.

    Note: jobs do not have parameters, only run parameters, but we need to pass
    harvester_id and passing them as run parameters is not possible when
    scheduling the job from the administration interface. Because of that,
    we create a new subclass of this job for each harvester with the harvester
    code embedded in the job ID.

    The id of the job is then "harvest_oaipmh_records_{harvester_id}" and
    this id is never used.
    """

    title = "OAI Harvester Job"
    """Title is passed by the subclass when harvester is created."""

    description = "Harvester job for OAI-PMH Harvester"
    """Description of the harvester job."""

    task = harvest_oaipmh_records

    @classmethod
    @override
    def build_task_arguments(
        cls,
        job_obj: Job,
        since: datetime | str | None = None,
        **kwargs: Any,
    ) -> dict:
        """Build task arguments for the job."""
        if since and isinstance(since, datetime):
            # keep full ISO here; create_readers formats it to the
            # OAI-PMH datestamp granularity advertised by the target server
            since = since.isoformat()
        return {
            "harvester_id": cls.id.replace("harvest_oaipmh_records_", ""),
            "since": since,
            "oai_ids": kwargs.get("oai_ids"),
        }

id = 'harvest_oaipmh_records' class-attribute instance-attribute

ID is passed by the subclass when harvester is created.

Note: jobs do not have parameters, only run parameters, but we need to pass harvester_id and passing them as run parameters is not possible when scheduling the job from the administration interface. Because of that, we create a new subclass of this job for each harvester with the harvester code embedded in the job ID.

The id of the job is then "harvest_oaipmh_records_{harvester_id}" and this id is never used.

title = 'OAI Harvester Job' class-attribute instance-attribute

Title is passed by the subclass when harvester is created.

description = 'Harvester job for OAI-PMH Harvester' class-attribute instance-attribute

Description of the harvester job.

build_task_arguments(job_obj: Job, since: datetime | str | None = None, **kwargs: Any) -> dict classmethod

Build task arguments for the job.

Source code in oarepo_oaipmh_harvester/jobs.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
@classmethod
@override
def build_task_arguments(
    cls,
    job_obj: Job,
    since: datetime | str | None = None,
    **kwargs: Any,
) -> dict:
    """Build task arguments for the job."""
    if since and isinstance(since, datetime):
        # keep full ISO here; create_readers formats it to the
        # OAI-PMH datestamp granularity advertised by the target server
        since = since.isoformat()
    return {
        "harvester_id": cls.id.replace("harvest_oaipmh_records_", ""),
        "since": since,
        "oai_ids": kwargs.get("oai_ids"),
    }

register_oaipmh_job(harvester_id: str) -> None

Register a new OAI-PMH harvester job.

Note: current invenio-jobs does not support job parameters, only run parameters. Because harvesters are created dynamically in a database and each harvester needs its own job (otherwise it would not be possible to schedule them separately), we need to create a new job class for each harvester.

This is done dynamically whenever a new harvester is created or during server startup inside the finalize_apps function.

Source code in oarepo_oaipmh_harvester/jobs.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def register_oaipmh_job(harvester_id: str) -> None:
    """Register a new OAI-PMH harvester job.

    Note: current invenio-jobs does not support job parameters, only run parameters.
    Because harvesters are created dynamically in a database and each harvester
    needs its own job (otherwise it would not be possible to schedule them separately),
    we need to create a new job class for each harvester.

    This is done dynamically whenever a new harvester is created or during server startup
    inside the finalize_apps function.
    """
    # create and register the job
    job_id = get_oaipmh_job_id(harvester_id)
    # unregister if exists
    current_jobs.registry._jobs.pop(job_id, None)  # noqa SLF001 no public accessor
    current_jobs.registry.register(
        type(
            f"OAIHarvestJob_{job_id}",
            (OAIHarvestJob,),
            {
                "id": job_id,
                "title": f"OAI Harvester Job for harvester {harvester_id}",
            },
        )
    )

get_oaipmh_job_id(harvester_id: str) -> str

Get the job ID for the given harvester ID.

Source code in oarepo_oaipmh_harvester/jobs.py 
103
104
105
def get_oaipmh_job_id(harvester_id: str) -> str:
    """Get the job ID for the given harvester ID."""
    return f"harvest_oaipmh_records_{harvester_id}"

get_oaipmh_job(harvester_id: str) -> Job

Get the job for the given harvester ID.

Source code in oarepo_oaipmh_harvester/jobs.py 
108
109
110
111
112
def get_oaipmh_job(harvester_id: str) -> Job:
    """Get the job for the given harvester ID."""
    job_type_id = get_oaipmh_job_id(harvester_id)
    job_type: JobType = current_jobs.registry.get(job_type_id)
    return db.session.query(Job).filter_by(task=job_type.task).one()

unregister_oaipmh_job(job_id: str) -> None

Unregister an OAI-PMH harvester job.

Source code in oarepo_oaipmh_harvester/jobs.py 
115
116
117
118
def unregister_oaipmh_job(job_id: str) -> None:
    """Unregister an OAI-PMH harvester job."""
    # unregister the job
    current_jobs.registry._jobs.pop(job_id, None)  # noqa SLF001 no public accessor

register_current_harvesters() -> None

Register jobs for all harvesters stored in the database.

Source code in oarepo_oaipmh_harvester/jobs.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def register_current_harvesters() -> None:
    """Register jobs for all harvesters stored in the database."""
    try:
        for harvester in db.session.query(OAIHarvester).all():
            register_oaipmh_job(harvester.id)
    # An exception might happen for example during invenio db init when
    # the application is started but the database is not yet initialized.
    # We ignore these exceptions with a general except clause, as different
    # kinds of exceptions might happen depending on the situation.
    # We only re-raise SystemExit and KeyboardInterrupt to allow proper
    # shutdown of the application in case of user interruption.
    except SystemExit:
        raise
    except KeyboardInterrupt:
        raise
    except Exception:  # noqa BLE001 ok here
        db.session.rollback()