diff --git a/tests/examples/upload_or_post.py b/tests/examples/upload_or_post.py new file mode 100644 index 0000000..71703c1 --- /dev/null +++ b/tests/examples/upload_or_post.py @@ -0,0 +1,60 @@ +from socketify import App + +### +# We always recomend check res.aborted in async operations +### + +def upload(res, req): + print(f"Posted to {req.get_url()}") + + def on_data(res, chunk, is_end): + print(f"Got chunk of data with length {len(chunk)}, is_end: {is_end}") + if (is_end): + res.end("Thanks for the data!") + + res.on_data(on_data) + +async def upload_chunks(res, req): + print(f"Posted to {req.get_url()}") + #await all the data, returns received chunks if fail (most likely fail is aborted requests) + data = await res.get_data() + + print(f"Got {len(data)} chunks if data!") + for chunk in data: + print(f"Got chunk of data with length {len(chunk)}") + + #We respond when we are done + res.end("Thanks for the data!") + +async def upload_json(res, req): + print(f"Posted to {req.get_url()}") + #await all the data and parses as json, returns None if fail + people = await res.get_json() + + if isinstance(people, list) and isinstance(people[0], dict): + print(f"First person is named: {people[0]['name']}") + + #We respond when we are done + res.end("Thanks for the data!") + +async def upload_text(res, req): + print(f"Posted to {req.get_url()}") + #await all the data and decode as text, returns None if fail + text = await res.get_text() #first parameter is the encoding (default utf-8) + + print(f"Your text is ${text}") + + #We respond when we are done + res.end("Thanks for the data!") + + + +app = App() +app.post("/", upload) +app.post("/chunks", upload_chunks) +app.post("/json", upload_json) +app.post("/text", upload_text) + +app.any("/*", lambda res,_: res.write_status(404).end("Not Found")) +app.listen(3000, lambda config: print("Listening on port http://localhost:%d now\n" % config.port)) +app.run() \ No newline at end of file diff --git a/tests/tests.py b/tests/tests.py new file mode 100644 index 0000000..078f52c --- /dev/null +++ b/tests/tests.py @@ -0,0 +1,17 @@ +# https://github.com/Tinche/aiofiles +# https://github.com/uNetworking/uWebSockets/issues/1426 + +# import os.path + +# def in_directory(file, directory): +# #make both absolute +# directory = os.path.join(os.path.realpath(directory), '') +# file = os.path.realpath(file) + +# #return true, if the common prefix of both is equal to directory +# #e.g. /a/b/c/d.rst and directory is /a/b, the common prefix is /a/b +# return os.path.commonprefix([file, directory]) == directory + +# application/x-www-form-urlencoded +# application/x-www-form-urlencoded +# multipart/form-data \ No newline at end of file