secret
Last active

upload files using cloudfiles with multiprocessing/concurrent.futures modules

  • Download Gist
connectionpool_cloudfiles.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
from concurrent import futures # pip install futures
 
from cloudfiles.connection import ConnectionPool
 
def get_container(name, conn):
#NOTE: ConnectionPool ensures that `conn` is owned by one thread
# at a time; get/set themself on a dict are atomic in CPython for
# tuples of strings, ints
try: return get_container.cache[name, id(conn)]
except KeyError:
ret = get_container.cache[name, id(conn)] = conn.create_container(name)
return ret
get_container.cache = {}
 
def uploader(connection_pool, filename):
conn = connection_pool.get()
try:
obj = get_container('test', conn).create_object(filename)
obj.load_from_filename(filename)
finally:
connection_pool.put(conn)
 
nconnections = 16
username = '---'
api_key = '---'
connection_pool = ConnectionPool(username, api_key, poolsize=nconnections)
 
with futures.ThreadPoolExecutor(max_workers=nconnections) as executor:
jobs = [executor.submit(uploader, connection_pool, f)
for f in get_filenames()]
 
for future in futures.as_completed(jobs):
if future.exception() is not None:
print(future.exception()) # error
multiprocessing_pool_cloudfiles.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
import multiprocessing
import cloudfiles
 
USERNAME = '---'
API_KEY = '---'
 
def get_container():
if get_container.value is None:
conn = cloudfiles.get_connection(USERNAME, API_KEY)
get_container.value = conn.create_container('test')
return get_container.value
get_container.value = None
 
def upload_file(filename):
container = get_container()
obj = container.create_object(filename)
obj.load_from_filename(filename)
 
def main():
NPROCESSES = 4
pool = multiprocessing.Pool(processes=NPROCESSES)
for _ in pool.imap_unordered(upload_file, get_filenames()):
pass
 
if __name__=="__main__":
multiprocessing.freeze_support()
main()

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.