[Python Tornado Example Application] #python #python3 #tornado #example #basic #simple

1. README.md

There are two examples...

1. Simple
2. Advanced

2. Simple Example.py

import os

import tornado.autoreload
import tornado.httpclient
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import uvloop

app_settings = {
    "env": os.environ.get("ENV", "dev"),
    "port": os.environ.get("APP_PORT", "9000"),
}

http_client = tornado.httpclient.AsyncHTTPClient()


class FooHandler(tornado.web.RequestHandler):
    async def get(self, *args, **kwargs):
        resp = await http_client.fetch("https://httpbin.org/get")
        self.finish(resp.body)


class App(tornado.web.Application):
    def __init__(self):
        super().__init__([
            (r"/", FooHandler),
        ], **app_settings)


if __name__ == "__main__":
    uvloop.install()

    tornado.options.options.logging = None  # configure tornado to leave logging alone

    if app_settings["env"] == "dev":
        tornado.autoreload.start()
        App().listen(app_settings["port"])
    else:
        server = tornado.httpserver.HTTPServer(App())
        server.bind(app_settings["port"])
        server.start(0)  # multi process mode (one process per cpu)

    tornado.ioloop.IOLoop.current().start()

3. Python Tornado Example Application.py

#!/usr/bin/env python

"""This is our complete Tornado application.

Example endpoints...

curl localhost:9000/headers
curl localhost:9000/static/foo.js
curl localhost:9000/
"""

import logging
import os
import structlog

# from tornado import gen
from tornado.httpclient import AsyncHTTPClient
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.options import options
from tornado.process import cpu_count
from tornado.web import Application, RequestHandler, StaticFileHandler
from tornado.web import ErrorHandler as BaseErrorHandler

app_settings = {
    "compress_response": True,
    "default_handler_args": dict(status_code=404),
    "default_handler_class": ErrorHandler,
    "env": os.environ.get("ENV", "dev"),
    "log_level": os.environ.get("LOG_LEVEL", "INFO"),
    "port": os.environ.get("APP_PORT", "9000"),
    "static_path": os.path.join(os.path.dirname(__file__), "static"),
    "static_handler_class": RobotsHandler,  # probably should be more generically named
    "template_path": os.path.join(os.path.dirname(__file__), "templates"),
    # debug=True (can also use explicit: tornado.autoreload.start())
  	#
  	# Example:
  	#
  	# if settings.get("cluster") == "dev":
    #     tornado.autoreload.start()
}

logging.basicConfig(
    level=getattr(logging, app_settings["log_level"]),
    format="[%(levelname)s %(asctime)s path:%(pathname)s lineno:%(lineno)s] %(message)s",  # noqa
    datefmt="%Y/%m/%d %I:%M:%S"
)

structlog.configure(logger_factory=structlog.stdlib.LoggerFactory())
log = structlog.get_logger()


class HeadersHandler(RequestHandler):
    """Serves application header information."""

    # Following method is here to demonstrate older Tornado syntax
    # Pre-async/await support version, where you needed Tornado's gen module
    #
    # @gen.coroutine
    # def get(self, *args, **kwargs):
    #     """A function that will return the content for the health endpoint.""" # noqa
    #     log.info("Endpoint hit", handler="health")
    #     http_client = AsyncHTTPClient()
    #     response = yield http_client.fetch("https://httpbin.org/headers")
    #     self.finish(response.body)

    def initialize(self, some_data_object):
        self.some_data_object = some_data_object
    
    async def get(self, *args, **kwargs):
        """A function that will return the content for the header endpoint."""
        log.info("Endpoint hit", handler="health")
        http_client = AsyncHTTPClient()
        response = await http_client.fetch("https://httpbin.org/headers")
        self.finish(response.body)

    def data_received(self, chunk):
        """Defined to avoid abstract-method lint issue."""
        pass


class HomeHandler(RequestHandler):
    """Serves application home information."""

    async def get(self, *args, **kwargs):
        """A function that will return the content for the home endpoint."""
        log.info("Endpoint hit", handler="home")
        self.render(
            "index.html",
            header_text="Header goes here",
            footer_text="Footer goes here"
        )

    def data_received(self, chunk):
        """Defined to avoid abstract-method lint issue."""
        pass


class RobotsHandler(StaticFileHandler):
    async def get(self, path="robots.txt", include_body=True):
        return await super().get(path, include_body)

    def set_default_headers(self):
        self.clear_header("Server")

    def set_extra_headers(self, path):
        self.set_header("foo", "bar")
        
        
class ErrorHandler(BaseErrorHandler):
    def write_error(self, status_code, **kwargs):
        log = logger.bind(status=status_code)
        log.error("incoming request")
        
        # notice these methods... 
        self.set_default_headers()
        self.set_extra_headers()
        # ...are no longer automatically called
        #
        # so I manually call them
        #
        # i believe this is because tornado.web.ErrorHandler doesn't call the parent RequestHandler __init__ 
        
        self.write("Error %s" % status_code)

    def set_default_headers(self):
        self.clear_header("Server")

    def set_extra_headers(self, path=None):
        for header, value in get_headers(self.get_status(), key=path):
            self.set_header(header, value)
      
 
def robots_path() -> str:
    return os.path.join(os.path.dirname(__file__), "static", "robots.txt")


class App(Application):
    """Base Application class to define the app's routes and settings."""

    def __init__(self):
        """Setup the routes and import the application settings."""
        app_handlers = [
            (r"/", HomeHandler),
            (r"/headers", HeadersHandler, {"some_data_object": ["i", "could", "be", "a", "class"]}),
            (r"/robots.txt", RobotsHandler, {"path": robots_path()}),  # MUST set `static_handler_class` in app_settings
        ]

        super().__init__(app_handlers, **app_settings)


if __name__ == "__main__":
    options.logging = None  # configure tornado to leave logging alone
    log = log.bind(port=app_settings["port"])

    if app_settings["env"] == "dev":
        log.info("Starting applications", mode="single")
        App().listen(app_settings["port"])
    else:
        log.info("Starting applications", mode="forked", cpu_count=cpu_count())
        server = HTTPServer(App())
        server.bind(app_settings["port"])
        server.start(0)  # multi process mode (one process per cpu)

    IOLoop.current().start()

3. static - foo.js

alert('hello');

3. templates - index.html

{{ static_url("foo.js") }}
<!--
creates a hash based on the content of the file and appends it to the end of the URL
/static/foo.js?v=da9209eb23a61fd3633a1fd140069bae
-->

{% extends layout.html %}
<!--
alternatively we could 'include' another file (path is relevative to the current 'templates' directory):

{% include partials/_head.html %}

...so the file would be located: templates/partials/_head.html
-->

{% block header %}
    <h1>{{ header_text }}</h1>
{% end %}

{% block body %}
    <p>Hello from the child template!</p>
{% end %}

{% block footer %}
    <p>{{ footer_text }}</p>
    {% set mailLink = "<a href=\"mailto:[email protected]\">Contact Us</a>" %}
    <p>{{ mailLink }}</p>
    <p>{% raw mailLink %}</p>
{% end %}

3. templates - layout.html

<html>
<body>
    <header>
        {% block header %}{% end %}
    </header>
    <content>
        {% block body %}{% end %}
    </content>
    <footer>
        {% block footer %}{% end %}
    </footer>
</body>
</html>