Last active
June 28, 2019 09:41
-
-
Save northtree/8b560a6b552a975640ec406c9f701731 to your computer and use it in GitHub Desktop.
Test memory leakage on uploading file to Google Cloud Storage
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
FROM python:3.7 | |
COPY . /app | |
WORKDIR /app | |
RUN pip install --upgrade pip | |
RUN pip install memory_profiler | |
RUN pip install google-cloud-storage | |
ENTRYPOINT ["python", "-m", "memory_profiler", "test_gcp_storage.py"] |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
[[source]] | |
name = "pypi" | |
url = "https://pypi.org/simple" | |
verify_ssl = true | |
[dev-packages] | |
[packages] | |
memory-profiler = "*" | |
google-cloud-storage = "*" | |
[requires] | |
python_version = "3.7" |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from datetime import datetime | |
from google.cloud import storage | |
from memory_profiler import profile | |
@profile | |
def test_upload_big_file(): | |
client = storage.Client() | |
m_bytes = 64 | |
filename = int(datetime.utcnow().timestamp()) | |
blob_name = f'test/{filename}' | |
bucket_name = 'insights_scraping' | |
bucket = client.get_bucket(bucket_name) | |
with open(f'/tmp/{filename}', 'wb+') as file_obj: | |
file_obj.seek(m_bytes * 1024 * 1024 - 1) | |
file_obj.write(b'\0') | |
file_obj.seek(0) | |
blob = bucket.blob(blob_name) | |
blob.upload_from_file(file_obj) | |
# blob.upload_from_filename(f'/tmp/{filename}') | |
# del blob | |
blob = bucket.get_blob(blob_name) | |
print(blob.size) | |
if __name__ == '__main__': | |
test_upload_big_file() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Install:
pip install google-cloud-storage
Run:
python -m memory_profiler test_gcp_storage.py