aws s3 put object api 调用示例:
#!/usr/bin/python
#coding=utf-8
import sys, os, base64, datetime, hashlib, hmac, urllib, json, mimetypes
import requests # pip install requests
'''
请在linux环境下运行,因为对于一些特殊的文件名,在windows下会报错。
例如:
téléchargement.png
Tandläkargruppen.7z
ファイブクリエイツリミテッド様.zip
'''
def fileSize(filepath):
st = os.stat(filepath)
return st.st_size
def sortDicByKey(dic):
return [(k, dic[k]) for k in sorted(dic.keys())]
def sign(key, msg):
return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
def getSignatureKey(key, date_stamp, regionName, serviceName):
kDate = sign(('AWS4' + key).encode('utf-8'), date_stamp)
kRegion = sign(kDate, regionName)
kService = sign(kRegion, serviceName)
kSigning = sign(kService, 'aws4_request')
return kSigning
def uploadFile(filepath, s3key):
payload = readFile(filepath)
payload_md5 = base64.b64encode(hashlib.md5(payload).digest())
payload_hash = 'UNSIGNED-PAYLOAD' #hashlib.sha256(payload).hexdigest()
t = datetime.datetime.utcnow()
amz_date = t.strftime('%Y%m%dT%H%M%SZ')
date_stamp = amz_date[0:8]
canonical_uri = canonicalUri(s3key)
canonical_querystring = canonicalQueryStr('')
headers = {
'Host':host, #Global host
'Content-MD5':payload_md5,
'X-Amz-Date':amz_date,
'X-Amz-Content-Sha256':payload_hash,
'X-Amz-Storage-Class':'STANDARD', #REDUCED_REDUNDANCY
'X-Amz-Meta-Owner': 'Token Yi',
'X-Amz-Meta-Create': '2019-02-15T15:34'
}
mime = mimetypes.MimeTypes().guess_type(filepath)
if mime[0]:
headers['Content-Type'] = mime[0]
authorization_header = buildAuthorizationHeader(canonical_uri, canonical_querystring, payload_hash, headers)
headers['Authorization'] = authorization_header
return httpPut(endpoint + canonical_uri.lstrip('/'), data=payload, headers=headers)
def readFile(filepath):
filepath = unicode(filepath, 'utf-8')#适应中文路径
fh = open(filepath, 'rb')
content = fh.read()
fh.close()
return content
def canonicalUri(s3objectKey):
return '/' + urllib.quote(s3objectKey.lstrip('/')).replace('%2F', '/')
def canonicalQueryStr(querystring):
return urllib.quote(querystring)
def httpPut(url, data, headers):
return requests.put(url, data=data, headers=headers)
def buildAuthorizationHeader(canonical_uri, canonical_querystring, payload_hash, headers):
canonical_headers = canonicalHeaders(headers)
signed_headers = signedHeaders(headers)
canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash #global method
algorithm = 'AWS4-HMAC-SHA256'
date_stamp = headers['X-Amz-Date'][0:8]
credential_scope = date_stamp + '/' + region + '/' + service + '/' + 'aws4_request' #global region, service
string_to_sign = algorithm + '\n' + headers['X-Amz-Date'] + '\n' + credential_scope + '\n' + hashlib.sha256(canonical_request).hexdigest()
signing_key = getSignatureKey(secret_key, date_stamp, region, service)
signature = hmac.new(signing_key, (string_to_sign).encode('utf-8'), hashlib.sha256).hexdigest()
authorization_header = algorithm + ' ' + 'Credential=' + access_key + '/' + credential_scope + ', ' + 'SignedHeaders=' + signed_headers + ', ' + 'Signature=' + signature #global access_key
return authorization_header
def canonicalHeaders(headers):
sortedHeaders = sortDicByKey(headers)
output = ''
for tupOfkv in sortedHeaders:
output += tupOfkv[0].strip().lower() + ':' + tupOfkv[1].strip() + '\n'
return output
def signedHeaders(headers):
return ';'.join(sorted([k.lower() for k in headers.keys()]))
access_id = 'xxxxxxxxxx'
secret_key = 'xxxxxxxxx'
region = 'eu-west-1'
bucket = 'my-bucket'
access_key = access_id
secret_key = secret_key
if access_key is None or secret_key is None:
print 'No access key is available.'
sys.exit()
method = 'PUT'
service = 's3'
host = bucket + '.s3.amazonaws.com'
endpoint = 'https://' + bucket + '.s3.amazonaws.com/'
r = uploadFile('./readme.txt', s3key = '/testupload/py_upload中文.txt')
print '\nRESPONSE++++++++++++++++++++++++++++++++++++'
print 'Response code: %d\n' % r.status_code
print r.text