How to perform file upload in Sanic
After a long struggle I found the following code to be working
@app.route("/upload", methods=['POST'])
async def omo(request):
from sanic import response
import os
import aiofiles
if not os.path.exists(appConfig["upload"]):
os.makedirs(appConfig["upload"])
async with aiofiles.open(appConfig["upload"]+"/"+request.files["file"][0].name, 'wb') as f:
await f.write(request.files["file"][0].body)
f.close()
return response.json(True)
Here's an example of file upload for a specific file type (this one is for pdf files)
from sanic import Sanic
from sanic.response import json
from pathlib import os
from datetime import datetime
app = Sanic()
config = {}
config["upload"] = "./tests/uploads"
@app.route("/upload", methods=['POST'])
def post_json(request):
if not os.path.exists(config["upload"]):
os.makedirs(config["upload"])
test_file = request.files.get('file')
file_parameters = {
'body': test_file.body,
'name': test_file.name,
'type': test_file.type,
}
if file_parameters['name'].split('.')[-1] == 'pdf':
file_path = f"{config['upload']}/{str(datetime.now())}.pdf"
with open(file_path, 'wb') as f:
f.write(file_parameters['body'])
f.close()
print('file wrote to disk')
return json({ "received": True, "file_names": request.files.keys(), "success": True })
else:
return json({ "received": False, "file_names": request.files.keys(), "success": False, "status": "invalid file uploaded" })
For examples on other request types, refer official docs (https://sanic.readthedocs.io/en/latest/sanic/request_data.html)
The answers above are great. A few minor improvements:
(1) Since we are using Sanic, let's try to do the file io asynchronously:
async def write_file(path, body):
async with aiofiles.open(path, 'wb') as f:
await f.write(body)
f.close()
(2) Make sure that the file isn't too large so as to crash your server:
def valid_file_size(file_body):
if len(file_body) < 10485760:
return True
return False
(3) Check both the file name and file type for the proper type of file:
def valid_file_type(file_name, file_type):
file_name_type = file_name.split('.')[-1]
if file_name_type == "pdf" and file_type == "application/pdf":
return True
return False
(4) Ensure the filename doesn't have dangerous/insecure characters. You can use the secure_filename function in werkzeug.utils: http://flask.pocoo.org/docs/0.12/patterns/fileuploads/
(5) This code brings it all together:
async def process_upload(request):
# Create upload folder if doesn't exist
if not os.path.exists(app.config.UPLOAD_DIR):
os.makedirs(app.config.UPLOAD_DIR)
# Ensure a file was sent
upload_file = request.files.get('file_names')
if not upload_file:
return redirect("/?error=no_file")
# Clean up the filename in case it creates security risks
filename = secure_filename(upload_file.name)
# Ensure the file is a valid type and size, and if so
# write the file to disk and redirect back to main
if not valid_file_type(upload_file.name, upload_file.type):
return redirect('/?error=invalid_file_type')
elif not valid_file_size(upload_file.body):
return redirect('/?error=invalid_file_size')
else:
file_path = f"{app.config.UPLOAD_DIR}/{str(datetime.now())}.pdf"
await write_file(file_path, upload_file.body)
return redirect('/?error=none')
I created a blog post on how I handle file upload in Sanic. I added some file validation and also asynchronous file writing. I hope others find this helpful:
https://blog.fcast.co/2019/06/16/file-upload-handling-using-asynchronous-file-writing/