minio-py/minio/api.py at master · s-t-e-v-e-n-k/minio-py

Repository navigation

10BC8

# -*- coding: utf-8 -*-

# MinIO Python Library for Amazon S3 Compatible Cloud Storage, (C)

# 2015, 2016, 2017 MinIO, Inc.

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

# http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

# pylint: disable=too-many-lines

"""

minio.api

~~~~~~~~~~~~

This module implements the API.

:license: Apache 2.0, see LICENSE for more details.

"""

from __future__ import absolute_import

import io

import itertools

import json

import os

import platform

from datetime import datetime, timedelta

from threading import Thread

import certifi

import dateutil.parser

import urllib3

from . import __title__, __version__

from .compat import range # pylint: disable=redefined-builtin

from .compat import basestring, quote, urlencode, urlsplit

from .credentials import Chain, Credentials, EnvAWS, EnvMinio, Static

from .definitions import Object, UploadPart

from .error import (AccessDenied, InvalidArgumentError, InvalidSizeError,

InvalidXMLError, NoSuchBucket, ResponseError)

from .fold_case_dict import FoldCaseDict

from .helpers import (DEFAULT_PART_SIZE, MAX_MULTIPART_COUNT, MAX_PART_SIZE,

MAX_POOL_SIZE, MIN_PART_SIZE, amzprefix_user_metadata,

check_bucket_name, check_non_empty_string, check_sse,

check_ssec, dump_http, get_md5_base64digest,

get_s3_region_from_endpoint, get_scheme_host,

get_sha256_hexdigest, get_target_url, is_amz_header,

is_supported_header, is_valid_endpoint,

is_valid_notification_config, is_valid_policy_type,

mkdir_p, optimal_part_info, read_full)

from .parsers import (parse_assume_role, parse_copy_object,

parse_get_bucket_notification, parse_list_buckets,

parse_list_multipart_uploads, parse_list_object_versions,

parse_list_objects, parse_list_objects_v2,

parse_list_parts, parse_location_constraint,

parse_multi_delete_response,

parse_multipart_upload_result,

parse_new_multipart_upload)

from .select import SelectObjectReader

from .signer import (_SIGN_V4_ALGORITHM, _UNSIGNED_PAYLOAD,

generate_credential_string, post_presign_signature,

presign_v4, sign_v4)

from .thread_pool import ThreadPool

from .xml_marshal import (marshal_bucket_notifications,

marshal_complete_multipart,

xml_marshal_bucket_constraint,

xml_marshal_bucket_encryption,

xml_marshal_delete_objects, xml_marshal_select,

xml_to_dict)

try:

from json.decoder import JSONDecodeError

except ImportError:

JSONDecodeError = ValueError

_DEFAULT_USER_AGENT = "MinIO ({os}; {arch}) {lib}/{ver}".format(

os=platform.system(), arch=platform.machine(),

lib=__title__, ver=__version__,

)

# Duration of 7 days in seconds

_MAX_EXPIRY_TIME = 604800 # 7 days in seconds

# Number of parallel workers which upload parts

_PARALLEL_UPLOADERS = 3

class Minio: # pylint: disable=too-many-public-methods

"""

Simple Storage Service (aka S3) client to perform bucket and object

operations.

:param endpoint: Hostname of a S3 service.

:param access_key: Access key (aka user ID) of your account in S3 service.

:param secret_key: Secret Key (aka password) of your account in S3 service.

:param session_token: Session token of your account in S3 service.

:param secure: Flag to indicate to use secure (TLS) connection to S3

service or not.

:param region: Region name of buckets in S3 service.

:param http_client: Customized HTTP client.

:param credentials: Credentials of your account in S3 service.

:return: :class:`Minio <Minio>` object

Example::

client = Minio('play.min.io')

client = Minio('s3.amazonaws.com', 'ACCESS_KEY', 'SECRET_KEY')

client = Minio('play.min.io', 'ACCESS_KEY', 'SECRET_KEY',

region='us-east-1')

**NOTE on concurrent usage:** The `Minio` object is thread safe when using

the Python `threading` library. Specifically, it is **NOT** safe to share

it between multiple processes, for example when using

`multiprocessing.Pool`. The solution is simply to create a new `Minio`

object in each process, and not share it between processes.

"""

# pylint: disable=too-many-function-args

def __init__(self, endpoint, access_key=None,

secret_key=None,

session_token=None,

secure=True,

region=None,

http_client=None,

credentials=None):

# Validate endpoint.

is_valid_endpoint(endpoint)

# Validate http client has correct base class.

if http_client and not isinstance(

http_client,

urllib3.poolmanager.PoolManager):

raise InvalidArgumentError(

'HTTP client should be of instance'

' `urllib3.poolmanager.PoolManager`'

)

# Default is a secured connection.

scheme = 'https://' if secure else 'http://'

self._region = region or get_s3_region_from_endpoint(endpoint)

self._region_map = dict()

self._endpoint_url = urlsplit(scheme + endpoint).geturl()

self._is_ssl = secure

self._access_key = access_key

self._secret_key = secret_key

self._session_token = session_token

self._user_agent = _DEFAULT_USER_AGENT

self._trace_output_stream = None

self._enable_s3_accelerate = False

self._accelerate_endpoint_url = scheme + 's3-accelerate.amazonaws.com'

self._credentials = credentials or Credentials(

provider=Chain(

providers=[

Static(access_key, secret_key, session_token),

EnvAWS(),

EnvMinio(),

]

)

# Load CA certificates from SSL_CERT_FILE file if set

ca_certs = os.environ.get('SSL_CERT_FILE') or certifi.where()

self._http = http_client or urllib3.PoolManager(

timeout=urllib3.Timeout.DEFAULT_TIMEOUT,

maxsize=MAX_POOL_SIZE,

cert_reqs='CERT_REQUIRED',

ca_certs=ca_certs,

retries=urllib3.Retry(

total=5,

backoff_factor=0.2,

status_forcelist=[500, 502, 503, 504]

)

def set_app_info(self, app_name, app_version):

"""

Set your application name and version to user agent header.

:param app_name: Application name.

:param app_version: Application version.

Example::

client.set_app_info('my_app', '1.0.2')

"""

if not (app_name and app_version):

raise ValueError("Application name/version cannot be empty.")

self._user_agent = "{0} {1}/{2}".format(

_DEFAULT_USER_AGENT, app_name, app_version,

)

def trace_on(self, stream):

"""

Enable http trace.

:param output_stream: Stream for writing HTTP call tracing.

"""

if not stream:

raise ValueError('Input stream for trace output is invalid.')

# Save new output stream.

self._trace_output_stream = stream

def trace_off(self):

"""

Disable HTTP trace.

"""

self._trace_output_stream = None

def use_s3_accelerate(self, value):

"""Enable AWS S3 accelerated endpoint."""

_, host = get_scheme_host(urlsplit(self._endpoint_url))

if 's3.amazonaws.com' in host:

self._enable_s3_accelerate = value is True

def select_object_content(self, bucket_name, object_name, opts):

"""

Select content of an object by SQL expression.

:param bucket_name: Name of the bucket.

:param object_name: Object name in the bucket.

:param opts: Options for select object.

:return: A reader contains requested records and progress information.

Example::

data = client.select_object_content('foo', 'test.csv', options)

"""

check_bucket_name(bucket_name)

check_non_empty_string(object_name)

content = xml_marshal_select(opts)

url_values = {

"select": "",

"select-type": "2",

}

headers = {

'Content-Length': str(len(content)),

'Content-Md5': get_md5_base64digest(content)

}

content_sha256_hex = get_sha256_hexdigest(content)

response = self._url_open(

'POST',

bucket_name=bucket_name,

object_name=object_name,

query=url_values,

headers=headers,

body=content,

content_sha256=content_sha256_hex,

preload_content=False)

return SelectObjectReader(response)

def make_bucket(self, bucket_name, location='us-east-1',

object_lock=False):

"""

Create a bucket with region and object lock.

:param bucket_name: Name of the bucket.

:param location: Region in which the bucket will be created.

:param object_lock: Flag to set object-lock feature.

Examples::

minio.make_bucket('foo')

minio.make_bucket('foo', 'us-west-1')

minio.make_bucket('foo', 'us-west-1', object_lock=True)

"""

check_bucket_name(bucket_name, True)

# Default region for all requests.

region = self._region or 'us-east-1'

# Validate if caller requested bucket location is same as current

# region

if self._region and self._region != location:

raise InvalidArgumentError(

"Configured region {0}, requested {1}".format(

self._region, location))

method = 'PUT'

# Set user agent once before the request.

headers = {'User-Agent': self._user_agent}

if object_lock:

headers["x-amz-bucket-object-lock-enabled"] = "true"

content = None

if location and location != 'us-east-1':

content = xml_marshal_bucket_constraint(location)

headers['Content-Length'] = str(len(content))

headers['Content-Md5'] = get_md5_base64digest(content)

content_sha256_hex = get_sha256_hexdigest(content)

# In case of Amazon S3. The make bucket issued on already

# existing bucket would fail with 'AuthorizationMalformed'

# error if virtual style is used. So we default to 'path

# style' as that is the preferred method here. The final

# location of the 'bucket' is provided through XML

# LocationConstraint data with the request.

# Construct target url.

url = self._endpoint_url + '/' + bucket_name + '/'

# Get signature headers if any.

headers = sign_v4(method, url, region,

headers,

self._credentials,

content_sha256_hex,

datetime.utcnow())

if self._trace_output_stream:

dump_http(method, url, headers, None,

self._trace_output_stream)

response = self._http.urlopen(method, url,

body=content,

headers=headers)

if self._trace_output_stream:

dump_http(method, url, headers, response,

self._trace_output_stream)

if response.status != 200:

raise ResponseError(response, method, bucket_name).get_exception()

self._set_bucket_region(bucket_name, region=location)

def list_buckets(self):

"""

List information of all accessible buckets.

:return: An iterator contains bucket information.

Example::

bucket_list = minio.list_buckets()

for bucket in bucket_list:

print(bucket.name, bucket.created_date)

"""

method = 'GET'

url = get_target_url(self._endpoint_url)

# Set user agent once before the request.

headers = {'User-Agent': self._user_agent}

# default for all requests.

region = self._region or 'us-east-1'

# Get signature headers if any.

headers = sign_v4(method, url, region,

headers,

self._credentials,

None, datetime.utcnow())

if self._trace_output_stream:

dump_http(method, url, headers, None,

self._trace_output_stream)

response = self._http.urlopen(method, url,

body=None,

headers=headers)

if self._trace_output_stream:

dump_http(method, url, headers, response,

self._trace_output_stream)

if response.status != 200:

raise ResponseError(response, method).get_exception()

try:

return parse_list_buckets(response.data)

except InvalidXMLError:

if self._endpoint_url.endswith("s3.amazonaws.com") and (

not self._access_key or not self._secret_key):

raise AccessDenied(response)

def bucket_exists(self, bucket_name):

"""

Check if a bucket exists.

:param bucket_name: Name of the bucket.

:return: True if the bucket exists.

Example::

found = minio.bucket_exists("my-bucketname")

if found:

print("my-bucketname exists")

else:

print("my-bucketname does not exist")

"""

check_bucket_name(bucket_name)

try:

self._url_open('HEAD', bucket_name=bucket_name)

return True

except NoSuchBucket:

# If bucket has not been created yet, MinIO returns NoSuchBucket

# error.

return False

def remove_bucket(self, bucket_name):

"""

Remove an empty bucket.

:param bucket_name: Name of the bucket.

Example::

minio.remove_bucket("my-bucketname")

"""

check_bucket_name(bucket_name)

self._url_open('DELETE', bucket_name=bucket_name)

# Make sure to purge bucket_name from region cache.

self._delete_bucket_region(bucket_name)

def get_bucket_policy(self, bucket_name):

"""

Get bucket policy configuration of a bucket.

:param bucket_name: Name of the bucket.

:return: Bucket policy configuration as JSON string.

Example::

config = minio.get_bucket_policy("my-bucketname")

"""

check_bucket_name(bucket_name)

response = self._url_open("GET",

bucket_name=bucket_name,

query={"policy": ""})

return response.data

def delete_bucket_policy(self, bucket_name):

"""

Delete bucket policy configuration of a bucket.

:param bucket_name: Name of the bucket.

Example::

minio.delete_bucket_policy("my-bucketname")

"""

self._url_open("DELETE",

bucket_name=bucket_name,

query={"policy": ""})

def set_bucket_policy(self, bucket_name, policy):

"""

Set bucket policy configuration to a bucket.

:param bucket_name: Name of the bucket.

:param policy: Bucket policy configuration as JSON string.

Example::

minio.get_bucket_policy("my-bucketname", config)

"""

is_valid_policy_type(policy)

check_bucket_name(bucket_name)

headers = {

'Content-Length': str(len(policy)),

'Content-Md5': get_md5_base64digest(policy)

}

content_sha256_hex = get_sha256_hexdigest(policy)

self._url_open("PUT",

bucket_name=bucket_name,

query={"policy": ""},

headers=headers,

body=policy,

content_sha256=content_sha256_hex)

def get_bucket_notification(self, bucket_name):

"""

Get notification configuration of a bucket.

:param bucket_name: Name of the bucket.

:return: Notification configuration.

Example::

config = minio.get_bucket_notification("my-bucketname")

"""

check_bucket_name(bucket_name)

response = self._url_open(

"GET",

bucket_name=bucket_name,

query={"notification": ""},

)

data = response.data.decode('utf-8')

return parse_get_bucket_notification(data)

def set_bucket_notification(self, bucket_name, notifications):

"""

Set notification configuration of a bucket.

:param bucket_name: Name of the bucket.

:param notifications: Notification configuration to be set.

Example::

minio.set_bucket_notification("my-bucketname", config)

"""

check_bucket_name(bucket_name)

is_valid_notification_config(notifications)

content = marshal_bucket_notifications(notifications)

headers = {

'Content-Length': str(len(content)),

'Content-Md5': get_md5_base64digest(content)

}

content_sha256_hex = get_sha256_hexdigest(content)

self._url_open(

'PUT',

bucket_name=bucket_name,

query={"notification": ""},

headers=headers,

body=content,

content_sha256=content_sha256_hex

)

def remove_all_bucket_notification(self, bucket_name):

"""

Remove notification configuration of a bucket. On success, S3 service

stops notification of events previously set of the bucket.

:param bucket_name: Name of the bucket.

Example::

minio.remove_all_bucket_notification("my-bucketname")

"""

check_bucket_name(bucket_name)

content_bytes = marshal_bucket_notifications({})

headers = {

'Content-Length': str(len(content_bytes)),

'Content-Md5': get_md5_base64digest(content_bytes)

}

content_sha256_hex = get_sha256_hexdigest(content_bytes)

self._url_open(

'PUT',

bucket_name=bucket_name,

query={"notification": ""},

headers=headers,

body=content_bytes,

content_sha256=content_sha256_hex

)

def put_bucket_encryption(self, bucket_name, enc_config):

"""

Set encryption configuration of a bucket.

:param bucket_name: Name of the bucket.

:param enc_config: Encryption configuration as dictionary to be set.

Example::

minio.put_bucket_encryption("my-bucketname", config)

"""

check_bucket_name(bucket_name)

# 'Rule' is a list, so we need to go through each one of

# its key/value pair and collect the encryption values.

rules = enc_config['ServerSideEncryptionConfiguration']['Rule']

enc_config_xml = xml_marshal_bucket_encryption(rules)

headers = {

'Content-Length': str(len(enc_config_xml)),

'Content-Md5': get_md5_base64digest(enc_config_xml)

}

content_sha256_hex = get_sha256_hexdigest(enc_config_xml)

10BC8

self._url_open("PUT",

bucket_name=bucket_name,

query={"encryption": ""},

headers=headers,

body=enc_config_xml,

content_sha256=content_sha256_hex

)

def get_bucket_encryption(self, bucket_name):

"""

Get encryption configuration of a bucket.

:param bucket_name: Name of the bucket.

:return: Encryption configuration.

Example::

config = minio.get_bucket_encryption("my-bucketname")

"""

check_bucket_name(bucket_name)

response = self._url_open(

"GET",

bucket_name=bucket_name,

query={"encryption": ""}

)

return xml_to_dict(response.data.decode('utf-8'))

def delete_bucket_encryption(self, bucket_name):

"""

Delete encryption configuration of a bucket.

:param bucket_name: Name of the bucket.

Example::

minio.delete_bucket_encryption("my-bucketname")

"""

check_bucket_name(bucket_name)

self._url_open(

'DELETE',

bucket_name=bucket_name,

query={"encryption": ""}

)

def listen_bucket_notification(self, bucket_name, prefix='', suffix='',

events=('s3:ObjectCreated:*',

's3:ObjectRemoved:*',

's3:ObjectAccessed:*')):

"""

Listen events of object prefix and suffix of a bucket. Caller should

iterate returned iterator to read new events.

:param bucket_name: Name of the bucket.

:param prefix: Listen events of object starts with prefix.

:param suffix: Listen events of object ends with suffix.

:param events: Events to listen.

:return: Iterator contains event records.

Example::

iter = minio.listen_bucket_notification(

"my-bucketname",

events=('s3:ObjectCreated:*', 's3:ObjectAccessed:*'),

)

for events in iter:

print(events)

"""

check_bucket_name(bucket_name)

# If someone explicitly set prefix to None convert it to empty string.

prefix = prefix or ''

# If someone explicitly set suffix to None convert it to empty string.

suffix = suffix or ''

url_components = urlsplit(self._endpoint_url)

if url_components.hostname == 's3.amazonaws.com':

raise InvalidArgumentError(

'Listening for event notifications on a bucket is a MinIO '

'specific extension to bucket notification API. It is not '

'supported by Amazon S3')

query = {

'prefix': prefix,

'suffix': suffix,

'events': events,

}

while True:

response = self._url_open('GET', bucket_name=bucket_name,

query=query, preload_content=False)

try:

for line in response.stream():

if line.strip():

if hasattr(line, 'decode'):

line = line.decode('utf-8')

event = json.loads(line)

if event['Records']:

yield event

except JSONDecodeError:

response.close()

continue

def _do_bucket_versioning(self, bucket_name, status):

"""Do versioning support in a bucket."""

check_bucket_name(bucket_name)

config = (

'<VersioningConfiguration '

'xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'

'<Status>{0}</Status></VersioningConfiguration>'

).format("Enabled" if status else "Suspended")

body = config.encode()

headers = {

'Content-Length': str(len(body)),

'Content-Md5': get_md5_base64digest(body)

}

content_sha256_hex = get_sha256_hexdigest(body)

self._url_open(

"PUT",

bucket_name=bucket_name,

query={"versioning": ""},

headers=headers,

body=body,

content_sha256=content_sha256_hex,

)

def enable_bucket_versioning(self, bucket_name):

"""

Enable object versioning feature in a bucket.

:param bucket_name: Name of the bucket.

Example::

minio.enable_bucket_versioning("my-bucketname")

"""

self._do_bucket_versioning(bucket_name, True)

def disable_bucket_versioning(self, bucket_name):

"""

Disable object versioning feature in a bucket.

:param bucket_name: Name of the bucket.

Example::

minio.disable_bucket_versioning("my-bucketname")

"""

self._do_bucket_versioning(bucket_name, False)

def fput_object(self, bucket_name, object_name, file_path,

content_type='application/octet-stream',

metadata=None, sse=None, progress=None,

part_size=DEFAULT_PART_SIZE):

"""

Uploads data from a file to an object in a bucket.

:param bucket_name: Name of the bucket.

:param object_name: Object name in the bucket.

:param file_path: Name of file to upload.

:param content_type: Content type of the object.

:param metadata: Any additional metadata to be uploaded along

with your PUT request.

:param sse: Server-side encryption.

:param progress: A progress object

:param part_size: Multipart part size

:return: etag and version ID if available.

Example::

minio.fput_object('foo', 'bar', 'filepath', 'text/plain')

"""

# Open file in 'read' mode.

with open(file_path, 'rb') as file_data:

file_size = os.stat(file_path).st_size

return self.put_object(bucket_name, object_name, file_data,

file_size, content_type, metadata, sse,

progress, part_size)

def fget_object(self, bucket_name, object_name, file_path,

request_headers=None, sse=None, version_id=None,

extra_query_params=None):

"""

Downloads data of an object to file.

:param bucket_name: Name of the bucket.

:param object_name: Object name in the bucket.

:param file_path: Name of file to upload.

:param request_headers: Any additional headers to be added with GET

request.

:param sse: Server-side encryption customer key.

:param version_id: Version-ID of the object.

:param extra_query_params: Extra query parameters for advanced usage.

:return: Object information.

Example::

minio.fget_object('foo', 'bar', 'localfile')

minio.fget_object(

'foo', 'bar', 'localfile', version_id='VERSION-ID',

)

"""

check_bucket_name(bucket_name)

check_non_empty_string(object_name)

if os.path.isdir(file_path):

raise OSError("file is a directory.")

# Create top level directory if needed.

top_level_dir = os.path.dirname(file_path)

if top_level_dir:

mkdir_p(top_level_dir)

stat = self.stat_object(

bucket_name,

object_name,

sse,

version_id=version_id,

)

# Write to a temporary file "file_path.part.minio" before saving.

file_part_path = file_path + "." + stat.etag + '.part.minio'

# Open file in 'overwrite' mode.

with open(file_part_path, 'wb') as file_part_data:

# Save current file_part statinfo.

file_statinfo = os.stat(file_part_path)

# Get partial object.

response = self.get_object(

bucket_name,

object_name,

offset=file_statinfo.st_size,

length=0,

request_headers=request_headers,

sse=sse,

version_id=version_id,

extra_query_params=extra_query_params,

)

# Save content_size to verify if we wrote more data.

content_size = int(response.headers['content-length'])

# Save total_written.

total_written = 0

for data in response.stream(amt=1024 * 1024):

file_part_data.write(data)

total_written += len(data)

# Release the connection from the response at this point.

response.release_conn()

# Verify if we wrote data properly.

if total_written < content_size:

msg = ('Data written {0} bytes is smaller than the specified'

' size {1} bytes').format(total_written, content_size)

raise InvalidSizeError(msg)

if total_written > content_size:

msg = ('Data written {0} bytes is in excess than the specified'

' size {1} bytes').format(total_written, content_size)

raise InvalidSizeError(msg)

# Delete existing file to be compatible with Windows

if os.path.exists(file_path):

os.remove(file_path)

# Rename with destination file path

os.rename(file_part_path, file_path)

# Return the stat

return stat

def get_object(self, bucket_name, object_name, offset=0, length=0,

request_headers=None, sse=None, version_id=None,

extra_query_params=None):

"""

Get data of an object. Returned response should be closed after use to

release network resources. To reuse the connection, its required to

call `response.release_conn()` explicitly.

:param bucket_name: Name of the bucket.

:param object_name: Object name in the bucket.

:param offset: Start byte position of object data.

:param length: Number of bytes of object data from offset.

:param request_headers: Any additional headers to be added with GET

request.

:param sse: Server-side encryption customer key.

:param version_id: Version-ID of the object.

:param extra_query_params: Extra query parameters for advanced usage.

:return: :class:`urllib3.response.HTTPResponse` object.

Example::

// Get entire object data.

try:

response = minio.get_object('foo', 'bar')

// Read data from response.

finally:

response.close()

response.release_conn()

// Get object data for offset/length.

try:

response = minio.get_partial_object('foo', 'bar', 2, 4)

// Read data from response.

finally:

response.close()

response.release_conn()

"""

check_bucket_name(bucket_name)

check_non_empty_string(object_name)

check_ssec(sse)

headers = sse.headers() if sse else {}

headers.update(request_headers or {})

if offset or length:

headers['Range'] = 'bytes={}-{}'.format(

offset, offset + length - 1 if length else "")

if version_id:

extra_query_params = extra_query_params or {}

extra_query_params["versionId"] = version_id

return self._url_open(

"GET",

bucket_name=bucket_name,

object_name=object_name,

headers=headers,

preload_content=False,

query=extra_query_params,

)

def copy_object(self, bucket_name, object_name, object_source,

conditions=None, source_sse=None, sse=None, metadata=None):

"""

Create an object by server-side copying data from another object.

In this API maximum supported source object size is 5GiB.

:param bucket_name: Name of the bucket.

:param object_name: Object name in the bucket.

:param object_source: Source object to be copied.

:param conditions: :class:`CopyConditions` object. Collection of

supported CopyObject conditions.

:param source_sse: Server-side encryption customer key of source

object.

:param sse: Server-side encryption of destination object.

:param metadata: Any user-defined metadata to be copied along with

destination object.

:return: :class:`CopyObjectResult`

Example::

minio.copy_object(

"my-bucketname",

"my-objectname",

"my-source-bucketname/my-source-bucketname",

)

minio.copy_object(

"my-bucketname",

"my-objectname",

"my-source-bucketname/my-source-bucketname"

"?versionId=b6602757-7c9c-449b-937f-fed504d04c94",

)

"""

check_bucket_name(bucket_name)

check_non_empty_string(object_name)

check_non_empty_string(object_source)

headers = {}

# Preserving the user-defined metadata in headers

if metadata:

headers = amzprefix_user_metadata(metadata)

headers["x-amz-metadata-directive"] = "REPLACE"

if conditions:

headers.update(conditions)

# Source sse must be SSE-C if not null.

check_ssec(source_sse)

headers.update(source_sse.copy_headers() if source_sse else {})

# Destination sse can be any Sse type if not null.

check_sse(sse)

headers.update(sse.headers() if sse else {})

headers['X-Amz-Copy-Source'] = quote(object_source)

response = self._url_open('PUT',

bucket_name=bucket_name,

object_name=object_name,

headers=headers)

return parse_copy_object(bucket_name, object_name, response.data)

def put_object(self, bucket_name, object_name, data, length,

content_type='application/octet-stream',

metadata=None, sse=None, progress=None,

part_size=DEFAULT_PART_SIZE):

"""

Uploads data from a stream to an object in a bucket.

:param bucket_name: Name of the bucket.

:param object_name: Object name in the bucket.

:param data: Contains object data.

:param content_type: Content type of the object.

:param metadata: Any additional metadata to be uploaded along

with your PUT request.

:param sse: Server-side encryption.

:param progress: A progress object

:param part_size: Multipart part size

:return: etag and version ID if available.

Example::

file_stat = os.stat('hello.txt')

with open('hello.txt', 'rb') as data:

minio.put_object(

'foo', 'bar', data, file_stat.st_size, 'text/plain',

)

"""

check_sse(sse)

check_bucket_name(bucket_name)

check_non_empty_string(object_name)

if progress:

if not isinstance(progress, Thread):

raise TypeError('Progress object should inherit the thread.')

# Set progress bar length and object name before upload

progress.set_meta(total_length=length, object_name=object_name)

View remainder of file in raw view

FilesExpand file tree

api.py

Latest commit

History

api.py

File metadata and controls