integrations/s3.py

+import boto3 as boto3
+
+from settings import REPORTS_FOLDER, ACCESS_KEY, SECRET_KEY, ENDPOINT_URL, REGION_NAME, BUCKET_NAME
+
+s3_client = boto3.client(
+    "s3",
+    aws_access_key_id=ACCESS_KEY,
+    aws_secret_access_key=SECRET_KEY,
+    endpoint_url=ENDPOINT_URL,
+    region_name=REGION_NAME,
+)
+
+
+def save_report_to_s3(file, report_id, is_public=False):
+    """
+        Saves the report to S3
+
+        Args:
+            file: Report file
+            report_id: Report ID
+            is_public: Flag, shows, might this report and all indicators from it become public
+    """
+    file_path = f"{REPORTS_FOLDER}/{report_id}.pdf"
+    s3_client.upload_file(file, BUCKET_NAME, file_path)
+
+    if is_public:
+        s3_client.put_object_acl(ACL="public-read", Bucket=BUCKET_NAME, Key=file_path)
+
+    return file_path
+
+
+def remove_report_from_s3(report_id):
+    """
+        Removes the report from S3
+
+        Args:
+            report_id: Report ID
+    """
+    file_path = f"{REPORTS_FOLDER}/{report_id}.pdf"
+    s3_client.delete_object(Bucket=BUCKET_NAME, Key=file_path)
+
+    return file_path
+
+
+def generate_report_url(report_id):
+    """
+        Generates the report URL
+
+        Args:
+            report_id: Report ID
+    """
+    file_path = f"{REPORTS_FOLDER}/{report_id}.pdf"
+    url = s3_client.generate_presigned_url("get_object", Params={
+        "Bucket": BUCKET_NAME,
+        "Key": file_path,
+    }, ExpiresIn=3600 * 24)
+
+    return url

settings.py

@@ -12,3 +12,11 @@ GITHUB_CLIENT_ID = getenv("GITHUB_CLIENT_ID")
 GITHUB_CLIENT_SECRET = getenv("GITHUB_CLIENT_SECRET")
 
 JWK_KEY_FILE = getenv("JWK_KEY_FILE") if getenv("JWK_KEY_FILE") is not None else "key.jwk"
+
+SECRET_KEY = getenv("S3_SECRET_KEY")
+ACCESS_KEY = getenv("S3_ACCESS_KEY")
+BUCKET_NAME = getenv("S3_BUCKET_NAME")
+REGION_NAME = getenv("S3_REGION_NAME")
+ENDPOINT_URL = getenv("S3_ENDPOINT_URL")
+REPORTS_FOLDER = getenv("S3_REPORTS_FOLDER") if getenv("S3_REPORTS_FOLDER") is not None else "reports"
+S3_BASE_URL = getenv("S3_BASE_URL")

tests/test_s3.py

+import os
+
+from requests import get
+
+from integrations.s3 import save_report_to_s3, remove_report_from_s3, generate_report_url
+from settings import S3_BASE_URL
+
+
+def test_s3_file_load_public():
+    pdf_url = "https://storage.yandexcloud.net/ivanprogramming/Network_Report.pdf"
+    pdf_path = "Network_Report.pdf"
+    with open(pdf_path, "wb") as f:
+        f.write(get(pdf_url).content)
+    save_report_to_s3(pdf_path, "test", is_public=True)
+
+    url = S3_BASE_URL + "test.pdf"
+    response = get(url)
+    assert response.status_code == 200
+
+    remove_report_from_s3("test")
+    os.remove(pdf_path)
+
+
+def test_s3_file_load_private():
+    pdf_url = "https://storage.yandexcloud.net/ivanprogramming/Network_Report.pdf"
+    pdf_path = "Network_Report.pdf"
+    with open(pdf_path, "wb") as f:
+        f.write(get(pdf_url).content)
+    save_report_to_s3(pdf_path, "test-private", is_public=False)
+
+    url = S3_BASE_URL + "test.pdf"
+    response = get(url)
+    assert response.status_code == 403
+    presigned_url = generate_report_url("test-private")
+    response = get(presigned_url)
+    assert response.status_code == 200
+
+    remove_report_from_s3("test-private")
+    os.remove(pdf_path)