-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
63 lines (49 loc) · 1.81 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
# encoding: utf-8
import magic
from base64 import b64decode
from io import BytesIO
from os import path
from pydantic.main import BaseModel
from fastapi import FastAPI, HTTPException, status, Query
from modules.settings import get_setting
from typing import Optional
from modules.s3_client import S3Client
class ImageException(HTTPException):
def __init__(self, exception):
super(ImageException, self).__init__(detail=exception, status_code=400)
class Image(BaseModel):
base64_content: str
filename: str
app = FastAPI()
c = S3Client()
@app.get("/status/")
def service_status():
r = {'status': 'ok', 'bucket_name': get_setting().bucket_name, 'errors': []}
try:
files = c.get_list('', skip_cache=True)
r['files_in_env'] = len(files)
except Exception as e:
r['status'] = 'error'
r['errors'].append(f'S3Client:{repr(e)}')
return r
@app.get("/items/")
def get_files(prefix: Optional[str] = Query('', title="S3 bucket prefix")):
return c.get_list(prefix)
@app.post("/items/", status_code=status.HTTP_201_CREATED)
def upload_file(image: Image, prefix: Optional[str] = Query('', title="S3 bucket prefix")):
try:
# build the image
image_content = bytes(image.base64_content, encoding="ascii")
image_content = b64decode(image_content)
content_type = magic.from_buffer(image_content, mime=True)
image_content = BytesIO(image_content)
except Exception as e:
raise ImageException(f'exception base64_content: {repr(e)}')
url = path.join(prefix, image.filename)
return {
'image_url': c.upload(url, image_content, content_type)
}
@app.delete("/items/")
def delete_file(filename: str, prefix: Optional[str] = Query('', title="S3 bucket prefix")):
url = path.join(prefix, filename)
return c.delete(url)