Source code for pyPreservica.uploadAPI
"""
pyPreservica UploadAPI module definition
A client library for the Preservica Repository Upload API
author: James Carr
licence: Apache License 2.0
"""
import csv
import shutil
import tempfile
import uuid
import xml
from datetime import datetime, timedelta, timezone
from time import sleep
from typing import Callable
from xml.dom import minidom
from xml.etree import ElementTree
from xml.etree.ElementTree import Element, SubElement
import boto3
import s3transfer.tasks
import s3transfer.upload
from botocore.session import get_session
from boto3.s3.transfer import TransferConfig, S3Transfer
from botocore.config import Config
from botocore.credentials import RefreshableCredentials
from botocore.exceptions import ClientError, NoCredentialsError, PartialCredentialsError
from dateutil.tz import tzlocal
from pyrate_limiter import Rate, Duration, Limiter, InMemoryBucket
from s3transfer import S3UploadFailedError
from tqdm import tqdm
from pyPreservica.common import *
from pyPreservica.common import _make_stored_zipfile
logger = logging.getLogger(__name__)
MB = 1024 * 1024
GB = 1024 ** 3
transfer_config = TransferConfig(multipart_threshold=int(32 * MB))
CONTENT_FOLDER = "content"
PRESERVATION_CONTENT_FOLDER = "p1"
ACCESS_CONTENT_FOLDER = "a1"
def upload_file(self, filename, bucket, key, callback=None, extra_args=None):
"""Upload a file to an S3 object.
Variants have also been injected into S3 client, Bucket and Object.
You don't have to use S3Transfer.upload_file() directly.
.. seealso::
:py:meth:`S3.Client.upload_file`
:py:meth:`S3.Client.upload_fileobj`
"""
if not isinstance(filename, str):
raise ValueError('Filename must be a string')
subscribers = self._get_subscribers(callback)
future = self._manager.upload(filename, bucket, key, extra_args, subscribers)
try:
return future.result()
# If a client error was raised, add the backwards compatibility layer
# that raises a S3UploadFailedError. These specific errors were only
# ever thrown for upload_parts but now can be thrown for any related
# client error.
except ClientError as e:
raise S3UploadFailedError("Failed to upload %s to %s: %s" % (filename, '/'.join([bucket, key]), e))
class PutObjectTask(s3transfer.tasks.Task):
# Copied from s3transfer/upload.py, changed to return the result of client.put_object.
def _main(self, client, fileobj, bucket, key, extra_args):
with fileobj as body:
response = client.put_object(Bucket=bucket, Key=key, Body=body, **extra_args)
return response
class CompleteMultipartUploadTask(s3transfer.tasks.Task):
# Copied from s3transfer/tasks.py, changed to return a result.
def _main(self, client, bucket, key, upload_id, parts, extra_args):
return client.complete_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id,
MultipartUpload={"Parts": parts},
**extra_args, )
s3transfer.upload.PutObjectTask = PutObjectTask
s3transfer.upload.CompleteMultipartUploadTask = CompleteMultipartUploadTask
def prettify(elem):
"""Return a pretty-printed XML string for the Element.
"""
rough_string = xml.etree.ElementTree.tostring(elem, 'utf-8')
re_parsed = minidom.parseString(rough_string)
return re_parsed.toprettyxml(indent=" ")
def __create_io__(xip=None, file_name=None, parent_folder=None, **kwargs):
if xip is None:
xip = Element('xip:XIP')
xip.set('xmlns:xip', 'http://preservica.com/XIP/v6.0')
assert xip is not None
io = SubElement(xip, 'xip:InformationObject')
ref = SubElement(io, 'xip:Ref')
if 'IO_Identifier_callback' in kwargs:
ident_callback = kwargs.get('IO_Identifier_callback')
ref.text = ident_callback()
else:
ref.text = str(uuid.uuid4())
title = SubElement(io, 'xip:Title')
title.text = kwargs.get('Title', file_name)
description = SubElement(io, 'xip:Description')
description.text = kwargs.get('Description', file_name)
security = SubElement(io, 'xip:SecurityTag')
security.text = kwargs.get('SecurityTag', "open")
custom_type = SubElement(io, 'xip:CustomType')
custom_type.text = kwargs.get('CustomType', "")
parent = SubElement(io, 'xip:Parent')
if hasattr(parent_folder, "reference"):
parent.text = parent_folder.reference
elif isinstance(parent_folder, str):
parent.text = parent_folder
return xip, ref.text
def __make_representation__(xip, rep_name, rep_type, io_ref):
representation = SubElement(xip, 'xip:Representation')
io_link = SubElement(representation, 'xip:InformationObject')
io_link.text = io_ref
access_name = SubElement(representation, 'xip:Name')
access_name.text = rep_name
access_type = SubElement(representation, 'xip:Type')
access_type.text = rep_type
content_objects = SubElement(representation, 'xip:ContentObjects')
content_object = SubElement(content_objects, 'xip:ContentObject')
content_object_ref = str(uuid.uuid4())
content_object.text = content_object_ref
return content_object_ref
def __make_content_objects__(xip, content_title, co_ref, io_ref, tag, content_description, content_type):
content_object = SubElement(xip, 'xip:ContentObject')
ref_element = SubElement(content_object, "xip:Ref")
ref_element.text = co_ref
title = SubElement(content_object, "xip:Title")
title.text = content_title
description = SubElement(content_object, "xip:Description")
description.text = content_description
security_tag = SubElement(content_object, "xip:SecurityTag")
security_tag.text = tag
custom_type = SubElement(content_object, "xip:CustomType")
custom_type.text = content_type
parent = SubElement(content_object, "xip:Parent")
parent.text = io_ref
def __make_generation__(xip, filename, co_ref, generation_label, location=None):
generation = SubElement(xip, 'xip:Generation', {"original": "true", "active": "true"})
content_object = SubElement(generation, "xip:ContentObject")
content_object.text = co_ref
label = SubElement(generation, "xip:Label")
if generation_label:
label.text = generation_label
else:
label.text = os.path.splitext(filename)[0]
effective_date = SubElement(generation, "xip:EffectiveDate")
effective_date.text = datetime.now().isoformat()
bitstreams = SubElement(generation, "xip:Bitstreams")
bitstream = SubElement(bitstreams, "xip:Bitstream")
bitstream.text = f"{location}/{filename}"
SubElement(generation, "xip:Formats")
SubElement(generation, "xip:Properties")
def __make_bitstream__(xip, file_name, full_path, callback, location=None):
bitstream = SubElement(xip, 'xip:Bitstream')
filename_element = SubElement(bitstream, "xip:Filename")
filename_element.text = file_name
filesize = SubElement(bitstream, "xip:FileSize")
file_stats = os.stat(full_path)
filesize.text = str(file_stats.st_size)
physical_location = SubElement(bitstream, "xip:PhysicalLocation")
physical_location.text = location
fixities = SubElement(bitstream, "xip:Fixities")
fixity_result = callback(file_name, full_path)
if type(fixity_result) == tuple:
fixity = SubElement(fixities, "xip:Fixity")
fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef")
fixity_value = SubElement(fixity, "xip:FixityValue")
fixity_algorithm_ref.text = fixity_result[0]
fixity_value.text = fixity_result[1]
elif type(fixity_result) == dict:
for key, val in fixity_result.items():
fixity = SubElement(fixities, "xip:Fixity")
fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef")
fixity_value = SubElement(fixity, "xip:FixityValue")
fixity_algorithm_ref.text = key
fixity_value.text = val
else:
logger.error("Could Not Find Fixity Value")
raise RuntimeError("Could Not Find Fixity Value")
def __make_representation_multiple_co__(xip, rep_name, rep_type, rep_files, io_ref):
representation = SubElement(xip, 'xip:Representation')
io_link = SubElement(representation, 'xip:InformationObject')
io_link.text = io_ref
access_name = SubElement(representation, 'xip:Name')
access_name.text = rep_name
access_type = SubElement(representation, 'xip:Type')
access_type.text = rep_type
content_objects = SubElement(representation, 'xip:ContentObjects')
refs_dict = {}
for f in rep_files:
content_object = SubElement(content_objects, 'xip:ContentObject')
content_object_ref = str(uuid.uuid4())
content_object.text = content_object_ref
refs_dict[content_object_ref] = f
return refs_dict
def cvs_to_cmis_xslt(csv_file, xml_namespace, root_element, title="Metadata Title", export_folder=None,
additional_namespaces=None):
"""
Deprecated function, use csv_to_cmis_xslt instead.
"""
return csv_to_cmis_xslt(csv_file, xml_namespace, root_element, title, export_folder, additional_namespaces)
[docs]
def csv_to_cmis_xslt(csv_file, xml_namespace, root_element, title="Metadata Title", export_folder=None,
additional_namespaces=None):
"""
Create a custom CMIS XSLT transform to display metadata within Universal Access (UA).
Reads the column headers from the first row of ``csv_file`` and generates an XSL
stylesheet that maps the XML elements produced by :func:`csv_to_xml` to the Tessella
CMIS metadata display format used by Preservica Universal Access.
:param csv_file: Path to the CSV file whose headers define the metadata fields.
:type csv_file: str
:param xml_namespace: The XML namespace URI used in the metadata documents.
:type xml_namespace: str
:param root_element: The root element name used in the metadata documents.
:type root_element: str
:param title: Human-readable title shown in UA for this metadata group.
Defaults to ``"Metadata Title"``.
:type title: str
:param export_folder: Directory where the generated ``.xslt`` file is written.
If ``None``, the file is written to the current working directory.
:type export_folder: str or None
:param additional_namespaces: Mapping of namespace prefix to URI for any extra
namespaces referenced in the CSV headers (e.g. ``{"dc": "http://..."}``.
:type additional_namespaces: dict or None
:returns: Path to the generated XSLT file.
:rtype: str
"""
headers = set()
with open(csv_file, encoding='utf-8-sig', newline='') as csvfile:
reader = csv.reader(csvfile)
for row in reader:
for header in row:
xml_tag = header.strip()
xml_tag = xml_tag.replace(" ", "")
xml_tag = xml_tag.replace("-", "")
headers.add(xml_tag)
break
namespaces = {"version": "2.0", "xmlns:xsl": "http://www.w3.org/1999/XSL/Transform",
"xmlns:fn": "http://www.w3.org/2005/xpath-functions", "xmlns:xs": "http://www.w3.org/2001/XMLSchema",
"xmlns:csv": xml_namespace, "xmlns": "http://www.tessella.com/sdb/cmis/metadata",
"exclude-result-prefixes": "csv"}
if additional_namespaces is not None:
for prefix, uri in additional_namespaces.items():
namespaces["xmlns:" + prefix] = uri
xml_stylesheet = xml.etree.ElementTree.Element("xsl:stylesheet", namespaces)
xml.etree.ElementTree.SubElement(xml_stylesheet, "xsl:output", {"method": "xml", "indent": "yes"})
xml_template = xml.etree.ElementTree.SubElement(xml_stylesheet, "xsl:template", {"match": "csv:" + root_element})
xml_group = xml.etree.ElementTree.SubElement(xml_template, "group")
xml_title = xml.etree.ElementTree.SubElement(xml_group, "title")
xml_title.text = title
xml_templates = xml.etree.ElementTree.SubElement(xml_group, "xsl:apply-templates")
elements = ""
for header in headers:
if ":" in header:
elements = elements + "|" + header
else:
elements = elements + "|csv:" + header
elements = elements[1:]
xml_matches = xml.etree.ElementTree.SubElement(xml_stylesheet, "xsl:template", {"match": elements})
xml_item = xml.etree.ElementTree.SubElement(xml_matches, "item")
xml_name = xml.etree.ElementTree.SubElement(xml_item, "name")
xml_name_value = xml.etree.ElementTree.SubElement(xml_name, "xsl:value-of", {
"select": "fn:replace(translate(local-name(), '_', ' '), '([a-z])([A-Z])', '$1 $2')"})
xml_value = xml.etree.ElementTree.SubElement(xml_item, "value")
xml_value_value = xml.etree.ElementTree.SubElement(xml_value, "xsl:value-of", {"select": "."})
xml_type = xml.etree.ElementTree.SubElement(xml_item, "type")
xml_type_value = xml.etree.ElementTree.SubElement(xml_type, "xsl:value-of", {
"select": "fn:replace(translate(local-name(), '_', ' '), '([a-z])([A-Z])', '$1 $2')"})
xml_request = xml.etree.ElementTree.tostring(xml_stylesheet, encoding='utf-8', xml_declaration=True)
cmis_xslt = root_element + "-cmis.xslt"
if export_folder is not None:
cmis_xslt = os.path.join(export_folder, cmis_xslt)
file = open(cmis_xslt, "wt", encoding="utf-8")
file.write(xml_request.decode("utf-8"))
file.close()
return cmis_xslt
def cvs_to_xsd(csv_file, xml_namespace, root_element, export_folder=None, additional_namespaces=None):
"""
Deprecated function, use csv_to_xsd instead.
"""
return csv_to_xsd(csv_file, xml_namespace, root_element, export_folder, additional_namespaces)
[docs]
def csv_to_xsd(csv_file, xml_namespace, root_element, export_folder=None, additional_namespaces=None):
"""
Create an XSD schema definition derived from the column headers of a CSV file.
Reads the column headers from the first row of ``csv_file`` and generates an XML
Schema Definition (``.xsd``) file that validates the XML documents produced by
:func:`csv_to_xml`.
:param csv_file: Path to the CSV file whose headers define the element names.
:type csv_file: str
:param xml_namespace: The target namespace URI for the generated schema.
:type xml_namespace: str
:param root_element: The root element name that the schema will validate.
:type root_element: str
:param export_folder: Directory where the generated ``.xsd`` file is written.
If ``None``, the file is written to the current working directory.
:type export_folder: str or None
:param additional_namespaces: Mapping of namespace prefix to URI for any extra
namespaces referenced in the CSV headers.
:type additional_namespaces: dict or None
:returns: Path to the generated XSD file.
:rtype: str
"""
headers = set()
with open(csv_file, encoding='utf-8-sig', newline='') as csvfile:
reader = csv.reader(csvfile)
for row in reader:
for header in row:
xml_tag = header.strip()
xml_tag = xml_tag.replace(" ", "")
xml_tag = xml_tag.replace("-", "")
headers.add(xml_tag)
break
namespaces = {"xmlns:xs": "http://www.w3.org/2001/XMLSchema", "attributeFormDefault": "unqualified",
"elementFormDefault": "qualified",
"targetNamespace": xml_namespace}
if additional_namespaces is not None:
for prefix, uri in additional_namespaces.items():
namespaces["xmlns:" + prefix.strip()] = uri.strip()
xml_schema = xml.etree.ElementTree.Element("xs:schema", namespaces)
if additional_namespaces is not None:
for prefix, namespace in additional_namespaces.items():
xml_import = xml.etree.ElementTree.SubElement(xml_schema, "xs:import", {"namespace": namespace})
xml_element = xml.etree.ElementTree.SubElement(xml_schema, "xs:element", {"name": root_element})
xml_complex_type = xml.etree.ElementTree.SubElement(xml_element, "xs:complexType")
xml_sequence = xml.etree.ElementTree.SubElement(xml_complex_type, "xs:sequence")
for header in headers:
if ":" in header:
prefix, sep, tag = header.partition(":")
try:
namespace = additional_namespaces[prefix]
xml.etree.ElementTree.SubElement(xml_sequence, "xs:element",
{"ref": header, "xmlns:" + prefix: namespace})
except KeyError:
xml.etree.ElementTree.SubElement(xml_sequence, "xs:element", {"type": "xs:string", "name": header})
else:
xml.etree.ElementTree.SubElement(xml_sequence, "xs:element", {"type": "xs:string", "name": header})
xml_request = xml.etree.ElementTree.tostring(xml_schema, encoding='utf-8', xml_declaration=True)
xsd_file = root_element + ".xsd"
if export_folder is not None:
xsd_file = os.path.join(export_folder, xsd_file)
file = open(xsd_file, "wt", encoding="utf-8")
file.write(xml_request.decode("utf-8"))
file.close()
return xsd_file
def csv_to_search_xml(csv_file, xml_namespace, root_element, title="Metadata Title", export_folder=None,
additional_namespaces=None):
"""
Create a custom Preservica search index configuration derived from a CSV file.
Reads the column headers from the first row of ``csv_file`` and generates a Preservica
custom search index XML file (``*-index.xml``) that registers each column as a
searchable ``STRING_DEFAULT`` field.
:param csv_file: Path to the CSV file whose headers define the indexed fields.
:type csv_file: str
:param xml_namespace: The XML namespace URI of the metadata documents to be indexed.
:type xml_namespace: str
:param root_element: The root element name used in the metadata documents.
:type root_element: str
:param title: Human-readable name for the search schema group. Defaults to
``"Metadata Title"``.
:type title: str
:param export_folder: Directory where the generated index XML file is written.
If ``None``, the file is written to the current working directory.
:type export_folder: str or None
:param additional_namespaces: Mapping of namespace prefix to URI for any extra
namespaces referenced in the CSV headers.
:type additional_namespaces: dict or None
:returns: Path to the generated search index XML file.
:rtype: str
"""
headers = set()
with open(csv_file, encoding='utf-8-sig', newline='') as csvfile:
reader = csv.reader(csvfile)
for row in reader:
for header in row:
xml_tag = header.strip()
xml_tag = xml_tag.replace(" ", "")
xml_tag = xml_tag.replace("-", "")
headers.add(xml_tag)
break
xml_index = xml.etree.ElementTree.Element("index", {"xmlns": "http://www.preservica.com/customindex/v1"})
short_name = "csv"
xml_schema_name = xml.etree.ElementTree.SubElement(xml_index, "schemaName")
xml_schema_name.text = title
xml_schema_uri = xml.etree.ElementTree.SubElement(xml_index, "schemaUri")
xml_schema_uri.text = xml_namespace
xml_short_name = xml.etree.ElementTree.SubElement(xml_index, "shortName")
xml_short_name.text = short_name
for header in headers:
if ":" in header:
xpath_expression = f"//{short_name}:{root_element}/{header}"
else:
xpath_expression = f"//{short_name}:{root_element}/{short_name}:{header}"
attr = {"indexName": header, "displayName": header, "xpath": xpath_expression, "indexType": "STRING_DEFAULT"}
xml_term = xml.etree.ElementTree.SubElement(xml_index, "term", attr)
if additional_namespaces is not None:
for prefix, uri in additional_namespaces.items():
xml.etree.ElementTree.SubElement(xml_index, "namespaceMapping", {"key": prefix, "value": uri})
xml_request = xml.etree.ElementTree.tostring(xml_index, encoding='utf-8', xml_declaration=True)
search_xml = root_element + "-index.xml"
if export_folder is not None:
search_xml = os.path.join(export_folder, search_xml)
file = open(search_xml, "wt", encoding="utf-8")
file.write(xml_request.decode("utf-8"))
file.close()
return search_xml
def cvs_to_xml(csv_file, xml_namespace, root_element, file_name_column="filename", export_folder=None,
additional_namespaces=None):
"""
Deprecated function, use csv_to_xml instead.
"""
return csv_to_xml(csv_file, xml_namespace, root_element, file_name_column, export_folder, additional_namespaces)
[docs]
def csv_to_xml(csv_file, xml_namespace, root_element, file_name_column="filename", export_folder=None,
additional_namespaces=None):
"""
Export the rows of a CSV file as individual XML metadata documents for Preservica assets.
Each data row in ``csv_file`` is converted to an XML document whose root element is
``root_element`` and whose child elements correspond to the CSV column headers. One
XML file is written per data row; the file is named after the value in the column
identified by ``file_name_column``.
:param csv_file: Path to the CSV file to process.
:type csv_file: str
:param xml_namespace: The XML namespace URI applied to the generated documents.
:type xml_namespace: str
:param root_element: The root element name for each generated XML document.
:type root_element: str
:param file_name_column: The CSV column header whose value is used to name each
output XML file. Defaults to ``"filename"``.
:type file_name_column: str
:param export_folder: Directory where the generated XML files are written.
If ``None``, files are written to the current working directory.
:type export_folder: str or None
:param additional_namespaces: Mapping of namespace prefix to URI for any extra
namespaces to declare on the root element.
:type additional_namespaces: dict or None
:returns: Generator that yields the path to each written XML file.
:rtype: Generator[str, None, None]
"""
headers = list()
link_column_id = 0
with open(csv_file, encoding='utf-8-sig', newline='') as csvfile:
reader = csv.reader(csvfile)
for row in reader:
col_id = 0
for header in row:
col_id += 1
if header == file_name_column:
link_column_id = col_id
xml_tag = header.strip()
xml_tag = xml_tag.replace(" ", "")
xml_tag = xml_tag.replace("-", "")
headers.append(xml_tag)
break
if link_column_id > 0:
namespaces = {"xmlns": xml_namespace}
if additional_namespaces is not None:
for prefix, uri in additional_namespaces.items():
namespaces["xmlns:" + prefix] = uri
for row in reader:
col_id = 0
xml_object = xml.etree.ElementTree.Element(root_element, namespaces)
for value, header in zip(row, headers):
col_id += 1
xml.etree.ElementTree.SubElement(xml_object, header).text = value
if col_id == link_column_id:
file_name = value
xml_request = xml.etree.ElementTree.tostring(xml_object, encoding='utf-8', xml_declaration=True)
name = file_name + ".xml"
name = sanitize(name)
if export_folder is not None:
name = os.path.join(export_folder, name)
file = open(name, "wt", encoding="utf-8")
file.write(xml_request.decode("utf-8"))
file.close()
yield name
def generic_asset_package(preservation_files_dict: dict|None=None, access_files_dict: dict|None=None, export_folder=None, parent_folder=None,
compress=True,
**kwargs):
"""
Create a Preservica ingest package containing an asset with an arbitrary number of representations.
Both ``preservation_files_dict`` and ``access_files_dict`` map representation names to
lists of file paths. Each key becomes a named representation; each file path in the
associated list becomes a separate content object within that representation.
:param preservation_files_dict: Mapping of preservation representation name to a list
of file paths. Pass ``None`` to omit preservation representations.
:type preservation_files_dict: dict[str, list[str]] or None
:param access_files_dict: Mapping of access representation name to a list of file
paths. Pass ``None`` to omit access representations.
:type access_files_dict: dict[str, list[str]] or None
:param export_folder: Directory where the package ZIP file is written. Defaults to
the system temporary directory when ``None``.
:type export_folder: str or None
:param parent_folder: The Preservica folder (or its UUID string) under which the asset
will be ingested. Required.
:type parent_folder: Folder or str
:param compress: Whether to compress the package ZIP file. Defaults to ``True``.
:type compress: bool
:param kwargs: Optional keyword arguments:
- ``Title`` *(str)* – Asset title.
- ``Description`` *(str)* – Asset description.
- ``SecurityTag`` *(str)* – Asset security tag (default ``"open"``).
- ``CustomType`` *(str)* – Asset custom type.
- ``Preservation_Content_Title`` *(str or dict)* – Content object title for
preservation representations.
- ``Preservation_Content_Description`` *(str or dict)* – Content object description
for preservation representations.
- ``Access_Content_Title`` *(str or dict)* – Content object title for access
representations.
- ``Access_Content_Description`` *(str or dict)* – Content object description for
access representations.
- ``Preservation_Generation_Label`` *(str)* – Generation label for preservation
representations.
- ``Access_Generation_Label`` *(str)* – Generation label for access representations.
- ``Preservation_files_fixity_callback`` *(callable)* – Fixity callback for
preservation files.
- ``Access_files_fixity_callback`` *(callable)* – Fixity callback for access files.
- ``IO_Identifier_callback`` *(callable)* – Callback to supply the asset UUID.
- ``Identifiers`` *(dict)* – Mapping of identifier type to identifier value.
- ``Asset_Metadata`` *(dict)* – Mapping of schema URI to metadata file path or
XML string.
:returns: Path to the generated package ZIP file.
:rtype: str
:raises RuntimeError: If ``export_folder`` does not exist or ``parent_folder`` is
``None``.
"""
# some basic validation
if export_folder is None:
export_folder = tempfile.gettempdir()
if not os.path.isdir(export_folder):
logger.error("Export Folder Does Not Exist")
raise RuntimeError(export_folder, "Export Folder Does Not Exist")
if parent_folder is None:
logger.error("You must specify a parent folder for the package asset")
raise RuntimeError("You must specify a parent folder for the package asset")
io_ref = None
xip = None
default_asset_title = None
preservation_representation_refs_dict = {}
access_representation_refs_dict = {}
security_tag = kwargs.get('SecurityTag', "open")
content_type = kwargs.get('CustomType', "")
if not compress:
shutil.register_archive_format(name="szip", function=_make_stored_zipfile, extra_args=None, description="UnCompressed ZIP file")
has_preservation_files = bool((preservation_files_dict is not None) and (len(preservation_files_dict) > 0))
has_access_files = bool((access_files_dict is not None) and (len(access_files_dict) > 0))
if has_preservation_files:
if default_asset_title is None:
key = list(preservation_files_dict.keys())[0]
preservation_files_list = preservation_files_dict[key]
default_asset_title = os.path.splitext(os.path.basename(preservation_files_list[0]))[0]
# create the asset
xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs)
if has_access_files:
if default_asset_title is None:
key = list(access_files_dict.keys())[0]
access_files_list = access_files_dict[key]
default_asset_title = os.path.splitext(os.path.basename(access_files_list[0]))[0]
if io_ref is None:
xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs)
# loop over preservation_files_map
if has_preservation_files:
for representation_name in preservation_files_dict.keys():
preservation_files_list = preservation_files_dict[representation_name]
preservation_refs_dict = __make_representation_multiple_co__(xip, rep_name=representation_name,
rep_type="Preservation",
rep_files=preservation_files_list,
io_ref=io_ref)
preservation_representation_refs_dict[representation_name] = preservation_refs_dict
if has_access_files:
for representation_name in access_files_dict.keys():
access_files_list = access_files_dict[representation_name]
access_refs_dict = __make_representation_multiple_co__(xip, rep_name=representation_name, rep_type="Access",
rep_files=access_files_list, io_ref=io_ref)
access_representation_refs_dict[representation_name] = access_refs_dict
if has_preservation_files:
for representation_name in preservation_representation_refs_dict.keys():
preservation_refs_dict = preservation_representation_refs_dict[representation_name]
for content_ref, filename in preservation_refs_dict.items():
default_content_objects_title = os.path.splitext(os.path.basename(filename))[0]
preservation_content_title = kwargs.get('Preservation_Content_Title', default_content_objects_title)
preservation_content_description = kwargs.get('Preservation_Content_Description',
default_content_objects_title)
if isinstance(preservation_content_title, dict):
preservation_content_title = preservation_content_title.get("filename",
default_content_objects_title)
if isinstance(preservation_content_description, dict):
preservation_content_description = preservation_content_description.get("filename",
default_content_objects_title)
__make_content_objects__(xip, preservation_content_title, content_ref, io_ref, security_tag,
preservation_content_description, content_type)
if has_access_files:
for representation_name in access_representation_refs_dict.keys():
access_refs_dict = access_representation_refs_dict[representation_name]
for content_ref, filename in access_refs_dict.items():
default_content_objects_title = os.path.splitext(os.path.basename(filename))[0]
access_content_title = kwargs.get('Access_Content_Title', default_content_objects_title)
access_content_description = kwargs.get('Access_Content_Description', default_content_objects_title)
if isinstance(access_content_title, dict):
access_content_title = access_content_title.get("filename", default_content_objects_title)
if isinstance(access_content_description, dict):
access_content_description = access_content_title.get("filename", default_content_objects_title)
__make_content_objects__(xip, access_content_title, content_ref, io_ref, security_tag,
access_content_description,
content_type)
if has_preservation_files:
for representation_name in preservation_representation_refs_dict.keys():
location = sanitize(representation_name)
preservation_refs_dict = preservation_representation_refs_dict[representation_name]
preservation_generation_label = kwargs.get('Preservation_Generation_Label', "")
for content_ref, filename in preservation_refs_dict.items():
preservation_file_name = os.path.basename(filename)
__make_generation__(xip, preservation_file_name, content_ref, preservation_generation_label, location)
if has_access_files:
for representation_name in access_representation_refs_dict.keys():
location = sanitize(representation_name)
access_refs_dict = access_representation_refs_dict[representation_name]
access_generation_label = kwargs.get('Access_Generation_Label', "")
for content_ref, filename in access_refs_dict.items():
access_file_name = os.path.basename(filename)
__make_generation__(xip, access_file_name, content_ref, access_generation_label, location)
if has_preservation_files:
if 'Preservation_files_fixity_callback' in kwargs:
callback = kwargs.get('Preservation_files_fixity_callback')
else:
callback = Sha1FixityCallBack()
for representation_name in preservation_representation_refs_dict.keys():
location = sanitize(representation_name)
preservation_refs_dict = preservation_representation_refs_dict[representation_name]
for content_ref, filename in preservation_refs_dict.items():
preservation_file_name = os.path.basename(filename)
__make_bitstream__(xip, preservation_file_name, filename, callback, location)
if has_access_files:
if 'Access_files_fixity_callback' in kwargs:
callback = kwargs.get('Access_files_fixity_callback')
else:
callback = Sha1FixityCallBack()
for representation_name in access_representation_refs_dict.keys():
location = sanitize(representation_name)
access_refs_dict = access_representation_refs_dict[representation_name]
for content_ref, filename in access_refs_dict.items():
access_file_name = os.path.basename(filename)
__make_bitstream__(xip, access_file_name, filename, callback, location)
if 'Identifiers' in kwargs:
identifier_map = kwargs.get('Identifiers')
for identifier_key, identifier_value in identifier_map.items():
if identifier_key:
if identifier_value:
identifier = SubElement(xip, 'xip:Identifier')
id_type = SubElement(identifier, "xip:Type")
id_type.text = identifier_key
id_value = SubElement(identifier, "xip:Value")
id_value.text = identifier_value
id_io = SubElement(identifier, "xip:Entity")
id_io.text = io_ref
if 'Asset_Metadata' in kwargs:
metadata_map = kwargs.get('Asset_Metadata')
for metadata_ns, metadata_path in metadata_map.items():
if metadata_ns:
if metadata_path:
if os.path.exists(metadata_path) and os.path.isfile(metadata_path):
descriptive_metadata = xml.etree.ElementTree.parse(source=metadata_path)
metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns})
metadata_ref = SubElement(metadata, 'xip:Ref')
metadata_ref.text = str(uuid.uuid4())
entity = SubElement(metadata, 'xip:Entity')
entity.text = io_ref
content = SubElement(metadata, 'xip:Content')
content.append(descriptive_metadata.getroot())
elif isinstance(metadata_path, str):
try:
descriptive_metadata = xml.etree.ElementTree.fromstring(metadata_path)
metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns})
metadata_ref = SubElement(metadata, 'xip:Ref')
metadata_ref.text = str(uuid.uuid4())
entity = SubElement(metadata, 'xip:Entity')
entity.text = io_ref
content = SubElement(metadata, 'xip:Content')
content.append(descriptive_metadata)
except RuntimeError:
logging.info(f"Could not parse asset metadata in namespace {metadata_ns}")
if xip is not None:
export_folder = export_folder
top_level_folder = os.path.join(export_folder, io_ref)
os.mkdir(top_level_folder)
inner_folder = os.path.join(top_level_folder, io_ref)
os.mkdir(inner_folder)
content_folder = os.path.join(inner_folder, CONTENT_FOLDER)
os.mkdir(content_folder)
metadata_path = os.path.join(inner_folder, "metadata.xml")
metadata = open(metadata_path, "wt", encoding='utf-8')
metadata.write(prettify(xip))
metadata.close()
for representation_name in preservation_representation_refs_dict.keys():
location = sanitize(representation_name)
Path(os.path.join(content_folder, location)).mkdir(parents=True, exist_ok=True)
preservation_refs_dict = preservation_representation_refs_dict[representation_name]
for content_ref, filename in preservation_refs_dict.items():
src_file = filename
dst_file = os.path.join(os.path.join(content_folder, location), os.path.basename(filename))
shutil.copyfile(src_file, dst_file)
for representation_name in access_representation_refs_dict.keys():
location = sanitize(representation_name)
Path(os.path.join(content_folder, location)).mkdir(parents=True, exist_ok=True)
access_refs_dict = access_representation_refs_dict[representation_name]
for content_ref, filename in access_refs_dict.items():
src_file = filename
dst_file = os.path.join(os.path.join(content_folder, location), os.path.basename(filename))
shutil.copyfile(src_file, dst_file)
if compress:
shutil.make_archive(top_level_folder, 'zip', top_level_folder)
else:
shutil.make_archive(top_level_folder, 'szip', top_level_folder)
shutil.rmtree(top_level_folder)
return top_level_folder + ".zip"
def multi_asset_package(asset_file_list:list|None=None, export_folder=None, parent_folder=None, compress=True, **kwargs):
"""
Create a single ingest package containing multiple assets, one per file.
Each file in ``asset_file_list`` becomes a separate asset (Information Object) with a
single preservation representation. All assets share the same ``parent_folder``.
:param asset_file_list: List of file paths; one asset is created per file.
:type asset_file_list: list[str] or None
:param export_folder: Directory where the package ZIP file is written. Defaults to
the system temporary directory when ``None``.
:type export_folder: str or None
:param parent_folder: The Preservica folder (or its UUID string) under which all
assets will be ingested. Required.
:type parent_folder: Folder or str
:param compress: Whether to compress the package ZIP file. Defaults to ``True``.
:type compress: bool
:param kwargs: Optional keyword arguments:
- ``SecurityTag`` *(str)* – Security tag applied to all assets (default ``"open"``).
- ``CustomType`` *(str)* – Custom type applied to all assets.
- ``Preservation_files_fixity_callback`` *(callable)* – Fixity callback used when
computing checksums for the content files.
- ``Identifiers`` *(dict)* – Mapping of file path to a dict of identifier type/value
pairs that should be attached to the corresponding asset.
:returns: Path to the generated package ZIP file.
:rtype: str
:raises RuntimeError: If ``export_folder`` does not exist or ``parent_folder`` is
``None``.
"""
# some basic validation
if export_folder is None:
export_folder = tempfile.gettempdir()
if not os.path.isdir(export_folder):
logger.error("Export Folder Does Not Exist")
raise RuntimeError(export_folder, "Export Folder Does Not Exist")
if parent_folder is None:
logger.error("You must specify a parent folder for the package asset")
raise RuntimeError("You must specify a parent folder for the package asset")
security_tag = kwargs.get('SecurityTag', "open")
content_type = kwargs.get('CustomType', "")
if not compress:
shutil.register_archive_format("szip", _make_stored_zipfile, None, "UnCompressed ZIP file")
if 'Preservation_files_fixity_callback' in kwargs:
fixity_callback = kwargs.get('Preservation_files_fixity_callback')
else:
fixity_callback = Sha1FixityCallBack()
package_id = str(uuid.uuid4())
export_folder = export_folder
top_level_folder = os.path.join(export_folder, package_id)
os.mkdir(top_level_folder)
inner_folder = os.path.join(top_level_folder, package_id)
os.mkdir(inner_folder)
os.mkdir(os.path.join(inner_folder, CONTENT_FOLDER))
asset_map = dict()
xip = Element('xip:XIP')
xip.set('xmlns:xip', 'http://preservica.com/XIP/v6.0')
for file in asset_file_list:
default_asset_title = os.path.splitext(os.path.basename(file))[0]
xip, io_ref = __create_io__(xip, file_name=default_asset_title, parent_folder=parent_folder, **kwargs)
asset_map[file] = io_ref
representation = SubElement(xip, 'xip:Representation')
io_link = SubElement(representation, 'xip:InformationObject')
io_link.text = io_ref
access_name = SubElement(representation, 'xip:Name')
access_name.text = "Preservation"
access_type = SubElement(representation, 'xip:Type')
access_type.text = "Preservation"
content_objects = SubElement(representation, 'xip:ContentObjects')
content_object = SubElement(content_objects, 'xip:ContentObject')
content_object_ref = str(uuid.uuid4())
content_object.text = content_object_ref
default_content_objects_title = os.path.splitext(os.path.basename(file))[0]
content_object = SubElement(xip, 'xip:ContentObject')
ref_element = SubElement(content_object, "xip:Ref")
ref_element.text = content_object_ref
title = SubElement(content_object, "xip:Title")
title.text = default_content_objects_title
description = SubElement(content_object, "xip:Description")
description.text = default_content_objects_title
security_tag_element = SubElement(content_object, "xip:SecurityTag")
security_tag_element.text = security_tag
custom_type = SubElement(content_object, "xip:CustomType")
custom_type.text = content_type
parent = SubElement(content_object, "xip:Parent")
parent.text = io_ref
generation = SubElement(xip, 'xip:Generation', {"original": "true", "active": "true"})
content_object = SubElement(generation, "xip:ContentObject")
content_object.text = content_object_ref
label = SubElement(generation, "xip:Label")
label.text = os.path.splitext(os.path.basename(file))[0]
effective_date = SubElement(generation, "xip:EffectiveDate")
effective_date.text = datetime.now().isoformat()
bitstreams = SubElement(generation, "xip:Bitstreams")
bitstream = SubElement(bitstreams, "xip:Bitstream")
bitstream.text = os.path.basename(file)
SubElement(generation, "xip:Formats")
SubElement(generation, "xip:Properties")
bitstream = SubElement(xip, 'xip:Bitstream')
filename_element = SubElement(bitstream, "xip:Filename")
filename_element.text = os.path.basename(file)
filesize = SubElement(bitstream, "xip:FileSize")
file_stats = os.stat(file)
filesize.text = str(file_stats.st_size)
physical_location = SubElement(bitstream, "xip:PhysicalLocation")
fixities = SubElement(bitstream, "xip:Fixities")
fixity_result = fixity_callback(filename_element.text, file)
if type(fixity_result) == tuple:
fixity = SubElement(fixities, "xip:Fixity")
fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef")
fixity_value = SubElement(fixity, "xip:FixityValue")
fixity_algorithm_ref.text = fixity_result[0]
fixity_value.text = fixity_result[1]
elif type(fixity_result) == dict:
for key, val in fixity_result.items():
fixity = SubElement(fixities, "xip:Fixity")
fixity_algorithm_ref = SubElement(fixity, "xip:FixityAlgorithmRef")
fixity_value = SubElement(fixity, "xip:FixityValue")
fixity_algorithm_ref.text = key
fixity_value.text = val
else:
logger.error("Could Not Find Fixity Value")
raise RuntimeError("Could Not Find Fixity Value")
if 'Identifiers' in kwargs:
identifier_map = kwargs.get('Identifiers')
if str(file) in identifier_map:
identifier_map_values = identifier_map[str(file)]
for identifier_key, identifier_value in identifier_map_values.items():
if identifier_key:
if identifier_value:
identifier = SubElement(xip, 'xip:Identifier')
id_type = SubElement(identifier, "xip:Type")
id_type.text = identifier_key
id_value = SubElement(identifier, "xip:Value")
id_value.text = identifier_value
id_io = SubElement(identifier, "xip:Entity")
id_io.text = io_ref
src_file = file
dst_file = os.path.join(os.path.join(inner_folder, CONTENT_FOLDER), os.path.basename(file))
shutil.copyfile(src_file, dst_file)
if xip is not None:
metadata_path = os.path.join(inner_folder, "metadata.xml")
metadata = open(metadata_path, "wt", encoding='utf-8')
metadata.write(prettify(xip))
metadata.close()
if compress:
shutil.make_archive(top_level_folder, 'zip', top_level_folder)
else:
shutil.make_archive(top_level_folder, 'szip', top_level_folder)
shutil.rmtree(top_level_folder)
return top_level_folder + ".zip"
[docs]
def complex_asset_package(preservation_files_list: list|None=None, access_files_list: list|None=None, export_folder=None, parent_folder=None,
compress=True,
**kwargs):
"""
Create a Preservica package containing a single Asset from a multiple preservation files
and optional access files.
The Asset contains multiple Content Objects within each representation.
If only the preservation files are provided the asset has one representation
:param list preservation_files_list: Paths to the preservation files
:param list access_files_list: Paths to the access files
:param str export_folder: The package location folder
:param Folder parent_folder: The folder to ingest the asset into
:param bool compress: Compress the ZIP file
:param str Title: Asset Title
:param str Description: Asset Description
:param str SecurityTag: Asset SecurityTag
:param str CustomType: Asset CustomType
:param str Preservation_Content_Title: Title of the Preservation Representation Content Object
:param str Preservation_Content_Description: Description of the Preservation Representation Content Object
:param str Access_Content_Title: Title of the Access Representation Content Object
:param str Access_Content_Description: Description of the Access Representation Content Object
:param dict Asset_Metadata: Dictionary of Asset metadata documents
:param dict Identifiers: Dictionary of Asset rd party identifiers
optional kwargs map
'Title' Asset Title
'Description' Asset Description
'SecurityTag' Asset Security Tag
'CustomType' Asset Type
'Preservation_Content_Title' Content Object Title of the Preservation Object
'Preservation_Content_Description' Content Object Description of the Preservation Object
'Access_Content_Title' Content Object Title of the Access Object
'Access_Content_Description' Content Object Description of the Access Object
'Preservation_Generation_Label' Generation Label for the Preservation Object
'Access_Generation_Label' Generation Label for the Access Object
'Asset_Metadata' Map of metadata schema/documents to add to asset
'Identifiers' Map of asset identifiers
'Preservation_files_fixity_callback' Callback to allow external generated fixity values
'Access_files_fixity_callback' Callback to allow external generated fixity values
'IO_Identifier_callback' Callback to allow external generated Asset identifier
'Preservation_Representation_Name' Name of the Preservation Representation
'Access_Representation_Name' Name of the Access Representation
"""
xml.etree.ElementTree.register_namespace("xip", "http://preservica.com/XIP/v6.0")
# some basic validation
if export_folder is None:
export_folder = tempfile.gettempdir()
if not os.path.isdir(export_folder):
logger.error("Export Folder Does Not Exist")
raise RuntimeError(export_folder, "Export Folder Does Not Exist")
if parent_folder is None:
logger.error("You must specify a parent folder for the package asset")
raise RuntimeError("You must specify a parent folder for the package asset")
io_ref = None
xip = None
default_asset_title = None
preservation_refs_dict = {}
access_refs_dict = {}
security_tag = kwargs.get('SecurityTag', "open")
content_type = kwargs.get('CustomType', "")
if not compress:
shutil.register_archive_format("szip", _make_stored_zipfile, None, "UnCompressed ZIP file")
has_preservation_files = bool((preservation_files_list is not None) and (len(preservation_files_list) > 0))
has_access_files = bool((access_files_list is not None) and (len(access_files_list) > 0))
if has_preservation_files:
if default_asset_title is None:
default_asset_title = os.path.splitext(os.path.basename(preservation_files_list[0]))[0]
# create the asset
if io_ref is None:
xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs)
if has_access_files:
if default_asset_title is None:
default_asset_title = os.path.splitext(os.path.basename(access_files_list[0]))[0]
if io_ref is None:
xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs)
if io_ref is None:
default_asset_title = kwargs.get('Title', None)
if default_asset_title is None:
default_asset_title = "New Asset"
xip, io_ref = __create_io__(file_name=default_asset_title, parent_folder=parent_folder, **kwargs)
if has_preservation_files:
# add the content objects
representation_name = kwargs.get('Preservation_Representation_Name', "Preservation")
preservation_refs_dict = __make_representation_multiple_co__(xip, rep_name=representation_name,
rep_type="Preservation",
rep_files=preservation_files_list, io_ref=io_ref)
if has_access_files:
# add the content objects
access_name = kwargs.get('Access_Representation_Name', "Access")
access_refs_dict = __make_representation_multiple_co__(xip, rep_name=access_name, rep_type="Access",
rep_files=access_files_list,
io_ref=io_ref)
if has_preservation_files:
for content_ref, filename in preservation_refs_dict.items():
default_content_objects_title = os.path.splitext(os.path.basename(filename))[0]
preservation_content_title = kwargs.get('Preservation_Content_Title', default_content_objects_title)
preservation_content_description = kwargs.get('Preservation_Content_Description',
default_content_objects_title)
if isinstance(preservation_content_title, dict):
preservation_content_title = preservation_content_title[filename]
if isinstance(preservation_content_description, dict):
preservation_content_description = preservation_content_description[filename]
__make_content_objects__(xip, preservation_content_title, content_ref, io_ref, security_tag,
preservation_content_description,
content_type)
if has_access_files:
for content_ref, filename in access_refs_dict.items():
default_content_objects_title = os.path.splitext(os.path.basename(filename))[0]
access_content_title = kwargs.get('Access_Content_Title', default_content_objects_title)
access_content_description = kwargs.get('Access_Content_Description', default_content_objects_title)
if isinstance(access_content_title, dict):
access_content_title = access_content_title[filename]
if isinstance(access_content_description, dict):
access_content_title = access_content_title[filename]
__make_content_objects__(xip, access_content_title, content_ref, io_ref, security_tag,
access_content_description, content_type)
if has_preservation_files:
preservation_generation_label = kwargs.get('Preservation_Generation_Label', "")
for content_ref, filename in preservation_refs_dict.items():
preservation_file_name = os.path.basename(filename)
__make_generation__(xip, preservation_file_name, content_ref, preservation_generation_label,
PRESERVATION_CONTENT_FOLDER)
if has_access_files:
access_generation_label = kwargs.get('Access_Generation_Label', "")
for content_ref, filename in access_refs_dict.items():
access_file_name = os.path.basename(filename)
__make_generation__(xip, access_file_name, content_ref, access_generation_label, ACCESS_CONTENT_FOLDER)
if has_preservation_files:
if 'Preservation_files_fixity_callback' in kwargs:
callback = kwargs.get('Preservation_files_fixity_callback')
else:
callback = Sha1FixityCallBack()
for content_ref, filename in preservation_refs_dict.items():
preservation_file_name = os.path.basename(filename)
__make_bitstream__(xip, preservation_file_name, filename, callback, PRESERVATION_CONTENT_FOLDER)
if has_access_files:
if 'Access_files_fixity_callback' in kwargs:
callback = kwargs.get('Access_files_fixity_callback')
else:
callback = Sha1FixityCallBack()
for content_ref, filename in access_refs_dict.items():
access_file_name = os.path.basename(filename)
__make_bitstream__(xip, access_file_name, filename, callback, ACCESS_CONTENT_FOLDER)
if 'Identifiers' in kwargs:
identifier_map = kwargs.get('Identifiers')
for identifier_key, identifier_value in identifier_map.items():
if identifier_key:
if identifier_value:
identifier = SubElement(xip, 'xip:Identifier')
id_type = SubElement(identifier, "xip:Type")
id_type.text = identifier_key
id_value = SubElement(identifier, "xip:Value")
id_value.text = identifier_value
id_io = SubElement(identifier, "xip:Entity")
id_io.text = io_ref
if 'Asset_Metadata' in kwargs:
metadata_map = kwargs.get('Asset_Metadata')
for metadata_ns, metadata_path in metadata_map.items():
if metadata_ns:
if metadata_path and isinstance(metadata_path, str):
if os.path.exists(metadata_path) and os.path.isfile(metadata_path):
descriptive_metadata = xml.etree.ElementTree.parse(source=metadata_path)
metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns})
metadata_ref = SubElement(metadata, 'xip:Ref')
metadata_ref.text = str(uuid.uuid4())
entity = SubElement(metadata, 'xip:Entity')
entity.text = io_ref
content = SubElement(metadata, 'xip:Content')
content.append(descriptive_metadata.getroot())
elif isinstance(metadata_path, str):
try:
descriptive_metadata = xml.etree.ElementTree.fromstring(metadata_path)
metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns})
metadata_ref = SubElement(metadata, 'xip:Ref')
metadata_ref.text = str(uuid.uuid4())
entity = SubElement(metadata, 'xip:Entity')
entity.text = io_ref
content = SubElement(metadata, 'xip:Content')
content.append(descriptive_metadata)
except RuntimeError:
logging.info(f"Could not parse asset metadata in namespace {metadata_ns}")
if metadata_path and isinstance(metadata_path, list):
for path in metadata_path:
if os.path.exists(path) and os.path.isfile(path):
descriptive_metadata = xml.etree.ElementTree.parse(source=path)
metadata = SubElement(xip, 'xip:Metadata', {'schemaUri': metadata_ns})
metadata_ref = SubElement(metadata, 'xip:Ref')
metadata_ref.text = str(uuid.uuid4())
entity = SubElement(metadata, 'xip:Entity')
entity.text = io_ref
content = SubElement(metadata, 'xip:Content')
content.append(descriptive_metadata.getroot())
if xip is not None:
export_folder = export_folder
top_level_folder = os.path.join(export_folder, io_ref)
os.mkdir(top_level_folder)
inner_folder = os.path.join(top_level_folder, io_ref)
os.mkdir(inner_folder)
content_folder = os.path.join(inner_folder, CONTENT_FOLDER)
os.mkdir(content_folder)
preservation_content_folder = os.path.join(content_folder, PRESERVATION_CONTENT_FOLDER)
os.mkdir(preservation_content_folder)
access_content_folder = os.path.join(content_folder, ACCESS_CONTENT_FOLDER)
os.mkdir(access_content_folder)
metadata_path = os.path.join(inner_folder, "metadata.xml")
metadata = open(metadata_path, "wt", encoding='utf-8')
metadata.write(prettify(xip))
metadata.close()
for content_ref, filename in preservation_refs_dict.items():
src_file = filename
dst_file = os.path.join(preservation_content_folder, os.path.basename(filename))
shutil.copyfile(src_file, dst_file)
for content_ref, filename in access_refs_dict.items():
src_file = filename
dst_file = os.path.join(access_content_folder, os.path.basename(filename))
shutil.copyfile(src_file, dst_file)
if compress:
shutil.make_archive(top_level_folder, 'zip', top_level_folder)
else:
shutil.make_archive(top_level_folder, 'szip', top_level_folder)
shutil.rmtree(top_level_folder)
return top_level_folder + ".zip"
[docs]
def simple_asset_package(preservation_file: str|None=None, access_file: str|None=None, export_folder:str|None=None, parent_folder=None,
compress=True, **kwargs):
"""
Create a Preservica package containing a single Asset from a single preservation file
and an optional access file.
The Asset contains one Content Object for each representation.
If only the preservation file is provided the asset has one representation
:param str preservation_file: Path to the preservation file
:param str access_file: Path to the access file
:param str export_folder: The package location folder
:param Folder parent_folder: The folder to ingest the asset into
:param bool compress: Compress the ZIP file
:param str Title: Asset Title
:param str Description: Asset Description
:param str SecurityTag: Asset SecurityTag
:param str CustomType: Asset CustomType
:param str Preservation_Content_Title: Title of the Preservation Representation Content Object
:param str Preservation_Content_Description: Description of the Preservation Representation Content Object
:param str Access_Content_Title: Title of the Access Representation Content Object
:param str Access_Content_Description: Description of the Access Representation Content Object
:param dict Asset_Metadata: Dictionary of Asset metadata documents
:param dict Identifiers: Dictionary of Asset rd party identifiers
"""
# some basic validation
if export_folder is None:
export_folder = tempfile.gettempdir()
if not os.path.isdir(export_folder):
logger.error("Export Folder Does Not Exist")
raise RuntimeError(export_folder, "Export Folder Does Not Exist")
if parent_folder is None:
logger.error("You must specify a parent folder for the package asset")
raise RuntimeError("You must specify a parent folder for the package asset")
preservation_file_list = list()
access_file_list = list()
if preservation_file is not None:
preservation_file_list.append(preservation_file)
if access_file is not None:
access_file_list.append(access_file)
return complex_asset_package(preservation_files_list=preservation_file_list, access_files_list=access_file_list,
export_folder=export_folder, parent_folder=parent_folder, compress=compress, **kwargs)
def upload_config():
return transfer_config
def _unpad(s):
return s[:-ord(s[len(s) - 1:])]
[docs]
class UploadAPI(AuthenticatedAPI):
def __init__(self, username: str|None = None, password: str|None = None, tenant: str|None = None, server: str|None = None,
use_shared_secret: bool = False, two_fa_secret_key: str|None = None,
protocol: str = "https", request_hook: Callable|None = None, credentials_path: str = 'credentials.properties'):
super().__init__(username, password, tenant, server, use_shared_secret, two_fa_secret_key,
protocol, request_hook, credentials_path)
self.uploads_per_minute: int = 180
self.limiter: Limiter = None
[docs]
def ingest_web_video(self, url=None, parent_folder=None, **kwargs):
"""
Download a web video (e.g. YouTube) and ingest it into Preservica as an asset.
Requires the optional ``youtube_dl`` package (``pip install --upgrade youtube-dl``).
The video is downloaded to the current working directory as ``<video_id>.mp4``,
wrapped in a simple preservation package, and then uploaded via
:meth:`upload_zip_package`.
:param url: URL of the web video to download and ingest.
:type url: str or None
:param parent_folder: The Preservica folder into which the video asset is ingested.
:type parent_folder: Folder or str or None
:param Title: Optional title for the asset. Defaults to the video title from
the metadata.
:type Title: str
:param Description: Optional description for the asset. Defaults to the video
description from the metadata.
:type Description: str
:param SecurityTag: Optional security tag for the asset. Defaults to ``"open"``.
:type SecurityTag: str
:param Identifiers: Optional mapping of identifier type to value; the video id is
automatically added under the key ``"Video-ID"``.
:type Identifiers: dict
:param Asset_Metadata: Optional mapping of schema URI to metadata document for the
asset.
:type Asset_Metadata: dict
:param callback: Optional upload progress callback.
:type callback: callable or None
:raises RuntimeError: If the ``youtube_dl`` package is not installed.
"""
try:
import youtube_dl
except ImportError:
logger.error("Package youtube_dl is required for this method. pip install --upgrade youtube-dl")
raise RuntimeError("Package youtube_dl is required for this method. pip install --upgrade youtube-dl")
ydl_opts = {}
def my_hook(d):
if d['status'] == 'finished':
logger.info('Download Complete. Uploading to Preservica ...')
ydl_opts = {'outtmpl': '%(id)s.mp4', 'progress_hooks': [my_hook], }
# if True:
# ydl_opts['writesubtitles'] = True
# ydl_opts['writeautomaticsub'] = True
# ydl_opts['subtitleslangs'] = ['en']
with youtube_dl.YoutubeDL(ydl_opts) as ydl:
meta = ydl.extract_info(url, download=True)
vid_id = meta.get('id')
if 'Title' in kwargs:
title = kwargs.get("Title")
else:
title = meta.get('title')
if 'Description' in kwargs:
description = kwargs.get("Description")
else:
description = meta.get('description')
if 'SecurityTag' in kwargs:
security_tag = kwargs.get("SecurityTag")
else:
security_tag = "open"
if 'Identifiers' in kwargs:
identifier_map = kwargs.get('Identifiers')
identifier_map["Video-ID"] = vid_id
else:
identifier_map = {"Video-ID": vid_id}
if 'Asset_Metadata' in kwargs:
descriptive_metadata = kwargs.get('Asset_Metadata')
else:
descriptive_metadata = {}
if 'callback' in kwargs:
callback = kwargs.get("callback")
else:
callback = None
upload_date = meta.get('upload_date')
duration = meta.get('duration')
package = simple_asset_package(preservation_file=f"{vid_id}.mp4", parent_folder=parent_folder, Title=title,
Description=description, Identifiers=identifier_map,
Asset_Metadata=descriptive_metadata,
Preservation_Content_Title=title, SecurityTag=security_tag)
self.upload_zip_package(path_to_zip_package=package, folder=parent_folder, callback=callback)
[docs]
def upload_credentials(self, location_id: str) -> dict:
"""
Retrieve temporary upload credentials for a specific upload location.
For AWS deployments, the credentials are Amazon STS temporary credentials
(``key``, ``secret``, ``sessionToken``, ``endpoint``). For Azure deployments
they are Azure SAS credentials (``key``, ``sessionToken``).
:param location_id: The API id of the upload location (from :meth:`upload_locations`).
:type location_id: str
:returns: Dictionary of credential fields appropriate for the deployment type.
:rtype: dict
:raises HTTPException: If the server returns a non-200 response.
"""
headers = {HEADER_TOKEN: self.token}
endpoint = f"/upload/{location_id}/upload-credentials"
request = self.session.get(f'https://{self.server}/api/location{endpoint}', headers=headers)
if request.status_code == requests.codes.ok:
json_response = str(request.content.decode('utf-8'))
return json.loads(json_response)
else:
exception = HTTPException(location_id, request.status_code, request.url, "upload_credentials",
request.content.decode('utf-8'))
logger.error(exception)
raise exception
[docs]
def clean_upload_bucket(self, bucket_name: str, older_than_days: int = 90):
"""
Delete objects in an upload bucket that are older than a given number of days.
Supports both AWS S3 buckets and Azure Blob Storage containers. The appropriate
cloud credentials are obtained automatically via :meth:`upload_credentials`.
:param bucket_name: The name of the S3 bucket or Azure container to clean.
:type bucket_name: str
:param older_than_days: Objects whose last-modified timestamp is more than this
many days in the past are deleted. Defaults to ``90``.
:type older_than_days: int
:returns: None
:rtype: None
"""
from azure.storage.blob import ContainerClient
for location in self.upload_locations():
if location['containerName'] == bucket_name:
if location['type'] != 'AWS':
credentials = self.upload_credentials(location['apiId'])
account_key = credentials['key']
session_token = credentials['sessionToken']
sas_url = f"https://{account_key}.blob.core.windows.net/{bucket_name}"
container = ContainerClient.from_container_url(container_url=sas_url, credential=session_token)
now = datetime.now(timezone.utc)
for blob in container.list_blobs():
if abs((blob.last_modified - now).days) > older_than_days:
logger.debug(f"Deleting expired object {blob.name}")
container.delete_blob(blob.name)
if location['type'] == 'AWS':
credentials = self.upload_credentials(location['apiId'])
access_key = credentials['key']
secret_key = credentials['secret']
session_token = credentials['sessionToken']
session = boto3.Session(aws_access_key_id=access_key, aws_secret_access_key=secret_key,
aws_session_token=session_token)
s3_client = session.client("s3")
paginator = s3_client.get_paginator('list_objects_v2')
now = datetime.now(timezone.utc)
for page in paginator.paginate(Bucket=bucket_name):
if 'Contents' in page:
for key in page['Contents']:
last_modified = key['LastModified']
if abs((last_modified - now).days) > older_than_days:
logger.debug(f"Deleting expired object {key['Key']}")
s3_client.delete_object(Bucket=bucket_name, Key=key['Key'])
[docs]
def upload_locations(self) -> dict:
"""
Return all configured upload locations for this Preservica system.
Upload locations are configured on the Preservica Sources page as *SIP Upload*
entries and represent S3 buckets or Azure containers connected to ingest workflows.
:returns: List of upload location dictionaries, each containing at least
``containerName``, ``type`` (``"AWS"`` or ``"Azure"``), and ``apiId``.
:rtype: list[dict]
:raises HTTPException: If the server returns a non-200 response.
"""
headers = {HEADER_TOKEN: self.token}
endpoint = "/api/location/upload"
request = self.session.get(f'https://{self.server}{endpoint}', headers=headers)
if request.status_code == requests.codes.ok:
json_response = str(request.content.decode('utf-8'))
return json.loads(json_response)['locations']
else:
exception = HTTPException("", request.status_code, request.url, "upload_locations",
request.content.decode('utf-8'))
logger.error(exception)
raise exception
[docs]
def upload_buckets(self) -> dict:
"""
Get a list of available upload buckets.
Alias for :meth:`upload_locations`; returns all upload locations configured
on the Preservica system (both S3 buckets and Azure containers).
:returns: List of upload location dictionaries (see :meth:`upload_locations`).
:rtype: list[dict]
:raises HTTPException: If the server returns a non-200 response.
"""
return self.upload_locations()
def crawl_filesystem(self, filesystem_path, bucket_name, preservica_parent, callback: bool = False,
security_tag: str = "open",
delete_after_upload: bool = True, max_MB_ingested: int = -1):
from pyPreservica import EntityAPI
def entity_value(client: EntityAPI, identifier: str) -> Entity:
back_off: int = 5
while True:
try:
entities = client.identifier("code", identifier)
if bool(len(entities) > 0):
return entities.pop()
else:
return None
except HTTPException as e:
sleep(back_off)
back_off = back_off * 2
def entity_exists(client: EntityAPI, identifier: str) -> bool:
back_off: int = 5
while True:
try:
entities = client.identifier("code", identifier)
return bool(len(entities) > 0)
except HTTPException as e:
sleep(back_off)
back_off = back_off * 2
def get_parent(client, identifier, parent_reference):
dirname_id: str = str(os.path.dirname(identifier))
if not dirname_id:
dirname_id = identifier
folder = entity_value(client, dirname_id)
if folder is not None:
folder = client.folder(folder.reference)
return folder.reference
else:
return parent_reference
def get_folder(client, name, tag, parent_reference, identifier):
folder = entity_value(client, identifier)
if folder is None:
logger.info(f"Creating new folder with name {name}")
folder = client.create_folder(name, name, tag, parent_reference)
client.add_identifier(folder, "code", identifier)
else:
logger.info(f"Found existing folder with name {name}")
return folder
entity_client = EntityAPI(username=self.username, password=self.password, server=self.server,
tenant=self.tenant,
two_fa_secret_key=self.two_fa_secret_key, use_shared_secret=self.shared_secret,
protocol=self.protocol)
if preservica_parent:
parent = entity_client.folder(preservica_parent)
logger.info(f"Folders will be created inside Preservica collection {parent.title}")
parent_ref = parent.reference
else:
parent_ref = None
bytes_ingested = 0
folder_path = os.path.normpath(filesystem_path)
for dirname, subdirs, files in os.walk(folder_path):
base = os.path.basename(dirname)
code = os.path.relpath(dirname, Path(folder_path).parent)
p = get_parent(entity_client, code, parent_ref)
f = get_folder(entity_client, base, security_tag, p, code)
identifiers = dict()
for file in list(files):
full_path = os.path.join(dirname, file)
if os.path.islink(full_path):
logger.info(f"Skipping link {file}")
files.remove(file)
continue
asset_code = os.path.join(code, file)
if not entity_exists(entity_client, asset_code):
bytes_ingested = bytes_ingested + os.stat(full_path).st_size
logger.info(f"Adding new file: {file} to package ready for upload")
file_identifiers = {"code": asset_code}
identifiers[full_path] = file_identifiers
else:
logger.info(f"Skipping file {file} already exists in repository")
files.remove(file)
if len(files) > 0:
full_path_list = [os.path.join(dirname, file) for file in files]
package = multi_asset_package(asset_file_list=full_path_list, parent_folder=f, SecurityTag=security_tag,
Identifiers=identifiers)
if callback:
progress_display = UploadProgressConsoleCallback(package)
else:
progress_display = None
if bucket_name is None:
self.upload_zip_package(path_to_zip_package=package, callback=progress_display,
delete_after_upload=delete_after_upload)
else:
self.upload_zip_to_Source(path_to_zip_package=package, container_name=bucket_name,
show_progress=bool(progress_display is not None),
delete_after_upload=delete_after_upload)
logger.info(f"Uploaded " + "{:.1f}".format(bytes_ingested / (1024 * 1024)) + " MB")
if max_MB_ingested > 0:
if bytes_ingested > (1024 * 1024 * max_MB_ingested):
logger.info(f"Reached Max Upload Limit")
break
[docs]
def upload_zip_to_Source(self, path_to_zip_package: str, container_name, folder=None, delete_after_upload=False,
show_progress=False):
"""
Upload a ZIP package to an upload source (Azure container or S3 bucket).
Automatically dispatches to :meth:`upload_zip_package_to_S3` or
:meth:`upload_zip_package_to_Azure` based on the deployment type reported by
:meth:`upload_locations`.
:param path_to_zip_package: Local path to the ZIP package file.
:type path_to_zip_package: str
:param container_name: Name of the S3 bucket or Azure container connected to an
ingest workflow.
:type container_name: str
:param folder: Optional Preservica folder (or UUID string) that the ingested
package should be placed into.
:type folder: Folder or str or None
:param delete_after_upload: When ``True``, the local ZIP file is deleted after a
successful upload. Defaults to ``False``.
:type delete_after_upload: bool
:param show_progress: When ``True``, an upload progress bar is displayed on the
console. Defaults to ``False``.
:type show_progress: bool
:returns: None
:rtype: None
"""
locations = self.upload_locations()
for location in locations:
if location['containerName'] == container_name:
if location['type'] == 'AWS':
callback = None
if show_progress:
callback = UploadProgressConsoleCallback(path_to_zip_package)
self.upload_zip_package_to_S3(path_to_zip_package=path_to_zip_package, bucket_name=container_name,
folder=folder,
callback=callback, delete_after_upload=delete_after_upload)
else:
self.upload_zip_package_to_Azure(path_to_zip_package=path_to_zip_package,
container_name=container_name, folder=folder,
delete_after_upload=delete_after_upload,
show_progress=show_progress)
[docs]
def upload_zip_package_to_Azure(self, path_to_zip_package: str, container_name: str, folder=None, delete_after_upload=False,
show_progress=False):
"""
Uploads a zip file package to an Azure container connected to a Preservica Cloud System
:param str path_to_zip_package: Path to the package
:param str container_name: container connected to the ingest workflow
:param Folder folder: The folder to ingest the package into
:param bool delete_after_upload: Delete the local copy of the package after the upload has completed
"""
if (self.major_version < 7) and (self.minor_version < 5):
raise RuntimeError(
"This call [upload_zip_package_to_Azure] is only available against v6.5 systems and above")
from azure.storage.blob import ContainerClient
locations = self.upload_locations()
for location in locations:
if location['containerName'] == container_name:
credentials = self.upload_credentials(location['apiId'])
account_key = credentials['key']
session_token = credentials['sessionToken']
sas_url = f"https://{account_key}.blob.core.windows.net/{container_name}"
container = ContainerClient.from_container_url(container_url=sas_url, credential=session_token)
upload_key = str(uuid.uuid4())
metadata = {'key': upload_key, 'name': upload_key + ".zip", 'bucket': container_name, 'status': 'ready'}
if hasattr(folder, "reference"):
metadata['collectionreference'] = folder.reference
elif isinstance(folder, str):
metadata['collectionreference'] = folder
properties = None
len_bytes = Path(path_to_zip_package).stat().st_size
if show_progress:
with tqdm.wrapattr(open(path_to_zip_package, 'rb'), "read", total=len_bytes) as data:
blob_client = container.upload_blob(name=upload_key, data=data, metadata=metadata,
length=len_bytes)
properties = blob_client.get_blob_properties()
else:
with open(path_to_zip_package, "rb") as data:
blob_client = container.upload_blob(name=upload_key, data=data, metadata=metadata,
length=len_bytes)
properties = blob_client.get_blob_properties()
if delete_after_upload:
os.remove(path_to_zip_package)
return properties
[docs]
def upload_zip_package_to_S3(self, path_to_zip_package: str, bucket_name: str, folder=None, callback=None,
delete_after_upload=False):
"""
Uploads a zip file package to an S3 bucket connected to a Preservica Cloud System
:param str path_to_zip_package: Path to the package
:param str bucket_name: Bucket connected to an ingest workflow
:param Folder folder: The folder to ingest the package into
:param Callable callback: Optional callback to allow the callee to monitor the upload progress
:param bool delete_after_upload: Delete the local copy of the package after the upload has completed
"""
if (self.major_version < 7) and (self.minor_version < 5):
raise RuntimeError("This call [upload_zip_package_to_S3] is only available against v6.5 systems and above")
logger.debug("Finding Upload Locations")
self.token = self.__token__()
locations = self.upload_locations()
for location in locations:
if location['containerName'] == bucket_name:
logger.debug(f"Found Upload Location {location['containerName']}")
logger.debug(f"Fetching Upload Credentials for {location['containerName']}")
credentials = self.upload_credentials(location['apiId'])
access_key = credentials['key']
secret_key = credentials['secret']
session_token = credentials['sessionToken']
endpoint = credentials['endpoint']
session = boto3.Session(aws_access_key_id=access_key, aws_secret_access_key=secret_key,
aws_session_token=session_token)
s3 = session.resource(service_name="s3")
logger.debug(f"S3 Session: {s3}")
upload_key = str(uuid.uuid4())
s3_object = s3.Object(bucket_name, upload_key)
metadata = {'key': upload_key, 'name': upload_key + ".zip", 'bucket': bucket_name, 'status': 'ready'}
if hasattr(folder, "reference"):
metadata['collectionreference'] = folder.reference
elif isinstance(folder, str):
metadata['collectionreference'] = folder
metadata['size'] = str(Path(path_to_zip_package).stat().st_size)
metadata['createdby'] = self.username
metadata_map = {'Metadata': metadata}
s3_object.upload_file(path_to_zip_package, Callback=callback, ExtraArgs=metadata_map,
Config=transfer_config)
if delete_after_upload:
os.remove(path_to_zip_package)
[docs]
def upload_zip_package(self, path_to_zip_package, folder=None, callback=None, delete_after_upload=False, limit_per_minute= 180):
"""
Upload a ZIP package directly to Preservica via the built-in S3 endpoint and start an ingest workflow.
Uses auto-refreshing Preservica token credentials so that long-running uploads
do not fail due to token expiry. Multipart chunk sizes are automatically adjusted
based on the package size.
This function can be rate limited to restrict the number of uploads per minute, it will default to a max of 180 / min.
:param path_to_zip_package: Local path to the ZIP package file.
:type path_to_zip_package: str
:param folder: Optional Preservica folder (or UUID string) that the ingested
package should be placed into.
:type folder: Folder or str or None
:param callback: Optional progress callback invoked during the upload. Must be
compatible with the boto3 ``Callback`` interface (i.e. a callable that accepts
the number of bytes transferred).
:type callback: callable or None
:param delete_after_upload: When ``True``, the local ZIP file is deleted after a
successful upload. Defaults to ``False``.
:type delete_after_upload: bool
:param limit_per_minute: Restrict the number of uploads per min, default is 180 (3 per second)
:type limit_per_minute: int
:returns: The ``preservica-progress-token`` header value returned by the server,
which can be used to monitor the ingest workflow progress.
:rtype: str
:raises NoCredentialsError: If AWS credentials cannot be resolved.
:raises ClientError: If the S3 upload fails.
"""
if self.limiter is None:
self.limiter = Limiter(Rate(limit_per_minute, Duration.MINUTE))
self.uploads_per_minute = limit_per_minute
if limit_per_minute != self.uploads_per_minute:
self.limiter = Limiter(Rate(limit_per_minute, Duration.MINUTE))
self.uploads_per_minute = limit_per_minute
self.limiter.try_acquire("upload_zip_package")
bucket = f'{self.tenant.lower()}.package.upload'
endpoint = f'{self.protocol}://{self.server}/api/s3/buckets'
self.token = self.__token__()
retries= {
'max_attempts': 5,
'mode': 'adaptive'
}
def new_credentials():
cred_metadata: dict = {}
cred_metadata['access_key'] = self.__token__()
cred_metadata['secret_key'] = "NOT_USED"
cred_metadata['token'] = ""
cred_metadata["expiry_time"] = (datetime.now(tzlocal()) + timedelta(minutes=12)).isoformat()
logger.info("Refreshing credentials at: " + str(datetime.now(tzlocal())))
return cred_metadata
session = get_session()
session_credentials = RefreshableCredentials.create_from_metadata(
metadata=new_credentials(),
refresh_using=new_credentials,
advisory_timeout = 4 * 60,
mandatory_timeout = 12 * 60,
method = 'Preservica'
)
autorefresh_session = boto3.Session(botocore_session=session)
session._credentials = session_credentials
config = Config(s3={'addressing_style': 'path'}, read_timeout=120, connect_timeout=120,
request_checksum_calculation="WHEN_REQUIRED",
response_checksum_validation="WHEN_REQUIRED",
retries=retries, tcp_keepalive=True)
s3_client = autorefresh_session.client('s3', endpoint_url=endpoint, config=config)
metadata = {}
if folder is not None:
if hasattr(folder, "reference"):
metadata = {'Metadata': {'structuralobjectreference': folder.reference}}
elif isinstance(folder, str):
metadata = {'Metadata': {'structuralobjectreference': folder}}
if os.path.exists(path_to_zip_package) and os.path.isfile(path_to_zip_package):
try:
key_id = str(uuid.uuid4()) + ".zip"
# how big is the package
package_size = os.path.getsize(path_to_zip_package)
if package_size > 1 * GB:
transfer_config.multipart_chunksize = 16 * MB ## Min 64 Chunks
if package_size > 8 * GB:
transfer_config.multipart_chunksize = 32 * MB ## Min 256 Chunks
if package_size > 24 * GB:
transfer_config.multipart_chunksize = 48 * MB ## Min 512 Chunks
if package_size > 48 * GB:
transfer_config.multipart_chunksize = 64 * MB
logger.info("Using Multipart Chunk Size: " + str(transfer_config.multipart_chunksize))
transfer = S3Transfer(client=s3_client, config=transfer_config)
transfer.PutObjectTask = PutObjectTask
transfer.CompleteMultipartUploadTask = CompleteMultipartUploadTask
transfer.upload_file = upload_file
response = transfer.upload_file(self=transfer, filename=path_to_zip_package, bucket=bucket,
key=key_id,
extra_args=metadata,
callback=callback)
if delete_after_upload:
os.remove(path_to_zip_package)
return response['ResponseMetadata']['HTTPHeaders']['preservica-progress-token']
except (NoCredentialsError, PartialCredentialsError) as ex:
logger.error(ex)
raise ex
except ClientError as ex:
logger.error(ex)
raise ex