8000 minio-py/minio/api.py at master · manmandong/minio-py · GitHub
[go: up one dir, main page]
Skip to content

api.py

Latest commit

 

History

History
1905 lines (1571 loc) · 70.1 KB

api.py

File metadata and controls

1905 lines (1571 loc) · 70.1 KB
Raw
# -*- 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.
"""
minio.api
~~~~~~~~~~~~
This module implements the API.
:copyright: (c) 2015, 2016, 2017 by MinIO, Inc.
:license: Apache 2.0, see LICENSE for more details.
"""
# Standard python packages
from __future__ import absolute_import
import platform
from threading import Thread
from datetime import datetime, timedelta
import io
import json
import os
import itertools
try:
from json.decoder import JSONDecodeError
except ImportError:
JSONDecodeError = ValueError
# Dependencies
import urllib3
import certifi
import dateutil.parser
# Internal imports
from . import __title__, __version__
from .compat import (urlsplit, queryencode,
range, basestring)
from .error import (ResponseError, NoSuchBucket, AccessDenied,
InvalidArgumentError, InvalidSizeError, InvalidXMLError)
from .definitions import Object, UploadPart
from .parsers import (parse_list_buckets,
parse_list_objects,
parse_list_objects_v2,
parse_list_parts,
parse_copy_object,
parse_list_multipart_uploads,
parse_new_multipart_upload,
parse_location_constraint,
parse_multipart_upload_result,
parse_get_bucket_notification,
parse_multi_object_delete_response)
from .helpers import (get_target_url, is_non_empty_string,
is_valid_endpoint,
get_sha256_hexdigest, get_md5_base64digest,
optimal_part_info,
is_valid_bucket_name, read_full,
get_s3_region_from_endpoint, is_valid_sse_object, is_valid_sse_c_object,
is_valid_source_sse_object,
is_valid_bucket_notification_config, is_valid_policy_type,
mkdir_p, dump_http, amzprefix_user_metadata,
is_supported_header,is_amz_header)
from .helpers import (MAX_MULTIPART_OBJECT_SIZE,
MAX_PART_SIZE,
MAX_POOL_SIZE,
MIN_PART_SIZE,
DEFAULT_PART_SIZE,
MAX_MULTIPART_COUNT)
from .signer import (sign_v4, presign_v4,
generate_credential_string,
post_presign_signature)
from .signer import (_UNSIGNED_PAYLOAD, _SIGN_V4_ALGORITHM)
from .xml_marshal import (xml_marshal_bucket_constraint,
xml_marshal_complete_multipart_upload,
xml_marshal_bucket_notifications,
xml_marshal_delete_objects)
from .fold_case_dict import FoldCaseDict
from .thread_pool import ThreadPool
# Comment format.
_COMMENTS = '({0}; {1})'
# App info format.
_APP_INFO = '{0}/{1}'
# MinIO (OS; ARCH) LIB/VER APP/VER .
_DEFAULT_USER_AGENT = 'MinIO {0} {1}'.format(
_COMMENTS.format(platform.system(),
platform.machine()),
_APP_INFO.format(__title__,
__version__))
_MAX_EXPIRY_TIME = 2147483647 #
# Number of parallel workers which upload parts
_PARALLEL_UPLOADERS = 3
class Minio(object):
"""
Constructs a :class:`Minio <Minio>`.
Examples:
client = Minio('play.min.io:9000')
client = Minio('s3.amazonaws.com', 'ACCESS_KEY', 'SECRET_KEY')
# To override auto bucket location discovery.
client = Minio('play.min.io:9000', 'ACCESS_KEY', 'SECRET_KEY',
region='us-east-1')
:param endpoint: Hostname of the cloud storage server.
:param access_key: Access key to sign self._http.request with.
:param secret_key: Secret key to sign self._http.request with.
:param session_token: Session token to sign self._http.request with.
:param secure: Set this value if wish to make secure requests.
Default is True.
:param region: Set this value to override automatic bucket
location discovery.
:param timeout: Set this value to control how long requests
are allowed to run before being aborted.
:return: :class:`Minio <Minio>` object
"""
def __init__(self, endpoint, access_key=None,
secret_key=None,
session_token=None,
secure=True,
region=None,
http_client=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.
endpoint_url = 'https://' + endpoint
if not secure:
endpoint_url = 'http://' + endpoint
# Parse url endpoints.
url_components = urlsplit(endpoint_url)
# Extract region if possible from endpoint.
if not region:
region = get_s3_region_from_endpoint(endpoint)
self._region = region
self._region_map = dict()
self._endpoint_url = url_components.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
# Load CA certificates from SSL_CERT_FILE file if set
ca_certs = os.environ.get('SSL_CERT_FILE')
if not ca_certs:
ca_certs = certifi.where()
if not http_client:
self._http = 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]
)
)
else:
self._http = http_client
# Set application information.
def set_app_info(self, app_name, app_version):
"""
Sets your application name and version to
default user agent in the following format.
MinIO (OS; ARCH) LIB/VER APP/VER
Example:
client.set_app_info('my_app', '1.0.2')
:param app_name: application name.
:param app_version: application version.
"""
if not (app_name and app_version):
raise ValueError('app_name and app_version cannot be empty.')
app_info = _APP_INFO.format(app_name,
app_version)
self._user_agent = ' '.join([_DEFAULT_USER_AGENT, app_info])
# enable HTTP trace.
def trace_on(self, stream):
"""
Enable http trace.
8 10BC8 :param output_stream: Stream where trace is written to.
"""
if not stream:
raise ValueError('Input stream for trace output is invalid.')
# Save new output stream.
self._trace_output_stream = stream
# disable HTTP trace.
def trace_off(self):
"""
Disable HTTP trace.
"""
self._trace_output_stream = None
# Bucket level
def make_bucket(self, bucket_name, location='us-east-1'):
"""
Make a new bucket on the server.
Optionally include Location.
['us-east-1', 'us-east-2', 'us-west-1', 'us-west-2', 'eu-west-1',
'eu-west-2', 'ca-central-1', 'eu-central-1', 'sa-east-1',
'cn-north-1', 'ap-southeast-1', 'ap-southeast-2',
'ap-northeast-1', 'ap-northeast-2']
Examples:
minio.make_bucket('foo')
minio.make_bucket('foo', 'us-west-1')
:param bucket_name: Bucket to create on server
:param location: Location to create bucket on
"""
is_valid_bucket_name(bucket_name)
# Default region for all requests.
region = 'us-east-1'
if self._region:
region = self._region
# Validate if caller requested bucket location is same as current region
if 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}
content = None
if location and location != 'us-east-1':
content = xml_marshal_bucket_constraint(location)
headers['Content-Length'] = str(len(content))
content_sha256_hex = get_sha256_hexdigest(content)
if content:
headers['Content-Md5'] = get_md5_base64digest(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._access_key,
self._secret_key,
self._session_token,
content_sha256_hex)
response = self._http.urlopen(method, url,
body=content,
headers=headers)
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 all buckets owned by the user.
Example:
bucket_list = minio.list_buckets()
for bucket in bucket_list:
print(bucket.name, bucket.created_date)
:return: An iterator of buckets owned by the current user.
"""
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 = 'us-east-1'
# region is set then use the region.
if self._region:
region = self._region
# Get signature headers if any.
headers = sign_v4(method, url, region,
headers, self._access_key,
self._secret_key,
self._session_token,
None)
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 the bucket exists and if the user has access to it.
:param bucket_name: To test the existence and user access.
:return: True on success.
"""
is_valid_bucket_name(bucket_name)
try:
self._url_open('HEAD', bucket_name=bucket_name)
# If the bucket has not been created yet, MinIO will return a "NoSuchBucket" error.
except NoSuchBucket:
return False
except ResponseError:
raise
return True
def remove_bucket(self, bucket_name):
"""
Remove a bucket.
:param bucket_name: Bucket to remove
"""
is_valid_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 of given bucket name.
:param bucket_name: Bucket name.
"""
is_valid_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):
self._url_open("DELETE",
bucket_name=bucket_name,
query={"policy": ""})
def set_bucket_policy(self, bucket_name, policy):
"""
Set bucket policy of given bucket name.
:param bucket_name: Bucket name.
:param policy: Access policy/ies in string format.
"""
is_valid_policy_type(policy)
is_valid_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 notifications configured for the given bucket.
:param bucket_name: Bucket name.
"""
is_valid_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 the given notifications on the bucket.
:param bucket_name: Bucket name.
:param notifications: Notifications structure
"""
is_valid_bucket_name(bucket_name)
is_valid_bucket_notification_config(notifications)
content = xml_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):
"""
Removes all bucket notification configs configured
previously, this call disable event notifications
on a bucket. This operation cannot be undone, to
set notifications again you should use
``set_bucket_notification``
:param bucket_name: Bucket name.
"""
is_valid_bucket_name(bucket_name)
content_bytes = xml_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 listen_bucket_notification(self, bucket_name, prefix='', suffix='',
events=['s3:ObjectCreated:*',
's3:ObjectRemoved:*',
's3:ObjectAccessed:*']):
"""
Yeilds new event notifications on a bucket, caller should iterate
to read new notifications.
NOTE: Notification is retried in case of `JSONDecodeError` otherwise
the function raises an exception.
:param bucket_name: Bucket name to listen event notifications from.
:param prefix: Object key prefix to filter notifications for.
:param suffix: Object key suffix to filter notifications for.
:param events: Enables notifications for specific event types.
of events.
"""
is_valid_bucket_name(bucket_name)
# If someone explicitly set prefix to None convert it to empty string.
if prefix is None:
prefix = ''
# If someone explicitly set suffix to None convert it to empty string.
if suffix is None:
suffix = ''
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'] is not None:
yield event
except JSONDecodeError:
response.close()
continue
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):
"""
Add a new object to the cloud storage server.
Examples:
minio.fput_object('foo', 'bar', 'filepath', 'text/plain')
:param bucket_name: Bucket to read object from.
:param object_name: Name of the object to read.
:param file_path: Local file path to be uploaded.
:param content_type: Content type of the object.
:param metadata: Any additional metadata to be uploaded along
with your PUT request.
:param progress: A progress object
:param part_size: Multipart part size
:return: etag
"""
# 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):
"""
Retrieves an object from a bucket and writes at file_path.
Examples:
minio.fget_object('foo', 'bar', 'localfile')
:param bucket_name: Bucket to read object from.
:param object_name: Name of the object to read.
:param file_path: Local file path to save the object.
:param request_headers: Any additional headers to be added with GET request.
"""
is_valid_bucket_name(bucket_name)
is_non_empty_string(object_name)
stat = self.stat_object(bucket_name, object_name, sse)
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)
# Write to a temporary file "file_path.part.minio" before saving.
file_part_path = file_path + stat.etag + '.part.minio'
# Open file in 'write+append' mode.
with open(file_part_path, 'ab') as file_part_data:
# Save current file_part statinfo.
file_statinfo = os.stat(file_part_path)
# Get partial object.
response = self._get_partial_object(bucket_name, object_name,
offset=file_statinfo.st_size,
length=0,
request_headers=request_headers,
sse=sse)
# 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, request_headers=None, sse=None):
"""
Retrieves an object from a bucket.
This function returns an object that contains an open network
connection to enable incremental consumption of the
response. To re-use the connection (if desired) on subsequent
requests, the user needs to call `release_conn()` on the
returned object after processing.
Examples:
my_object = minio.get_partial_object('foo', 'bar')
:param bucket_name: Bucket to read object from
:param object_name: Name of object to read
:param request_headers: Any additional headers to be added with GET request.
:return: :class:`urllib3.response.HTTPResponse` object.
"""
is_valid_bucket_name(bucket_name)
is_non_empty_string(object_name)
return self._get_partial_object(bucket_name,
object_name,
request_headers=request_headers,
sse=sse)
def get_partial_object(self, bucket_name, object_name, offset=0, length=0, request_headers=None, sse=None):
"""
Retrieves an object from a bucket.
Optionally takes an offset and length of data to retrieve.
This function returns an object that contains an open network
connection to enable incremental consumption of the
response. To re-use the connection (if desired) on subsequent
requests, the user needs to call `release_conn()` on the
returned object after processing.
Examples:
partial_object = minio.get_partial_object('foo', 'bar', 2, 4)
:param bucket_name: Bucket to retrieve object from
:param object_name: Name of object to retrieve
:param offset: Optional offset to retrieve bytes from.
Must be >= 0.
:param length: Optional number of bytes to retrieve.
Must be an integer.
:param request_headers: Any additional headers to be added with GET request.
:return: :class:`urllib3.response.HTTPResponse` object.
"""
is_valid_bucket_name(bucket_name)
is_non_empty_string(object_name)
return self._get_partial_object(bucket_name,
object_name,
offset, length,
request_headers=request_headers,
sse=sse)
def copy_object(self, bucket_name, object_name, object_source,
conditions=None, source_sse=None, sse=None, metadata=None):
"""
Copy a source object on object storage server to a new object.
NOTE: Maximum object size supported by this API is 5GB.
Examples:
:param bucket_name: Bucket of new object.
:param object_name: Name of new object.
:param object_source: Source object to be copied.
:param conditions: :class:`CopyConditions` object. Collection of
supported CopyObject conditions.
:param metadata: Any user-defined metadata to be copied along with
destination object.
"""
is_valid_bucket_name(bucket_name)
is_non_empty_string(object_name)
is_non_empty_string(object_source)
headers = {}
# Preserving the user-defined metadata in headers
if metadata is not None:
headers = amzprefix_user_metadata(metadata)
headers["x-amz-metadata-directive"] = "REPLACE"
if conditions:
for k, v in conditions.items():
headers[k] = v
# Source argument to copy_object can only be of type copy_SSE_C
if source_sse:
is_valid_source_sse_object(source_sse)
headers.update(source_sse.marshal())
#Destination argument to copy_object cannot be of type copy_SSE_C
if sse:
is_valid_sse_object(sse)
headers.update(sse.marshal())
headers['X-Amz-Copy-Source'] = queryencode(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):
"""
Add a new object to the cloud storage server.
NOTE: Maximum object size supported by this API is 5TiB.
Examples:
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')
- For length lesser than 5MB put_object automatically
does single Put operation.
- For length larger than 5MB put_object automatically
does resumable multipart operation.
:param bucket_name: Bucket of new object.
:param object_name: Name of new object.
:param data: Contents to upload.
:param length: Total length of object.
:param content_type: mime type of object as a string.
:param metadata: Any additional metadata to be uploaded along
with your PUT request.
:param progress: A progress object
:param part_size: Multipart part size
:return: etag
"""
is_valid_sse_object(sse)
is_valid_bucket_name(bucket_name)
is_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)
if not callable(getattr(data, 'read')):
raise ValueError(
'Invalid input data does not implement a callable read() method')
if length > (part_size * MAX_MULTIPART_COUNT):
raise InvalidArgumentError('Part size * max_parts(10000) is '
' lesser than input length.')
if part_size < MIN_PART_SIZE:
raise InvalidArgumentError('Input part size is smaller '
' than allowed minimum of 5MiB.')
if part_size > MAX_PART_SIZE:
raise InvalidArgumentError('Input part size is bigger '
' than allowed maximum of 5GiB.')
if not metadata:
metadata = {}
metadata = amzprefix_user_metadata(metadata)
metadata['Content-Type'] = 'application/octet-stream' if \
not content_type else content_type
if length > part_size:
return self._stream_put_object(bucket_name, object_name,
data, length, metadata=metadata,
sse=sse, progress=progress,
part_size=part_size)
current_data = data.read(length)
if len(current_data) != length:
raise InvalidArgumentError(
'Could not read {} bytes from data to upload'.format(length)
)
return self._do_put_object(bucket_name, object_name,
current_data, len(current_data),
metadata=metadata, sse=sse,
progress=progress)
def list_objects(self, bucket_name, prefix='', recursive=False):
"""
List objects in the given bucket.
Examples:
objects = minio.list_objects('foo')
for current_object in objects:
print(current_object)
# hello
# hello/
# hello/
# world/
objects = minio.list_objects('foo', prefix='hello/')
for current_object in objects:
print(current_object)
# hello/world/
objects = minio.list_objects('foo', recursive=True)
for current_object in objects:
print(current_object)
# hello/world/1
# world/world/2
# ...
objects = minio.list_objects('foo', prefix='hello/',
recursive=True)
for current_object in objects:
print(current_object)
# hello/world/1
# hello/world/2
:param bucket_name: Bucket to list objects from
:param prefix: String specifying objects returned must begin with
:param recursive: If yes, returns all objects for a specified prefix
:return: An iterator of objects in alphabetical order.
"""
is_valid_bucket_name(bucket_name)
# If someone explicitly set prefix to None convert it to empty string.
if prefix is None:
prefix = ''
method = 'GET'
# Initialize query parameters.
query = {
'max-keys': '1000',
'prefix': prefix
}
# Delimited by default.
if not recursive:
query['delimiter'] = '/'
marker = ''
is_truncated = True
while is_truncated:
if marker:
query['marker'] = marker
headers = {}
response = self._url_open(method,
bucket_name=bucket_name,
query=query,
headers=headers)
objects, is_truncated, marker = parse_list_objects(response.data,
bucket_name=bucket_name)
for obj in objects:
yield obj
def list_objects_v2(self, bucket_name, prefix='', recursive=False):
"""
List objects in the given bucket using the List objects V2 API.
Examples:
objects = minio.list_objects_v2('foo')
for current_object in objects:
print(current_object)
# hello
# hello/
# hello/
# world/
objects = minio.list_objects_v2('foo', prefix='hello/')
for current_object in objects:
print(current_object)
# hello/world/
objects = minio.list_objects_v2('foo', recursive=True)
for current_object in objects:
print(current_object)
# hello/world/1
# world/world/2
# ...
objects = minio.list_objects_v2('foo', prefix='hello/',
recursive=True)
for current_object in objects:
print(current_object)
# hello/world/1
# hello/world/2
:param bucket_name: Bucket to list objects from
:param prefix: String specifying objects returned must begin with
:param recursive: If yes, returns all objects for a specified prefix
:return: An iterator of objects in alphabetical order.
"""
is_valid_bucket_name(bucket_name)
# If someone explicitly set prefix to None convert it to empty string.
if prefix is None:
prefix = ''
# Initialize query parameters.
query = {
'list-type': '2',
'prefix': prefix
}
# Delimited by default.
if not recursive:
query['delimiter'] = '/'
continuation_token = None
is_truncated = True
while is_truncated:
if continuation_token is not None:
query['continuation-token'] = continuation_token
response = self._url_open(method='GET',
bucket_name=bucket_name,
query=query)
objects, is_truncated, continuation_token = parse_list_objects_v2(
response.data, bucket_name=bucket_name
)
for obj in objects:
yield obj
def stat_object(self, bucket_name, object_name, sse=None):
"""
Check if an object exists.
:param bucket_name: Bucket of object.
:param object_name: Name of object
:return: Object metadata if object exists
"""
headers = {}
if sse:
is_valid_sse_c_object(sse=sse)
headers.update(sse.marshal())
is_valid_bucket_name(bucket_name)
is_non_empty_string(object_name)
response = self._url_open('HEAD', bucket_name=bucket_name,
object_name=object_name, headers=headers)
etag = response.headers.get('etag', '').replace('"', '')
size = int(response.headers.get('content-length', '0'))
content_type = response.headers.get('content-type', '')
last_modified = response.headers.get('last-modified')
View remainder of file in raw view
0