python-3.x 如何在Sanic中执行文件上传

l0oc07j2  于 2023-02-17  发布在  Python
关注(0)|答案(4)|浏览(336)

我试图在Sanic上执行文件上传,但它不能正常工作, flask 的正常语法似乎不能很好地与Sanic在这里工作。
我甚至无法访问文件名或保存方法来将上传的文件保存到给定的目录。

hsvhsicv

hsvhsicv1#

经过长时间的努力,我发现下面的代码是工作

@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)
6mzjoqzu

6mzjoqzu2#

上面的答案很好。还有一些小的改进:
(1)因为我们使用的是Sanic,所以让我们尝试异步处理文件io:

async def write_file(path, body):
    async with aiofiles.open(path, 'wb') as f:
        await f.write(body)
    f.close()

(2)请确保该文件不会太大而导致服务器崩溃:

def valid_file_size(file_body):
    if len(file_body) < 10485760:
        return True
    return False

(3)检查文件名和文件类型,以确定文件类型是否正确:

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)确保文件名中没有危险/不安全的字符。您可以使用werkzeug.utils中的secure_filename函数:http://flask.pocoo.org/docs/0.12/patterns/fileuploads/
(5)此代码将所有功能组合在一起:

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')

我写了一篇博文,介绍了我如何在Sanic中处理文件上传。我添加了一些文件验证和异步文件写入。我希望其他人会觉得这很有帮助:
https://blog.fcast.co/2019/06/16/file-upload-handling-using-asynchronous-file-writing/

lzfw57am

lzfw57am3#

下面是一个特定文件类型的文件上传示例(这个是pdf文件)

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" })

有关其他请求类型的示例,请参阅官方文档(https://sanic.readthedocs.io/en/latest/sanic/request_data.html

ajsxfq5m

ajsxfq5m4#

上面的答案忽略了html文件的必要存在。也有一些bug在那里。现在我发布我的答案,在我的机器上工作。
代码结合了sanic和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)

以上不仅接收上传的文件,而且在另一个文件中返回响应。希望这对很多人有帮助。

相关问题