2023-03-21 23:58:22 +03:00
|
|
|
from typing import Union, Annotated
|
2023-03-22 18:22:43 +03:00
|
|
|
import uvicorn
|
2023-03-21 23:58:22 +03:00
|
|
|
|
|
|
|
from fastapi import FastAPI, File, UploadFile, Form, Response, status
|
|
|
|
|
|
|
|
import os
|
|
|
|
import magic
|
|
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/")
|
|
|
|
def read_root():
|
|
|
|
return {"Hello": "World"}
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/items/{version}/{name}/{file}")
|
|
|
|
def read_item(version: str, name: str, file: str = 'index.js'):
|
|
|
|
mime = magic.Magic(mime=True)
|
|
|
|
file_path = f"packages/{name}/{version}/{file}"
|
|
|
|
file_mime = mime.from_file(file_path)
|
|
|
|
if os.path.exists(file_path):
|
|
|
|
with open(file_path, 'rb') as f:
|
|
|
|
return Response(content=f.read(), media_type=file_mime)
|
|
|
|
|
|
|
|
return status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/files/")
|
|
|
|
async def create_file(file: Annotated[bytes | None, File()] = None):
|
|
|
|
if not file:
|
|
|
|
return {"message": "No file sent"}
|
|
|
|
else:
|
|
|
|
return {"file_size": len(file)}
|
|
|
|
|
|
|
|
|
|
|
|
@app.post("/uploadfile/")
|
|
|
|
async def create_upload_file(file: UploadFile | None, package: Annotated[str, Form()], version: Annotated[str, Form()], filename: Annotated[str, Form()]):
|
|
|
|
try:
|
|
|
|
contents = file.file.read()
|
|
|
|
dir_file = "packages/{package}/{version}".format(package=package, version=version)
|
|
|
|
if not os.path.exists(dir_file):
|
|
|
|
os.makedirs(dir_file)
|
|
|
|
with open(f"{dir_file}/{filename}", 'wb') as f:
|
|
|
|
f.write(contents)
|
2023-03-23 00:36:10 +03:00
|
|
|
except Exception as err:
|
|
|
|
return {"message": "There was an error uploading the file, error {err}".format(err=err)}
|
2023-03-21 23:58:22 +03:00
|
|
|
finally:
|
|
|
|
file.file.close()
|
|
|
|
|
2023-03-22 18:22:43 +03:00
|
|
|
return {"message": f"Successfully uploaded {file.filename}"}
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
2023-03-23 00:36:10 +03:00
|
|
|
uvicorn.run("server:app", port=5044, log_level="info")
|