swiftclient.SwiftService API

一个更高级的 API，旨在允许开发人员使用可配置的线程池以简单的方式异步执行多个操作。每个服务方法调用的文档可以在这里找到：swiftclient.service。

身份验证

本节涵盖了与 swift 对象存储进行身份验证的各种选项。每种身份验证版本所需的选项组合如下所示。再次说明，这些只是可以成功进行身份验证的选项的子集，但它们是最常见和推荐的。

相关的身份验证选项表示为 python 字典，应将其添加到您提供给 SwiftService 实例的任何其他选项中。如 python 代码所示，您还可以将这些选项设置为环境变量，如果未指定相关选项，则会自动加载它们。

SwiftService 身份验证尝试根据指定的选项组合自动选择身份验证版本，但提供来自不同身份验证版本的选项可能会导致意外行为。

注意

剩余的环境变量是授权失败时的常见混淆来源。

Keystone V3

{
    ...
    "auth_version": environ.get('ST_AUTH_VERSION'),  # Should be '3'
    "os_username": environ.get('OS_USERNAME'),
    "os_password": environ.get('OS_PASSWORD'),
    "os_project_name": environ.get('OS_PROJECT_NAME'),
    "os_project_domain_name": environ.get('OS_PROJECT_DOMAIN_NAME'),
    "os_auth_url": environ.get('OS_AUTH_URL'),
    ...
}

{
    ...
    "auth_version": environ.get('ST_AUTH_VERSION'),  # Should be '3'
    "os_username": environ.get('OS_USERNAME'),
    "os_password": environ.get('OS_PASSWORD'),
    "os_project_id": environ.get('OS_PROJECT_ID'),
    "os_project_domain_id": environ.get('OS_PROJECT_DOMAIN_ID'),
    "os_auth_url": environ.get('OS_AUTH_URL'),
    ...
}

Keystone V2

{
    ...
    "auth_version": environ.get('ST_AUTH_VERSION'),  # Should be '2.0'
    "os_username": environ.get('OS_USERNAME'),
    "os_password": environ.get('OS_PASSWORD'),
    "os_tenant_name": environ.get('OS_TENANT_NAME'),
    "os_auth_url": environ.get('OS_AUTH_URL'),
    ...
}

Legacy Auth

{
    ...
    "auth_version": environ.get('ST_AUTH_VERSION'),  # Should be '1.0'
    "auth": environ.get('ST_AUTH'),
    "user": environ.get('ST_USER'),
    "key": environ.get('ST_KEY'),
    ...
}

配置

当您创建 SwiftService 的实例时，您可以覆盖默认选项集合以适应您的用例。通常，默认值对于我们入门来说是合理的，但根据您的需要，您可能需要调整它们以提高性能（影响大型对象和线程计数的选项可以在正确的情况下显着改变性能）。

服务级别默认值和一些额外的选项也可以在每个操作（甚至在某些情况下每个对象）的基础上覆盖，稍后在文档中会说明哪些选项会影响哪些操作。

使用传递给 SwiftService 在初始化期间的选项字典来执行服务 API 的配置。以下描述了此字典中可用的选项及其默认值

选项

retries: 5

在放弃并报告失败之前，库应该尝试重试 HTTP 操作的次数。

container_threads: 10

object_dd_threads: 10

object_uu_threads: 10

segment_threads: 10

以上选项确定了执行 swift 操作的可用线程池的大小。容器操作（例如列出容器）在容器线程中运行，对象和分段线程也遵循类似的模式。

注意

对象线程分为两个单独的线程池：uu 和 dd。这代表“上传/更新”和“下载/删除”，相应的操作将在单独的线程池上运行。

segment_size: None

如果指定，此选项启用大型对象上传。如果上传的对象大于 5G，则此选项是强制性的，否则上传将失败。此选项应指定为字节大小。

use_slo: False

与上述选项结合使用，use_slo 将以静态方式而不是动态方式上传大型对象。只有静态大型对象才提供下载对象的错误检查，因此我们建议使用此选项。

segment_container: None

允许用户选择将大型对象分段上传到的容器。我们不建议更改此值，因为它可能会使在发生错误时定位孤立分段变得更加困难。

leave_segments: False

将此选项设置为 true 意味着在删除或覆盖大型对象时，其分段将保留在对象存储中，并且必须手动清理它们。此选项在更高级的场景中在多个对象之间共享大型对象分段时很有用，但必须小心处理，因为它可能导致存储使用量不断增加。

changed: None

此选项影响上传，并且仅意味着如果源的 mtime 和大小与现有对象相同，则已经存在于对象存储中的那些对象将不会被覆盖。

skip_identical: False

这是上述情况的更彻底的情况，但不是使用 mtime 和大小，而是使用对象的 MD5 sum。

yes_all: False

此选项仅影响下载和删除，并且在每种情况下都必须指定才能下载/删除帐户的全部内容。此选项对任何其他调用没有影响。

no_download: False

此选项仅影响下载，意味着所有操作都将正常进行，除了不会将任何数据写入磁盘。

header: []

用于上传和 post 操作，以在对象上设置标头。标头指定为冒号分隔的字符串，例如“content-type:text/plain”。

meta: []

用于以类似于标头的方式设置对象上的元数据。

注意

设置元数据是一种破坏性操作，因此在更新许多元数据值之一时，必须重新应用对象所需的所有元数据。

long: False

仅影响列表操作，并以牺牲性能为代价，使结果中提供更多的指标。

fail_fast: False

应用于删除和上传操作，并尝试在发生错误时中止排队的任务。

prefix: None

影响列表操作；只有具有给定前缀的对象才会被返回/影响。不建议在服务级别设置，因为调用列表以发现应该在其上运行对象的那些操作也会受到影响。

delimiter: None

影响列表操作，并且意味着列表仅包含对象名称中第一次出现分隔符为止的结果。这对于使用名称中包含“/”的对象以模拟文件夹结构很有用。

dir_marker: False

影响上传，并允许在上传源为 None 时创建空的“伪文件夹”对象。

checksum: True

影响上传和下载。如果设置，则检查传输的 md5 sum。

shuffle: False

在下载对象时，CLI 的默认行为是打乱对象列表，以便在多个客户端将相同文件下载到多个位置时，将负载分散到存储驱动器上（例如，在分发更新时）。当直接使用 SwiftService 时，对象下载按它们在容器列表中出现的顺序安排。当与单个下载线程结合使用时，这意味着对象将按词法顺序下载。将此选项设置为 True 给出与 CLI 相同的打乱行为。

destination: None

在复制对象时，这指定了对象将被复制到的目标位置。默认值为 None，表示复制将与源相同。

fresh_metadata: None

在复制对象时，这指定源上的对象元数据将 *不会* 应用于目标对象 - 目标对象将具有一组新的元数据，其中仅包含如果在 meta 选项中指定了任何元数据。

其他可用选项可以在 swiftclient/service.py 中找到，在 python-swiftclient 的源代码中。每个 SwiftService 方法还允许一个可选字典来覆盖初始化时指定的那些，并且适当的文档字符串显示哪些选项修改每个方法的行为。

可用操作

服务 API 提供的每个操作都可能引发 SwiftError 或 ClientException，用于任何完全失败的调用（或仅对帐户或容器级别执行单个操作的调用）。在成功调用的情况下，操作会返回以下之一

• 详细说明单个操作结果的字典。

• 生成结果字典的迭代器（对于执行多个子操作的调用）。

结果字典可以指示单个操作的成功或失败（在 success 键中详细说明），并且将包含成功结果，或者包含详细说明遇到的错误的 error 键（通常是 Exception 的实例）。

下面给出了一个示例结果字典

result = {
    'action': 'download_object',
    'success': True,
    'container': container,
    'object': obj,
    'path': path,
    'start_time': start_time,
    'finish_time': finish_time,
    'headers_receipt': headers_receipt,
    'auth_end_time': conn.auth_end_time,
    'read_length': bytes_read,
    'attempts': conn.attempts
}

下面详细说明了所有可能的 action 值

[
    'stat_account',
    'stat_container',
    'stat_object',
    'post_account',
    'post_container',
    'post_object',
    'list_part',          # list yields zero or more 'list_part' results
    'download_object',
    'create_container',   # from upload
    'create_dir_marker',  # from upload
    'upload_object',
    'upload_segment',
    'delete_container',
    'delete_object',
    'delete_segment',     # from delete_object operations
    'capabilities',
]

Stat

Stat 可以针对帐户、容器或对象列表调用，以获取帐户统计信息、容器统计信息或有关给定对象的信息。在前两种情况下，将返回包含操作结果的字典，在提供对象名称列表的情况下，将返回一个迭代器，该迭代器生成为每个对象生成的结果。

返回的信息包括给定对象/容器/帐户使用的量以及设置的任何标头或元数据（这包括用户设置的数据以及 content-type 和修改时间）。

请参阅 swiftclient.service.SwiftService.stat 以获取从方法文档字符串生成的文档。

此方法的有效调用如下

• stat([options]): 返回配置帐户的统计信息。

• stat(<container>, [options]): 返回给定容器的统计信息。

• stat(<container>, <object_list>, [options]): 返回给定容器中给定对象的统计信息（通过返回的迭代器）。

stat 的结果是字典，指示每个操作的成功或失败。如果对帐户或容器执行 stat 操作成功，该方法将立即返回以下结果之一

{
    'action': 'stat_account',
    'success': True,
    'items': items,
    'headers': headers
}

{
    'action': 'stat_container',
    'container': <container>,
    'success': True,
    'items': items,
    'headers': headers
}

如果对对象列表调用 stat，该方法将返回一个生成器，该生成器将在线程池上执行时返回单个对象 stat 操作的结果

{
    'action': 'stat_object',
    'object': <object_name>,
    'container': <container>,
    'success': True,
    'items': items,
    'headers': headers
}

如果操作失败，返回的字典将指示操作不成功，并将包含以下键

{
    'action': <'stat_object'|'stat_container'|'stat_account'>,
    'object': <'object_name'>,      # Only for stat with objects list
    'container': <container>,       # Only for stat with objects list or container
    'success': False,
    'error': <error>,
    'traceback': <trace>,
    'error_timestamp': <timestamp>
}

示例

以下代码演示了使用 stat 获取容器中给定对象列表的标头，使用 20 个线程。该代码创建从对象名称到标头的映射，然后将其美观地打印到日志中。

import logging
import pprint

from swiftclient.service import SwiftService
from sys import argv

logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

_opts = {'object_dd_threads': 20}
with SwiftService(options=_opts) as swift:
    container = argv[1]
    objects = argv[2:]
    header_data = {}
    stats_it = swift.stat(container=container, objects=objects)
    for stat_res in stats_it:
        if stat_res['success']:
            header_data[stat_res['object']] = stat_res['headers']
        else:
            logger.error(
                'Failed to retrieve stats for %s' % stat_res['object']
            )
    pprint.pprint(header_data)

列出

List 可以针对帐户或容器调用，以检索其中包含的容器或对象。每次调用都会返回一个迭代器，该迭代器返回结果页面（默认情况下，每个页面最多 10000 个结果）。

请参阅 swiftclient.service.SwiftService.list 以获取从方法文档字符串生成的文档。

如果给定的容器或帐户不存在，list 方法将引发 SwiftError，但对于所有其他成功/失败情况，将返回一个字典。每个成功列出的页面都将返回如下所述的字典

{
    'action': <'list_account_part'|'list_container_part'>,
    'container': <container>,      # Only for listing a container
    'prefix': <prefix>,            # The prefix of returned objects/containers
    'success': True,
    'listing': [Item],             # A list of results
                                   # (only in the event of success)
    'marker': <marker>             # The last item name in the list
                                   # (only in the event of success)
}

其中一个项目包含以下键

{
    'name': <name>,
    'bytes': 10485760,
    'last_modified': '2014-12-11T12:02:38.774540',
    'hash': 'fb938269cbeabe4c234e1127bbd3b74a',
    'content_type': 'application/octet-stream',
    'meta': <metadata>    # Full metadata listing from stat'ing each object
                          # this key only exists if 'long' is specified in options
}

任何列出存在帐户或容器的失败操作都将返回如下所述的失败字典

{
    'action': <'list_account_part'|'list_container_part'>,,
    'container': container,         # Only for listing a container
    'prefix': options['prefix'],
    'success': success,
    'marker': marker,
    'error': error,
    'traceback': <trace>,
    'error_timestamp': <timestamp>
}

示例

以下代码演示了使用 list 列出容器中所有大于 10MiB 的项目

import logging

from swiftclient.service import SwiftService, SwiftError
from sys import argv

logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

container = argv[1]
minimum_size = 10*1024**2
with SwiftService() as swift:
    try:
        list_parts_gen = swift.list(container=container)
        for page in list_parts_gen:
            if page["success"]:
                for item in page["listing"]:

                    i_size = int(item["bytes"])
                    if i_size > minimum_size:
                        i_name = item["name"]
                        i_etag = item["hash"]
                        print(
                            "%s [size: %s] [etag: %s]" %
                            (i_name, i_size, i_etag)
                        )
            else:
                raise page["error"]

    except SwiftError as e:
        logger.error(e.value)

Post

Post 可以针对帐户、容器或对象列表调用，以更新给定项目的附加元数据。在前两种情况下，将返回一个包含操作结果的单个字典，如果提供了对象列表，则返回一个迭代器，该迭代器为每个对象 post 生成结果。

对象列表的每个元素可以是对象名称的纯字符串，也可以是 SwiftPostObject，它允许更精细地控制应用于每个单个 post 操作的选项和元数据。当为对象名称提供字符串时，应用的选项和元数据是提供给 post() 调用以及 SwiftService 对象的默认值的组合。

如果给定的容器或帐户不存在，post 方法将引发 SwiftError。成功的元数据更新结果是如下所述的字典

{
    'action': <'post_account'|'post_container'|'post_object'>,
    'success': True,
    'container': <container>,
    'object': <object>,
    'headers': {},
    'response_dict': <HTTP response details>
}

注意

更新用户元数据键不仅会添加任何指定的键，还会删除先前设置的用户元数据。这意味着每次更新用户元数据时，都必须指定完整的所需键值对。

示例

以下代码演示了使用 post 在给定容器中设置一个存档文件夹，在 24 小时延迟后过期

import logging

from swiftclient.service import SwiftService, SwiftError
from sys import argv

logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

container = argv[1]
with SwiftService() as swift:
    try:
        list_options = {"prefix": "archive_2016-01-01/"}
        list_parts_gen = swift.list(container=container)
        for page in list_parts_gen:
            if page["success"]:
                objects = [obj["name"] for obj in page["listing"]]
                post_options = {"header": "X-Delete-After:86400"}
                for post_res in swift.post(
                        container=container,
                        objects=objects,
                        options=post_options):
                    if post_res['success']:
                        print("Object '%s' POST success" % post_res['object'])
                    else:
                        print("Object '%s' POST failed" % post_res['object'])
            else:
                raise page["error"]
    except SwiftError as e:
        logger.error(e.value)

下载

Download 可以针对整个帐户、单个容器或给定容器中的对象列表调用。对象列表的每个元素都是一个字符串，详细说明要下载的对象的完整名称。

为了下载整个帐户的完整内容，必须将 options 字典中 yes_all 的值设置为 True，该字典提供给 SwiftService 实例或对 download 的调用。

如果给定的容器或帐户不存在，download 方法将引发 SwiftError，否则将返回一个迭代器，该迭代器为每个对象下载生成结果。

请参阅 swiftclient.service.SwiftService.download 以获取从方法文档字符串生成的文档。

对于每个成功下载的对象，迭代器返回的结果将是如下所述的字典（不返回已完成的容器或对象段下载的结果）

{
    'action': 'download_object',
    'container': <container>,
    'object': <object name>,
    'success': True,
    'path': <local path to downloaded object>,
    'pseudodir': <if true, the download created an empty directory>,
    'start_time': <time download started>,
    'end_time': <time download completed>,
    'headers_receipt': <time the headers from the object were retrieved>,
    'auth_end_time': <time authentication completed>,
    'read_length': <bytes_read>,
    'attempts': <attempt count>,
    'response_dict': <HTTP response details>
}

任何上传对象失败都将返回如下所述的失败字典

{
    'action': 'download_object',
    'container': <container>,
    'object': <object name>,
    'success': False,
    'path': <local path of the failed download>,
    'pseudodir': <if true, the failed download was an empty directory>,
    'attempts': <attempt count>,
    'error': <error>,
    'traceback': <trace>,
    'error_timestamp': <timestamp>,
    'response_dict': <HTTP response details>
}

示例

以下代码演示了使用 download 从给定容器中的一个日期存档文件夹下载所有 PNG 图像

import logging

from swiftclient.service import SwiftService, SwiftError
from sys import argv

logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

def is_png(obj):
    return (
        obj["name"].lower().endswith('.png') or
        obj["content_type"] == 'image/png'
    )

container = argv[1]
with SwiftService() as swift:
    try:
        list_options = {"prefix": "archive_2016-01-01/"}
        list_parts_gen = swift.list(container=container)
        for page in list_parts_gen:
            if page["success"]:
                objects = [
                    obj["name"] for obj in page["listing"] if is_png(obj)
                ]
                for down_res in swift.download(
                        container=container,
                        objects=objects):
                    if down_res['success']:
                        print("'%s' downloaded" % down_res['object'])
                    else:
                        print("'%s' download failed" % down_res['object'])
            else:
                raise page["error"]
    except SwiftError as e:
        logger.error(e.value)

Upload

Upload 始终针对帐户和容器调用，并使用要上传的对象列表。对象列表的每个元素可以是详细说明要上传的对象的路径的纯字符串，也可以是 SwiftUploadObject，它允许更精细地控制各个操作的某些方面。

当提供一个简单的字符串来指定要上传的文件时，上传对象的名称是指定文件的完整路径，用于上传的选项是提供给 upload调用的选项。

构造一个 SwiftUploadObject 允许用户为上传的文件提供对象名称，并以单个文件的粒度修改 upload 使用的选项。

如果给定的容器或帐户不存在，upload 方法将引发 SwiftError，否则将返回一个迭代器，该迭代器为每个对象上传生成结果。

请参阅 swiftclient.service.SwiftService.upload 以获取从方法文档字符串生成的文档。

对于每个成功上传的对象（或对象段），迭代器返回的结果将是如下所述的字典

{
    'action': 'upload_object',
    'container': <container>,
    'object': <object name>,
    'success': True,
    'status': <'uploaded'|'skipped-identical'|'skipped-changed'>,
    'attempts': <attempt count>,
    'response_dict': <HTTP response details>
}

{
    'action': 'upload_segment',
    'for_container': <container>,
    'for_object': <object name>,
    'segment_index': <segment_index>,
    'segment_size': <segment_size>,
    'segment_location': <segment_path>
    'segment_etag': <etag>,
    'log_line': <object segment n>
    'success': True,
    'response_dict': <HTTP response details>,
    'attempts': <attempt count>
}

任何上传对象失败都将返回如下所述的失败字典

{
    'action': 'upload_object',
    'container': <container>,
    'object': <object name>,
    'success': False,
    'attempts': <attempt count>,
    'error': <error>,
    'traceback': <trace>,
    'error_timestamp': <timestamp>,
    'response_dict': <HTTP response details>
}

{
    'action': 'upload_segment',
    'for_container': <container>,
    'for_object': <object name>,
    'segment_index': <segment_index>,
    'segment_size': <segment_size>,
    'segment_location': <segment_path>,
    'log_line': <object segment n>,
    'success': False,
    'error': <error>,
    'traceback': <trace>,
    'error_timestamp': <timestamp>,
    'response_dict': <HTTP response details>,
    'attempts': <attempt count>
}

示例

以下代码演示了使用 upload 上传给定目录中的所有文件和文件夹，并将每个对象重命名为替换根目录名称为“my-<d>-objects”，其中 <d> 是上传目录的名称

import logging

from os import walk
from os.path import join
from swiftclient.multithreading import OutputManager
from swiftclient.service import SwiftError, SwiftService, SwiftUploadObject
from sys import argv

logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

_opts = {'object_uu_threads': 20}
dir = argv[1]
container = argv[2]
with SwiftService(options=_opts) as swift, OutputManager() as out_manager:
    try:
        # Collect all the files and folders in the given directory
        objs = []
        dir_markers = []
        for (_dir, _ds, _fs) in walk(dir):
            if not (_ds + _fs):
                dir_markers.append(_dir)
            else:
                objs.extend([join(_dir, _f) for _f in _fs])

        # Now that we've collected all the required files and dir markers
        # build the ``SwiftUploadObject``s for the call to upload
        objs = [
            SwiftUploadObject(
                o, object_name=o.replace(
                    dir, 'my-%s-objects' % dir, 1
                )
            ) for o in objs
        ]
        dir_markers = [
            SwiftUploadObject(
                None, object_name=d.replace(
                    dir, 'my-%s-objects' % dir, 1
                ), options={'dir_marker': True}
            ) for d in dir_markers
        ]

        # Schedule uploads on the SwiftService thread pool and iterate
        # over the results
        for r in swift.upload(container, objs + dir_markers):
            if r['success']:
                if 'object' in r:
                    print(r['object'])
                elif 'for_object' in r:
                    print(
                        '%s segment %s' % (r['for_object'],
                                           r['segment_index'])
                        )
            else:
                error = r['error']
                if r['action'] == "create_container":
                    logger.warning(
                        'Warning: failed to create container '
                        "'%s'%s", container, error
                    )
                elif r['action'] == "upload_object":
                    logger.error(
                        "Failed to upload object %s to container %s: %s" %
                        (container, r['object'], error)
                    )
                else:
                    logger.error("%s" % error)

    except SwiftError as e:
        logger.error(e.value)

删除

Delete 可以针对帐户或容器调用，以删除其中包含的容器或对象。每次对 delete 的调用都会返回一个迭代器，该迭代器返回每个结果子请求的结果。

如果请求的删除操作数量很大，并且目标 swift 集群正在运行批量中间件，则对 SwiftService.delete 的调用将使用批量操作，并且返回的结果迭代器将返回 bulk_delete 结果，而不是单个 delete_object、delete_container 或 delete_segment 结果。

请参阅 swiftclient.service.SwiftService.delete 以获取从方法文档字符串生成的文档。

对于每个成功删除的容器、对象或段，迭代器返回的结果将是如下所述的字典

{
    'action': <'delete_object'|'delete_segment'>,
    'container': <container>,
    'object': <object name>,
    'success': True,
    'attempts': <attempt count>,
    'response_dict': <HTTP response details>
}

{
    'action': 'delete_container',
    'container': <container>,
    'success': True,
    'response_dict': <HTTP response details>,
    'attempts': <attempt count>
}

{
    'action': 'bulk_delete',
    'container': <container>,
    'objects': <[objects]>,
    'success': True,
    'attempts': <attempt count>,
    'response_dict': <HTTP response details>
}

删除操作中的任何失败都将返回如下所述的失败字典

{
    'action': ('delete_object'|'delete_segment'),
    'container': <container>,
    'object': <object name>,
    'success': False,
    'attempts': <attempt count>,
    'error': <error>,
    'traceback': <trace>,
    'error_timestamp': <timestamp>,
    'response_dict': <HTTP response details>
}

{
    'action': 'delete_container',
    'container': <container>,
    'success': False,
    'error': <error>,
    'traceback': <trace>,
    'error_timestamp': <timestamp>,
    'response_dict': <HTTP response details>,
    'attempts': <attempt count>
}

{
    'action': 'bulk_delete',
    'container': <container>,
    'objects': <[objects]>,
    'success': False,
    'attempts': <attempt count>,
    'error': <error>,
    'traceback': <trace>,
    'error_timestamp': <timestamp>,
    'response_dict': <HTTP response details>
}

示例

以下代码演示了使用 delete 从指定容器中删除给定对象列表。当删除对象时，相关请求的事务 ID 以及对象名称和所需的尝试次数将打印出来。通过打印事务 ID，打印的操作可以轻松地链接到 swift 服务器日志中的事件

import logging

from swiftclient.service import SwiftService
from sys import argv


logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

_opts = {'object_dd_threads': 20}
container = argv[1]
objects = argv[2:]
with SwiftService(options=_opts) as swift:
    del_iter = swift.delete(container=container, objects=objects)
    for del_res in del_iter:
        c = del_res.get('container', '')
        o = del_res.get('object', '')
        a = del_res.get('attempts')
        if del_res['success'] and not del_res['action'] == 'bulk_delete':
            rd = del_res.get('response_dict')
            if rd is not None:
                t = dict(rd.get('headers', {}))
                if t:
                    print(
                        'Successfully deleted {0}/{1} in {2} attempts '
                        '(transaction id: {3})'.format(c, o, a, t)
                    )
                else:
                    print(
                        'Successfully deleted {0}/{1} in {2} '
                        'attempts'.format(c, o, a)
                    )

Copy

Copy 可以调用来复制对象或更新给定项目的元数据。

对象列表的每个元素可以是对象名称的纯字符串，也可以是 SwiftCopyObject，它允许更精细地控制应用于每个单个复制操作的选项（目标、fresh_metadata、options）。

目标应采用 /container/object 格式；如果未设置，则对象将被复制到自身。Fresh_metadata 设置元数据操作模式。如果未设置，则当前对象用户元数据将被复制/保留；如果设置，则所有当前用户元数据都将被删除。

返回一个迭代器，该迭代器为每个对象复制生成结果（也可能包括创建目标容器的结果）。

当为对象名称提供字符串时，目标和 fresh 元数据将默认为 None 和 None，这会导致将元数据添加到现有对象。

成功的复制结果是如下所述的字典

{
    'action': 'copy_object',
    'success': True,
    'container': <container>,
    'object': <object>,
    'destination': <destination>,
    'headers': {},
    'fresh_metadata': <boolean>,
    'response_dict': <HTTP response details>
}

复制操作中的任何失败都将返回如下所述的失败字典

{
    'action': 'copy_object',
    'success': False,
    'container': <container>,
    'object': <object>,
    'destination': <destination>,
    'headers': {},
    'fresh_metadata': <boolean>,
    'response_dict': <HTTP response details>,
    'error': <error>,
    'traceback': <traceback>,
    'error_timestamp': <timestamp>
}

示例

以下代码演示了使用 copy 为对象 a 和 b 添加新的用户元数据，并将对象 c 复制到 d（添加元数据）。

import logging

from swiftclient.service import SwiftService, SwiftCopyObject, SwiftError

logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

with SwiftService() as swift:
    try:
        obj = SwiftCopyObject("c", {"destination": "/cont/d"})
        for i in swift.copy(
                "cont", ["a", "b", obj],
                {"meta": ["foo:bar"], "destination": "/cc"}):
            if i["success"]:
                if i["action"] == "copy_object":
                    print(
                        "object %s copied from /%s/%s" %
                        (i["destination"], i["container"], i["object"])
                    )
                elif i["action"] == "create_container":
                    print(
                        "container %s created" % i["container"]
                    )
            else:
                if "error" in i and isinstance(i["error"], Exception):
                    raise i["error"]
    except SwiftError as e:
        logger.error(e.value)

功能

Capabilities 可以针对帐户或特定的代理 URL 调用，以确定 swift 集群的功能。这些功能包括有关配置选项和安装在代理管道中的中间件的详细信息。

请参阅 swiftclient.service.SwiftService.capabilities 以获取从方法文档字符串生成的文档。

对于每次成功调用 list capabilities，将返回一个结果字典，其中包含如下所述的内容

{
    'action': 'capabilities',
    'timestamp': <time of the call>,
    'success': True,
    'capabilities': <dictionary containing capability details>
}

capabilities 字典的内容在 swift 键下包含核心 swift capabilities；所有其他键显示部署在代理管道中的其他中间件的配置选项。以下是一个 capabilities 字典的示例

{
    'account_quotas': {},
    'bulk_delete': {
        'max_deletes_per_request': 10000,
        'max_failed_deletes': 1000
    },
    'bulk_upload': {
        'max_containers_per_extraction': 10000,
        'max_failed_extractions': 1000
    },
    'container_quotas': {},
    'container_sync': {'realms': {}},
    'formpost': {},
    'keystoneauth': {},
    'slo': {
        'max_manifest_segments': 1000,
        'max_manifest_size': 2097152,
        'min_segment_size': 1048576
    },
    'swift': {
        'account_autocreate': True,
        'account_listing_limit': 10000,
        'allow_account_management': True,
        'container_listing_limit': 10000,
        'extra_header_count': 0,
        'max_account_name_length': 256,
        'max_container_name_length': 256,
        'max_file_size': 5368709122,
        'max_header_size': 8192,
        'max_meta_count': 90,
        'max_meta_name_length': 128,
        'max_meta_overall_size': 4096,
        'max_meta_value_length': 256,
        'max_object_name_length': 1024,
        'policies': [
            {'default': True, 'name': 'Policy-0'}
        ],
        'strict_cors_mode': False,
        'version': '2.2.2'
    },
    'tempurl': {
        'methods': ['GET', 'HEAD', 'PUT']
    }
}

示例

以下代码演示了使用 capabilities 确定 Swift 集群是否支持静态大对象，如果支持，则确定单个清单文件中可以描述的最大段数以及这些对象的大小限制

import logging

from swiftclient.exceptions import ClientException
from swiftclient.service import SwiftService

logging.basicConfig(level=logging.ERROR)
logging.getLogger("requests").setLevel(logging.CRITICAL)
logging.getLogger("swiftclient").setLevel(logging.CRITICAL)
logger = logging.getLogger(__name__)

with SwiftService() as swift:
    try:
        capabilities_result = swift.capabilities()
        capabilities = capabilities_result['capabilities']
        if 'slo' in capabilities:
            print('SLO is supported')
        else:
            print('SLO is not supported')
    except ClientException as e:
        logger.error(e.value)