In our last post, we discussed how to bulk rename files in AWS S3 using Python. Today, let’s talk about how to access object storage using the S3 API in Python. To access object storage using the S3 API in Python, the demo code is as follows.
How to use boto3 to connect to S3?
#!/usr/bin/python3
import os
import time
import boto
import boto.s3.connection
import argparse
import sys
# You need to modify the secret_access_key and access_key_id for the corresponding user.
access_key = "ZQAC7EA0G95ZBUAQXDQP"
secret_key = "pfgP1XiMXxJMoJ8pQv46VmndkBMTLwz2TDt0Ogl5"
HOST = "127.0.0.1"
PORT = 7480
useage = """usage: s3client.py [-h] [-c host] [-p port] [--bucket bucket_name]
[--create_bucket bucket_name] [--delete_bucket bucket_name]
[--list_buckets] [--create_object object]
[--delete_object object] [--list_objects]
optional arguments:
-h, --help
show this help message and exit
-c host, --client host
The client to connect host. e.g. -c 172.30.28.181
-p port, --port port
The port of server to be connected. e.g. -p 7480
--bucket bucket_name
Specify a bucket. e.g. --bucket my-bucket
--create_bucket bucket_name
Creates a new S3 bucket. e.g. --create_bucket my-bucket
--delete_bucket bucket_name
Deletes a new S3 bucket. e.g. --delete_bucket my-bucket
--list_buckets
List all buckets. e.g. --list_buckets
--create_object
Adds an object to a bucket. e.g. --bucket my-bucket --create_object --key file.txt --body /root/file.txt
--key
Specified object key. e.g. --bucket my-bucket --create_object --key file.txt --body /root/file.txt
--body
Specified object body. e.g. --bucket my-bucket --create_object --key file.txt --body /root/file.txt
--get_object
Gets an object from a bucket. e.g. --bucket my-bucket --get_object --key file.txt --body /root/file.txt
--copy_object
Copy a new S3 bucket. e.g. --copy_object my-bucket/testfile --key testfile --bucket your-bucket
--delete_object
Removes an object from a bucket. e.g. --bucket my-bucket --delete_object --key file.txt
--list_objects
List all objects in a bucket. e.g. --bucket my-bucket --list_objects"""
parse = argparse.ArgumentParser(add_help=False)
parse.add_argument(
"-h",
"--help",
action='store_true',
)
parse.add_argument(
"-c",
"--client",
metavar="host",
)
parse.add_argument(
"-p",
"--port",
metavar="port",
)
parse.add_argument(
"--bucket",
metavar="bucket_name",
)
parse.add_argument(
"--create_bucket",
metavar="bucket_name",
)
parse.add_argument(
"--delete_bucket",
metavar="bucket_name",
)
parse.add_argument(
"--list_buckets",
action='store_true',
)
parse.add_argument(
"--create_object",
action='store_true',
)
parse.add_argument(
"--key",
metavar="key",
)
parse.add_argument(
"--body",
metavar="file",
)
parse.add_argument(
"--copy_object",
metavar="object",
)
parse.add_argument(
"--get_object",
action='store_true',
)
parse.add_argument(
"--delete_object",
action='store_true',
)
parse.add_argument(
"--list_objects",
action='store_true',
)
args = parse.parse_args()
class S3client:
def __init__(self, host, port) -> None:
self.conn = boto.connect_s3(
aws_access_key_id = access_key,
aws_secret_access_key = secret_key,
host = host,
port = port,
debug=0,
is_secure=False, # uncomment if you are not using ssl
calling_format = boto.s3.connection.OrdinaryCallingFormat(),
)
self.bucket = None
def __del__(self):
self.conn.close()
def get_bucket(self, bucket_name):
for i in self.conn.get_all_buckets():
if i.name == bucket_name:
self.bucket = i
break
@property
def create_bucket(self):
self.conn.create_bucket(args.create_bucket)
print("Bucket %s created" % args.create_bucket)
@property
def delete_bucket(self):
self.conn.delete_bucket(args.delete_bucket)
print("Bucket %s removed" % args.delete_bucket)
@property
def list_buckets(self):
for i in self.conn.get_all_buckets():
print(i.name)
@property
def create_object(self):
self.get_bucket(args.bucket)
start_time = time.time()
key = self.bucket.new_key(args.key)
key.set_contents_from_filename(os.path.basename(args.body))
print("upload: '%s' -> 's3://%s/%s' [1 of 1]" % (args.body, args.bucket, args.key))
print("Time: %.2fs " % float(time.time()-start_time))
@property
def get_object(self):
self.get_bucket(args.bucket)
key = self.bucket.get_key(args.key)
key.get_contents_to_filename(args.body)
print("download: 's3://%s/%s' to %s " % (args.bucket, args.key, args.body))
@property
def copy_object(self):
self.conn.get_bucket(args.bucket).copy_key(new_key_name=args.key,
src_bucket_name=args.copy_object.split("/")[0],
src_key_name='/'.join(args.copy_object.split("/")[1:]))
print("copy: 's3://%s' to 's3://%s/%s' " % (args.copy_object, args.bucket, args.key))
@property
def list_objects(self):
self.get_bucket(args.bucket)
for key in self.bucket.list():
print("%s\t%-10s\t%s" % (key.last_modified, key.size, key.name))
@property
def delete_object(self):
self.get_bucket(args.bucket)
self.bucket.delete_key(args.key)
print("delete: 's3://%s/%s'" % (args.bucket, args.key))
if __name__ == "__main__":
if args.help:
print(useage)
sys.exit(0)
host = HOST if not args.client else args.client
port = PORT if not args.port else args.port
s3 = S3client(host, port)
try:
for i in args._get_kwargs():
if getattr(args, i[0]):
hasattr(s3, i[0])
except Exception as e:
print("no valid command found; please check!")
print(useage)
How to use boto3 to connect to S3?
#!/usr/bin/python3
import json
import os
import time
import datetime
import argparse
import sys
import boto3
# You need to modify the secret_access_key and access_key_id for the corresponding user.
access_key = "0INSHOH6YXYN9ZW9P2VT"
secret_key = "ZZzQ2CvfmtI2Wn5UE11WvlQtBX92pIIpSfyhRalv"
# Modify the following parameters based on your actual environment:
http_protocol = "https"
ca = "/tmp/civetweb.pem"
HOST = "vip" # IP or domain name, 127.0.0.1/vip
PORT = 8090
useage = """usage: s3client.py [-h] [-c host] [-p port] [--bucket bucket_name]
[--create_bucket bucket_name] [--delete_bucket bucket_name]
[--list_buckets] [--put_object object] [--copy_object object]
[--delete_object object] [--list_objects] [--ca cert.pem]
[--http_protocol https]
optional arguments:
-h, --help
show this help message and exit
-c host, --client host
The client to connect host. e.g. -c 172.30.28.181
-p port, --port port
The port of server to be connected. e.g. -p 7480
--http_protocol https
Specify http protocol, usually used with --ca. e.g. --http_protocol https
--ca cert.pem
A filename of the CA cert bundle to uses. e.g. --ca /etc/ceph/ca/civetweb.pem
--bucket bucket_name
Specify a bucket. e.g. --bucket my-bucket
--create_bucket bucket_name
Creates a new S3 bucket. e.g. --create_bucket my-bucket
--delete_bucket bucket_name
Deletes a new S3 bucket. e.g. --delete_bucket my-bucket
--list_buckets
List all buckets. e.g. --list_buckets
--put_object
Adds an object to a bucket. e.g. --bucket my-bucket --put_object --key file.txt --body /root/file.txt
--key
Specified object key. e.g. --bucket my-bucket --put_object --key file.txt --body /root/file.txt
--body
Specified object body. e.g. --bucket my-bucket --put_object --key file.txt --body /root/file.txt
--get_object
Gets an object from a bucket. e.g. --bucket my-bucket --get_object --key file.txt --body /root/file.txt
--copy_object
Copy a new S3 bucket. e.g. --copy_object my-bucket/testfile --key testfile --bucket your-bucket
--delete_object
Removes an object from a bucket. e.g. --bucket my-bucket --delete_object --key file.txt
--list_objects
List all objects in a bucket. e.g. --bucket my-bucket --list_objects
Example:
1. Use HTTP protocol to get the bucket.
python3 s3client.py --list_buckets --http_protocol http
2. Use HTTPS protocol to get the bucket.
python3 s3client.py --list_buckets --http_protocol https --ca /etc/ceph/ca/civetweb.pem
3. Use HTTP protocol to get objects.
python3 s3demo.py --list_objects --bucket python -c 172.30.25.99 -p 7480 --http_protocol http
4. Use HTTPS protocol to get objects.
python3 s3demo.py --list_objects --bucket python -c vip -p 8090 --http_protocol https
"""
parse = argparse.ArgumentParser(add_help=False)
parse.add_argument(
"-h",
"--help",
action='store_true',
)
parse.add_argument(
"-c",
"--client",
metavar="host",
)
parse.add_argument(
"-p",
"--port",
metavar="port",
)
parse.add_argument(
"--http_protocol",
metavar="https",
)
parse.add_argument(
"--ca",
metavar="cert.pem",
)
parse.add_argument(
"--bucket",
metavar="bucket_name",
)
parse.add_argument(
"--create_bucket",
metavar="bucket_name",
)
parse.add_argument(
"--delete_bucket",
metavar="bucket_name",
)
parse.add_argument(
"--list_buckets",
action='store_true',
)
parse.add_argument(
"--put_object",
action='store_true',
)
parse.add_argument(
"--key",
metavar="key",
)
parse.add_argument(
"--body",
metavar="file",
)
parse.add_argument(
"--copy_object",
metavar="object",
)
parse.add_argument(
"--get_object",
action='store_true',
)
parse.add_argument(
"--delete_object",
action='store_true',
)
parse.add_argument(
"--list_objects",
action='store_true',
)
args = parse.parse_args()
class S3client:
def __init__(self, host, port) -> None:
self.client = boto3.client(
service_name='s3',
aws_access_key_id = access_key,
aws_secret_access_key = secret_key,
endpoint_url=f'{http}://{host}:{port}',
verify=ca
)
@property
def create_bucket(self):
response = self.client.create_bucket(Bucket=args.create_bucket)
self.print_json(response)
@property
def delete_bucket(self):
response = self.client.delete_bucket(Bucket=args.delete_bucket)
self.print_json(response)
@property
def list_buckets(self):
response = self.client.list_buckets()
self.print_json(response)
@property
def put_object(self):
try:
start_time = time.time()
body = os.path.abspath(args.body) if args.body else None
response = self.client.put_object(Body=body, Bucket=args.bucket, Key=args.key)
print("upload: '%s' -> 's3://%s/%s' [1 of 1]" % (args.body, args.bucket, args.key))
print("Time: %.2fs " % float(time.time()-start_time))
self.print_json(response)
except Exception as e:
print(e)
@property
def get_object(self):
try:
self.client.download_file(Bucket=args.bucket,
Key=args.key,
Filename=args.body)
print("download: 's3://%s/%s' to %s " % (args.bucket, args.key, args.body))
except Exception as e:
print(e)
@property
def copy_object(self):
response = self.client.copy_object(
Bucket=args.bucket,
CopySource=args.copy_object,
Key=args.key,
)
print("copy: 's3://%s' to 's3://%s/%s' " % (args.copy_object, args.bucket, args.key))
self.print_json(response)
@property
def list_objects(self):
response = self.client.list_objects(
Bucket=args.bucket,
)
self.print_json(response)
@property
def delete_object(self):
try:
response = self.client.delete_object(
Bucket=args.bucket,
Key=args.key,
)
print("delete: 's3://%s/%s'" % (args.bucket, args.key))
self.print_json(response)
except Exception as e:
print(e)
@staticmethod
def print_json(data):
print(json.dumps(data,cls=DateEncoder, sort_keys=True, indent=4, separators=(', ', ': '), ensure_ascii=False))
class DateEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime.datetime):
return obj.strftime("%Y-%m-%d %H:%M:%S")
else:
return json.JSONEncoder.default(self,obj)
if __name__ == "__main__":
if args.help:
print(useage)
sys.exit(0)
host = HOST if not args.client else args.client
port = PORT if not args.port else args.port
http = http_protocol if not args.http_protocol else args.http_protocol
if http_protocol == "https":
if args.ca:
ca = args.ca
else:
ca = ca
else:
ca = False
s3 = S3client(host, port)
try:
for i in args._get_kwargs():
if getattr(args, i[0]):
hasattr(s3, i[0])
except Exception as e:
print("no valid command found; please check!\nUsage: python3 s3demo.py -h")
