How to perform file upload in Sanic

Question

I am trying to perform file upload on Sanic but it is not working properly, the normal syntax for flask doesn't seem to work well with sanic here.

I can't access even the filename or the save method to save the uploaded file to a given directory.

Answer 1

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)

Answer 2

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/

Answer 3

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)

Answer 4

The above answers ignore the necessary existence of html files. Also there are some bugs out there. Now I post my answer that works on my machine.

The code combines sanic as well as html.

import asyncio

from sanic import Sanic
from sanic.response import html, file, text
from sanic.request import Request

import os
app = Sanic('something')

@app.route("/")
async def index(request: Request):
    return html("""
        <form action="/upload" method="post" enctype="multipart/form-data">
            <input type="file" name="file">
            <input type="submit" value="Upload">
        </form>
    """)

@app.route("/upload", methods=["POST"])
async def upload(request: Request):
    file_ = request.files.get("file")
    return await file((os.getcwd() + '/tempdir/some_file_feedback.csv'),  filename="your_match.pdf")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=6000)

The above not only receives uploaded file, but also gives back response in another file. Hope this helps for many people.