s3.py 1003 B

1234567891011121314151617181920212223242526272829303132333435363738
  1. from io import BytesIO
  2. import boto3
  3. from app.config import AWS_REGION, BUCKET, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY
  4. session = boto3.Session(
  5. aws_access_key_id=AWS_ACCESS_KEY_ID,
  6. aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
  7. region_name=AWS_REGION,
  8. )
  9. def upload_from_bytesio(key: str, bs: BytesIO, content_type="string") -> None:
  10. bs.seek(0)
  11. session.resource("s3").Bucket(BUCKET).put_object(
  12. Key=key, Body=bs, ContentType=content_type
  13. )
  14. def delete_file(key: str) -> None:
  15. o = session.resource("s3").Bucket(BUCKET).Object(key)
  16. o.delete()
  17. def get_url(key: str) -> str:
  18. """by default the link will expire in 1h (3600 seconds)"""
  19. s3_client = session.client("s3")
  20. return s3_client.generate_presigned_url(
  21. ClientMethod="get_object", Params={"Bucket": BUCKET, "Key": key}
  22. )
  23. if __name__ == "__main__":
  24. with open("/tmp/1.png", "rb") as f:
  25. upload_from_bytesio("1.png", BytesIO(f.read()))
  26. print(get_url(BUCKET, "1.png"))