wfexs_backend.pushers.nextcloud_export#

Module Contents#

Classes#

ExportMapping

NextcloudContentExporter

This code is a trimmed version of the one created for ES-FEGA (Spanish Federated EGA) infrastructure

NextcloudExportPlugin

Class to model exporting results to a Nextcloud instance

API#

class wfexs_backend.pushers.nextcloud_export.ExportMapping#

Bases: typing.NamedTuple

local_filename: wfexs_backend.common.AbsPath = None#
remote_dirname: wfexs_backend.common.RelPath = None#
remote_basename: wfexs_backend.common.RelPath = None#
class wfexs_backend.pushers.nextcloud_export.NextcloudContentExporter(nextcloud_url: wfexs_backend.common.URIType, nextcloud_user: str, nextcloud_token: str, nextcloud_base_directory: wfexs_backend.common.AbsPath, retention_tag_name: str | None = None)#

This code is a trimmed version of the one created for ES-FEGA (Spanish Federated EGA) infrastructure

Initialization

create_remote_path(reldir: RelPath | None = None, name: RelPath | None = None) Tuple[Any, AbsPath, RelPath]#
remove_remote_path(name: wfexs_backend.common.RelPath, reldir: RelPath | None = None) bool#
_chunked_uploader(fmapping: wfexs_backend.pushers.nextcloud_export.ExportMapping) extended_nc_client.extended_nc_client.DAVRequestResponse#
_chunked_file_batch_uploader(files_to_process: Sequence[ExportMapping]) Sequence[DAVRequestResponse]#
mappings_uploader(contents_to_process: Sequence[ExportMapping], destname: RelPath | None = None) Tuple[Sequence[DAVRequestResponse], AbsPath | None, RelPath | None]#
class wfexs_backend.pushers.nextcloud_export.NextcloudExportPlugin(refdir: pathlib.Path, setup_block: SecurityContextConfig | None = None, default_licences: Sequence[LicenceDescription] = [], default_orcids: Sequence[ResolvedORCID] = [], default_preferred_id: str | None = None)#

Bases: wfexs_backend.pushers.abstract_token_export.AbstractTokenExportPlugin

Class to model exporting results to a Nextcloud instance

Initialization

PLUGIN_NAME = 'cast(...)'#
ENABLED: ClassVar[bool] = True#
book_pid(preferred_id: str | None = None, initially_required_metadata: Mapping[str, Any] | None = None, initially_required_community_specific_metadata: Mapping[str, Any] | None = None, title: str | None = None, description: str | None = None, licences: Sequence[LicenceDescription] = [], resolved_orcids: Sequence[ResolvedORCID] = []) DraftEntry | None#
discard_booked_pid(pid_or_draft: str | DraftEntry) bool#

This method is used to release a previously booked PID, which has not been published.

When it returns False, it means that the provided id did exist, but it was not a draft

get_pid_metadata(pid: str) Mapping[str, Any] | None#
get_pid_draftentry(pid: str) DraftEntry | None#

This method is used to obtained the metadata associated to a PID, in case the destination allows it.

_prepare_upload_mappings(items: Sequence[AnyContent]) Sequence[ExportMapping]#
get_file_bucket_prefix(draft_entry: wfexs_backend.pushers.DraftEntry) str#

This is an accessory method which is used to build upload paths

upload_file_to_draft(draft_entry: wfexs_backend.pushers.DraftEntry, filename: str | IO[bytes], remote_filename: str | None, content_size: int | None = None) Mapping[str, Any]#
update_record_metadata(draft_entry: wfexs_backend.pushers.DraftEntry, metadata: Mapping[str, Any] | None = None, community_specific_metadata: Mapping[str, Any] | None = None, title: str | None = None, description: str | None = None, licences: Sequence[LicenceDescription] = [], resolved_orcids: Sequence[ResolvedORCID] = []) Mapping[str, Any]#
publish_draft_record(draft_entry: wfexs_backend.pushers.DraftEntry) Mapping[str, Any]#