Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 7 additions & 5 deletions filestack/uploads/intelligent_ingestion.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,14 @@
import os
import sys
import mimetypes
import multiprocessing
import hashlib
import logging
import functools
import threading
from urllib3.util.retry import Retry
from multiprocessing.pool import ThreadPool
# from multiprocessing.pool import ThreadPool
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please remove commented code

from concurrent.futures import ThreadPoolExecutor

from base64 import b64encode

Expand All @@ -27,7 +29,7 @@
CHUNK_SIZE = 8 * MB
MIN_CHUNK_SIZE = 32 * 1024
MAX_DELAY = 4
NUM_THREADS = 4
NUM_THREADS = multiprocessing.cpu_count() #4

lock = threading.Lock()

Expand Down Expand Up @@ -130,8 +132,8 @@ def upload(apikey, filepath, file_obj, storage, params=None, security=None):
upload_part, apikey, filename, filepath, filesize, storage, start_response
)

with ThreadPool(NUM_THREADS) as pool:
pool.map(fii_upload, parts)
with ThreadPoolExecutor(max_workers=NUM_THREADS) as executor:
list(executor.map(fii_upload, parts))

payload.update({
'uri': start_response['uri'],
Expand All @@ -147,7 +149,7 @@ def upload(apikey, filepath, file_obj, storage, params=None, security=None):

complete_url = 'https://{}/multipart/complete'.format(start_response['location_url'])
session = requests.Session()
retries = Retry(total=7, backoff_factor=0.2, status_forcelist=[202], method_whitelist=frozenset(['POST']))
retries = Retry(total=7, backoff_factor=0.2, status_forcelist=[202], allowed_methods=frozenset(['POST']))
session.mount('http://', requests.adapters.HTTPAdapter(max_retries=retries))
response = session.post(complete_url, json=payload, headers=config.HEADERS)
if response.status_code != 200:
Expand Down
11 changes: 6 additions & 5 deletions filestack/uploads/multipart.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@
import multiprocessing
from base64 import b64encode
from functools import partial
from multiprocessing.pool import ThreadPool
# from multiprocessing.pool import ThreadPool
from concurrent.futures import ThreadPoolExecutor

from filestack import config
from filestack.utils import requests
Expand Down Expand Up @@ -113,10 +114,10 @@ def multipart_upload(apikey, filepath, file_obj, storage, params=None, security=
chunks = make_chunks(filepath, file_obj, filesize)
start_response = multipart_request(config.MULTIPART_START_URL, payload, params, security)
upload_func = partial(upload_chunk, apikey, filename, storage, start_response)

with ThreadPool(upload_processes) as pool:
uploaded_parts = pool.map(upload_func, chunks)

with ThreadPoolExecutor(max_workers=upload_processes) as executor:
uploaded_parts = list(executor.map(upload_func, chunks))
location_url = start_response.pop('location_url')
payload.update(start_response)
payload['parts'] = uploaded_parts
Expand Down
4 changes: 2 additions & 2 deletions requirements.txt
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
pytest==4.6.3
pytest-cov==2.5.0
pytest>=4.6.3
pytest-cov>=2.5.0
requests==2.25.1
responses==0.14.0
trafaret==2.0.2