Tags: #python #pipenv #tornado #async #concurrency #httpclient #asynchttpclient #tox #ini #basicauth #auth
Set-up the environment:
See this gist for more information on Pipenv.
pipenv --python 3.7pipenv shellpipenv install tornado==5.0.2pipenv install --dev mypy tox flake8$ cat Pipfile
[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"
[dev-packages]
mypy = "*"
tox = "*"
"flake8" = "*"
[packages]
tornado = "==5.0.2"
[requires]
python_version = "3.7"
"""
simplest implementation
"""
from tornado.httpclient import AsyncHTTPClient
from tornado.ioloop import IOLoop
AsyncHTTPClient.configure(None, defaults=dict(user_agent="MyUserAgent"))
http_client = AsyncHTTPClient()
async def do_thing():
response = await http_client.fetch("https://www.integralist.co.uk")
print(response.code)
print(response.headers)
print(response.body)
io_loop = IOLoop.current()
io_loop.run_sync(do_thing)
"""
demonstrates the 'callback' variation of AsyncHTTPClient#fetch
"""
import json
from tornado.httpclient import AsyncHTTPClient
from tornado.ioloop import IOLoop
# if None, then use tornado.simple_httpclient.SimpleAsyncHTTPClient
# otherwise specify alternative client such as curl:
# configure.("tornado.curl_httpclient.CurlAsyncHTTPClient")
AsyncHTTPClient.configure(None, defaults=dict(user_agent="MyUserAgent"))
http_client = AsyncHTTPClient()
def handle_response(response):
if response.error:
print("Error: %s" % response.error)
else:
print('\n' + str(response.body) + '\n') # .body is a byte array
data = json.loads(response.body)
print(data)
async def get_content():
await http_client.fetch("http://httpbin.org/json", handle_response)
# Note:
#
# The above callback 'style' is different to...
#
# response = await http_client.fetch("http://httpbin.org/json")
# print(response.body)
#
# ...which forces the function to wait for the response, then print it,
# before returning to `main()` which attempts to print its own message.
#
# By using the callback style we end up letting `get_content` return immediately.
#
# This isn't to say that the non-callback style is synchronous, it is still asynchronous.
# In that other processes can be executed while it's happening, but it doesn't work quite the same.
# As you can see here in this example.
post_data = {"data": "test data"}
body = urllib.parse.urlencode(post_data)
response = await http_client.fetch("http://httpbin.org/post", method='POST', headers=None, body=body)
print('\n' + str(response.body) + '\n') # .body is a byte array
# As per the tornado documentation:
# http://www.tornadoweb.org/en/stable/httpclient.html#tornado.httpclient.HTTPRequest
#
# The AsyncHTTPClient can accept any keyword args that are supported by
# the `HTTPRequest` object, and so if you need to provide basic auth credentials...
#
# http_client.fetch(uri, auth_username=<...>, auth_password=<...>)
async def main():
await get_content()
print("I won't wait for get_content to finish. I'll show immediately.")
if __name__ == "__main__":
io_loop = IOLoop.current()
io_loop.run_sync(main) # only runs the specified function, then stops
# alternatively use IOLoop.current().start()
# as that won't stop until explicitly called with
# IOLoop.current().stop()
# Other useful ioloop functions:
#
# tornado.ioloop.IOLoop.add_callback
# tornado.ioloop.IOLoop.add_future
# tornado.ioloop.IOLoop.spawn_callback
# tornado.ioloop.PeriodicCallback(task, 1000).start()
# Tornado examples:
#
# https://gist.github.com/lbolla/3826189
# https://github.com/tornadoweb/tornado/blob/master/demos/webspider/webspider.py
[pytest]
# norecursedirs = lib
addopts = -p no:cacheprovider
[flake8]
max_line_length = 120
exclude = lib,node_modules
ignore = E116
# https://pep8.readthedocs.io/en/latest/intro.html#error-codes