oai_harvesters fixture - harvesters loaded from app_data/oai_harvesters.yaml.
Registered via the oarepo.fixtures entry point. Each YAML entry uses the keys of the
invenio oai harvesters create command (id, name, base-url,
metadata-prefix, setspec, loader, model, transformer, writer,
harvest-manager, comment). Existing harvesters are updated.
When model is given and transformer / writer are not, the defaults are
oai-import{model:...} and oai-service{model:...}, as in the command line.
community (optional) is the slug of a community; it is passed as the community option
to every oai-service writer, which adds all harvested records to that community.
harvest-manager entries are e-mail addresses of existing users (or {"user": id}
dictionaries); they are stored as user ids.
HarvestersFixture
Bases: FixtureMixin
Harvesters fixture.
Source code in oarepo_oaipmh_harvester/fixtures.py
| class HarvestersFixture(FixtureMixin):
"""Harvesters fixture."""
def transform_data(self, entry: dict[str, Any]) -> dict[str, Any]:
"""Transform the raw entry data into a harvester data dictionary."""
transformers = list(entry.pop("transformer", []))
writers = list(entry.pop("writer", []))
model = entry.pop("model", None)
if model:
if not transformers:
transformers = [f'oai-import{{model:"{model}"}}']
if not writers:
writers = [f'oai-service{{model:"{model}"}}']
community = entry.pop("community", None)
if community:
writers = [_with_community(writer, community) for writer in writers]
harvester_data = {
"id": entry.pop("id"),
"name": entry.pop("name"),
"base_url": entry.pop("base-url"),
"metadata_prefix": entry.pop("metadata-prefix"),
"setspec": entry.pop("setspec", ""),
"loader": entry.pop("loader", "oai-pmh"),
"transformers": transformers,
"writers": writers,
"comment": entry.pop("comment", ""),
}
harvest_managers = [
{"user": lookup_user(manager)} if isinstance(manager, str) else manager
for manager in entry.pop("harvest-manager", [])
]
if harvest_managers:
harvester_data["harvest_managers"] = harvest_managers
return harvester_data
def create(self, entry: dict[str, Any]) -> None:
"""Create a new OAI-PMH harvester."""
# Handle model option to auto-generate transformers and writers
harvester_data = self.transform_data(entry)
try:
try:
current_oai_harvester_service.create(system_identity, harvester_data)
except Exception: # noqa BLE001 fallback is ok
db.session.rollback()
current_oai_harvester_service.update(system_identity, harvester_data["id"], harvester_data)
db.session.commit()
current_oai_harvester_service.indexer.refresh()
current_app.logger.info(
"Harvester '%s' created successfully with ID: %s",
harvester_data["name"],
harvester_data["id"],
)
except Exception:
db.session.rollback()
current_app.logger.exception("Error creating harvester %s (transformed data: %s)", entry, harvester_data)
raise
|
Transform the raw entry data into a harvester data dictionary.
Source code in oarepo_oaipmh_harvester/fixtures.py
| def transform_data(self, entry: dict[str, Any]) -> dict[str, Any]:
"""Transform the raw entry data into a harvester data dictionary."""
transformers = list(entry.pop("transformer", []))
writers = list(entry.pop("writer", []))
model = entry.pop("model", None)
if model:
if not transformers:
transformers = [f'oai-import{{model:"{model}"}}']
if not writers:
writers = [f'oai-service{{model:"{model}"}}']
community = entry.pop("community", None)
if community:
writers = [_with_community(writer, community) for writer in writers]
harvester_data = {
"id": entry.pop("id"),
"name": entry.pop("name"),
"base_url": entry.pop("base-url"),
"metadata_prefix": entry.pop("metadata-prefix"),
"setspec": entry.pop("setspec", ""),
"loader": entry.pop("loader", "oai-pmh"),
"transformers": transformers,
"writers": writers,
"comment": entry.pop("comment", ""),
}
harvest_managers = [
{"user": lookup_user(manager)} if isinstance(manager, str) else manager
for manager in entry.pop("harvest-manager", [])
]
if harvest_managers:
harvester_data["harvest_managers"] = harvest_managers
return harvester_data
|
create(entry: dict[str, Any]) -> None
Create a new OAI-PMH harvester.
Source code in oarepo_oaipmh_harvester/fixtures.py
| def create(self, entry: dict[str, Any]) -> None:
"""Create a new OAI-PMH harvester."""
# Handle model option to auto-generate transformers and writers
harvester_data = self.transform_data(entry)
try:
try:
current_oai_harvester_service.create(system_identity, harvester_data)
except Exception: # noqa BLE001 fallback is ok
db.session.rollback()
current_oai_harvester_service.update(system_identity, harvester_data["id"], harvester_data)
db.session.commit()
current_oai_harvester_service.indexer.refresh()
current_app.logger.info(
"Harvester '%s' created successfully with ID: %s",
harvester_data["name"],
harvester_data["id"],
)
except Exception:
db.session.rollback()
current_app.logger.exception("Error creating harvester %s (transformed data: %s)", entry, harvester_data)
raise
|