File size: 2,401 Bytes
410351d cbb9517 2bb6627 410351d 3457223 2bb6627 cbb9517 811e0e2 410351d cbb9517 410351d cbb9517 410351d cbb9517 410351d cbb9517 410351d 85fe8e5 410351d 85fe8e5 410351d 85fe8e5 cbb9517 3457223 d259141 |
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 |
from fastapi import FastAPI, File, UploadFile, HTTPException, Form
import uuid
import mimetypes
from datetime import datetime, timedelta
app = FastAPI()
# **In-Memory Storage (FOR DEMO ONLY)**
# Replace with Redis, NoSQL DB, or Cloud Storage for Production
data_storage = {}
@app.post("/upload/file")
async def upload_file(file: UploadFile = File(...)):
upload_id = str(uuid.uuid4())
data_storage[upload_id] = {
'file_name': file.filename,
'file_data': await file.read(),
'mime_type': file.content_type,
'upload_time': datetime.datetime.utcnow()
}
return {"upload_id": upload_id, "message": "File uploaded successfully"}
@app.post("/upload/custom")
async def upload_custom(file: UploadFile = File(...), download_link: str = Form(...)):
if not download_link:
raise HTTPException(status_code=400, detail="Download link is required")
# Validate the download_link to be URL-safe
if not download_link.isalnum() and not all(c in string.ascii_letters + string.digits + '-_' for c in download_link):
raise HTTPException(status_code=400, detail="Invalid download link format")
# Assign the custom download link
data_storage[download_link] = {
'file_name': file.filename,
'file_data': await file.read(),
'mime_type': file.content_type,
'upload_time': datetime.datetime.utcnow()
}
return {"download_link": download_link, "message": "File uploaded successfully"}
@app.get("/download/{upload_id}")
async def download_file(upload_id: str):
# Remove expired files
expire_time = datetime.datetime.utcnow() - timedelta(minutes=5)
keys_to_remove = [key for key, value in data_storage.items() if value['upload_time'] < expire_time]
for key in keys_to_remove:
del data_storage[key]
if upload_id not in data_storage:
raise HTTPException(status_code=404, detail="Upload not found")
upload_data = data_storage[upload_id]
file_name = upload_data['file_name']
mime_type = upload_data['mime_type'] or mimetypes.guess_type(file_name)[0] or "application/octet-stream"
return Response(content=upload_data['file_data'], media_type=mime_type,
headers={"Content-Disposition": f"attachment; filename={file_name}"})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8083) |