API reference¶

Plugin¶

The package registers the following Transpiler-Mate entry point:

Property	Value
Entry-point group	`transpiler_mate.plugins`
Entry-point name	`invenio-publish`
Entry-point object	`invenio_publish.plugin:invenio_publish`

The execution function receives a normalized TranspilerContext and an InvenioPublisherOptions instance. Normally the Transpiler-Mate runtime builds both and invokes the plugin.

Invenio Publisher Transpiler-Mate Plugin.

Source code in src/invenio_publish/plugin.py

@transpiler_plugin(
    name="invenio-publish",
    description="Invenio Publisher Transpiler-Mate Plugin.",
    options_model=InvenioPublisherOptions,
)
def invenio_publish(
    context: TranspilerContext, options: InvenioPublisherOptions
) -> None:
    """Invenio Publisher Transpiler-Mate Plugin."""
    with InvenioClient(
        base_url=str(options.base_url), token=options.auth_token
    ) as invenio_rest_client:
        logger.debug("Setting up the HTTP logger...")
        from .utils import init_http_logging

        init_http_logging(invenio_rest_client.get_httpx_client())
        logger.debug("HTTP logger correctly setup")

        draft_id: str = ""

        if not context.metadata.identifier:
            logger.warning(
                "'identifier' key not found in source document, reserving a DOI..."
            )

            draft_record: Any | RDMRecord | ZenodoRecord | None = create_a_draft_record(
                client=invenio_rest_client, body=CreateADraftRecordBody()
            )

            if draft_record and isinstance(draft_record, (RDMRecord, ZenodoRecord)):
                draft_id = str(draft_record.id)

            logger.success(f"Successfully reserved a draft record with ID: {draft_id}")

            doi: Any | dict[str, Any] | None = reserve_a_doi(
                draft_id=draft_id, client=invenio_rest_client
            )

            if doi and isinstance(doi, dict):
                context.metadata.identifier = doi["doi"]
                context.metadata.same_as = doi["doi_url"]

                logger.success(
                    f"Successfully reserved a DOI with ID {context.metadata.identifier} (URL: {context.metadata.same_as})"
                )

                logger.warning(f"""Don't forget to update your source CWL Workflow with following metadata:
    s:identifier: {context.metadata.identifier}
    s:sameAs: {context.metadata.same_as}""")
        else:
            logger.info(
                f"Identifier {context.metadata.identifier} already assigned to {context.source}"
            )

            record_id: str = str(context.metadata.identifier).split(".")[-1]

            logger.info(
                f"Creating a new version for already existing Record {record_id}"
            )

            version: Any | RDMRecord | ZenodoRecord | None = create_a_new_version(
                record_id=record_id, client=invenio_rest_client
            )

            if (
                version
                and isinstance(version, (RDMRecord, ZenodoRecord))
                and version.id
            ):
                draft_id = str(version.id)

            logger.info(
                f"New version {draft_id} for already existing Record {record_id} created!"
            )

        invenio_metadata: Metadata = Metadata(
            identifiers=[
                AlternateIdentifier(
                    identifier=str(context.metadata.identifier),
                    scheme=IdentifierScheme.DOI,
                )
            ]
            if context.metadata.identifier
            else None,
            resource_type=ResourceType(id=ResourceTypeId.WORKFLOW),
            title=context.metadata.name,
            publication_date=date.fromtimestamp(time.time()).isoformat(),
            publisher=context.metadata.publisher.name,
            description=context.metadata.description
            if context.metadata.description
            else None,
            creators=list(
                map(
                    _to_creator,
                    context.metadata.author
                    if isinstance(context.metadata.author, list)
                    else [context.metadata.author],
                )
            ),
            contributors=list(
                map(
                    _to_contributor,
                    context.metadata.contributor
                    if isinstance(context.metadata.contributor, list)
                    else [context.metadata.contributor],
                )
            )
            if context.metadata.contributor
            else None,
            version=context.metadata.software_version,
        )

        _finalize(
            draft_id=draft_id,
            uploading_files=options.attach,
            session_client=invenio_rest_client,
            invenio_metadata=invenio_metadata,
        )

        logger.success(f"Record available on '{options.base_url}/records/{draft_id}'")

Options¶

Bases: BaseModel

Options accepted by the Invenio Publisher Transpiler-Mate Plugin plugin.

Source code in src/invenio_publish/plugin.py

class InvenioPublisherOptions(BaseModel):
    """Options accepted by the Invenio Publisher Transpiler-Mate Plugin plugin."""

    model_config = ConfigDict(extra="forbid")

    base_url: Annotated[AnyUrl, Field(description="The Invenio server base URL")]

    auth_token: Annotated[str, Field(description="The Invenio Access token")]

    attach: Annotated[
        list[Path],
        Field(
            default_factory=list,
            description="Generic textual/binary file(s) to attach to the Invenio record",
        ),
    ]

Client compatibility¶

The implementation uses invenio-rest-api-client's authenticated client and its draft, record-version, DOI, file-upload, metadata-update, and publication operations. Responses may be either RDMRecord or ZenodoRecord, allowing the same plugin flow to work with compatible InvenioRDM deployments and Zenodo.