apistar_multipart.py

import os
from typing import Any

from apistar import App, Route, http
from apistar.codecs.multipart import MultiPartCodec
from werkzeug.datastructures import FileStorage

UPLOAD_FOLDER = os.path.join('.', 'uploads')
PORT = 8080


def build_response(success: bool = False, error: str = None, data: Any = None):
    result = {
        'success': success,
        'error': error,
        'data': data
    }
    return result


def upload(request: http.Request, body: http.Body) -> dict:
    try:
        md = MultiPartCodec().decode(body, dict(request.headers))
        file_storage: FileStorage = md.get('file')
        filename = os.path.join(UPLOAD_FOLDER, str(file_storage.filename))
        file_storage.save(filename)

        return build_response(success=True, data="Upload complete")

    except Exception as all_exceptions:
        return build_response(error=str(all_exceptions))


routes = [
    Route('/upload', method='POST', handler=upload),
]

app = App(routes=routes)

if __name__ == '__main__':
    app.serve('0.0.0.0', 5000, debug=False)