File size: 2,374 Bytes
410351d cbb9517 2bb6627 410351d 3457223 2bb6627 cbb9517 811e0e2 410351d cbb9517 410351d 5832dd0 cbb9517 410351d cbb9517 410351d 5832dd0 410351d 5832dd0 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.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.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.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) |