【原创源码】【python】python3访问S3存储封装package

苍穹之麒麟

[Python] 纯文本查看 复制代码

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

# __ author:Jack
# date: 2020-04-26

# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html

import boto3
from boto3.s3.transfer import TransferConfig
from botocore.exceptions import ClientError

def get_s3_client(url, access_key, secret_key):
    return boto3.client('s3',
        endpoint_url = url,
        aws_access_key_id = access_key,
        aws_secret_access_key = secret_key)

def get_all_bucket(client):
    resp = client.list_buckets()
    return resp["Buckets"]

# 有bug
def get_all_s3_keys(client, bucket):
    keys = []
    kwargs = {'Bucket': bucket}
    while True:
        resp = client.list_objects_v2(**kwargs)
        for obj in resp['Contents']:
            keys.append(obj['Key'])
        try:
            kwargs['ContinuationToken'] = resp['NextContinuationToken']
        except KeyError as e:
            print(e)
            break
    return keys

def get_object_size(client, bucket, key):
    try:
        response = client.get_object(Bucket = bucket, Key = key)
        return response['ContentLength']
    except ClientError as e:
        errorCode = e.response['Error']['Code']
        print(errorCode)
    
    return None

if __name__=='__main__':
    client = get_s3_client('http://x.xx.x:81','code','code')
    #get_all_bucket(client)
    #print(get_all_s3_keys(client, 'bingodrive'))
    print(get_object_size(client,'xx','xx'))