beenull/simple-login
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

refactor s3.py

Browse source
This commit is contained in:
Son NK 2019-12-23 14:56:55 +00:00
parent db60043341
commit 3320a488f3
1 changed files with 5 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
app/s3.py
View file

@ -3,15 +3,9 @@ from io import BytesIO
import boto3 import boto3
import requests import requests
from app.config import ( from app.config import AWS_REGION, BUCKET, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
AWS_REGION,
BUCKET,
AWS_ACCESS_KEY_ID,
AWS_SECRET_ACCESS_KEY,
AVATAR_URL_EXPIRATION,
)
session = boto3.Session( _session = boto3.Session(
aws_access_key_id=AWS_ACCESS_KEY_ID, aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY, aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
region_name=AWS_REGION, region_name=AWS_REGION,
@ -20,7 +14,7 @@ session = boto3.Session(
def upload_from_bytesio(key: str, bs: BytesIO, content_type="string") -> None: def upload_from_bytesio(key: str, bs: BytesIO, content_type="string") -> None:
bs.seek(0) bs.seek(0)
session.resource("s3").Bucket(BUCKET).put_object( _session.resource("s3").Bucket(BUCKET).put_object(
Key=key, Body=bs, ContentType=content_type Key=key, Body=bs, ContentType=content_type
) )
@ -31,21 +25,14 @@ def upload_from_url(url: str, upload_path):
def delete_file(key: str) -> None: def delete_file(key: str) -> None:
o = session.resource("s3").Bucket(BUCKET).Object(key) o = _session.resource("s3").Bucket(BUCKET).Object(key)
o.delete() o.delete()
def get_url(key: str, expires_in=3600) -> str: def get_url(key: str, expires_in=3600) -> str:
s3_client = session.client("s3") s3_client = _session.client("s3")
return s3_client.generate_presigned_url( return s3_client.generate_presigned_url(
ExpiresIn=expires_in, ExpiresIn=expires_in,
ClientMethod="get_object", ClientMethod="get_object",
Params={"Bucket": BUCKET, "Key": key}, Params={"Bucket": BUCKET, "Key": key},
) )
if __name__ == "__main__":
with open("/tmp/1.png", "rb") as f:
upload_from_bytesio("1.png", BytesIO(f.read()))
print(get_url(BUCKET, "1.png"))