Options, Configuration & Logging

import datetime

from tornado.options import Error, OptionParser

options = OptionParser()
options.define("port", default=8888, type=int, group="server", help="port to listen on")
options.define("debug", default=False, help="development mode (never in production)")
options.define("request_timeout", default=datetime.timedelta(seconds=30), help="per-request budget")
options.define("opens_on", type=datetime.datetime, help="when the lending desk opens")
options.define("shards", type=int, multiple=True, help="database shards, ranges allowed")
options.define("origin", multiple=True, metavar="HOST", help="allowed browser origins")

rest = options.parse_command_line([
    "bookshelf.py",
    "--port=9000",
    "--debug",
    "--request-timeout=1h 30m",
    "--opens-on=2026-10-07 14:30",
    "--shards=1:3,7",
    "--origin=books.example,admin.example",
    "serve",
])
assert options.port == 9000 and options.debug is True
assert options.request_timeout == datetime.timedelta(minutes=90)
assert options.opens_on == datetime.datetime(2026, 10, 7, 14, 30)
assert options.shards == [1, 2, 3, 7]
assert options.origin == ["books.example", "admin.example"]
assert rest == ["serve"]        # everything after the last option is returned

options.parse_command_line(["bookshelf.py", "--debug=no"])
assert options.debug is True    # "no" is not one of false, 0 or f
options.parse_command_line(["bookshelf.py", "--debug=false"])
assert options.debug is False

options.parse_command_line(["bookshelf.py", "--origin=a", "--origin=b"])
assert options.origin == ["b"]  # a repeated flag replaces, it does not append

try:
    options.port = "9000"       # assignment is type-checked too
except Error as exc:
    print(exc)

Option 'port' is required to be a int (<class 'str'> given)

import contextlib
import io

for argv in (["bookshelf.py", "--no-such-flag=1"], ["bookshelf.py", "--port"], ["bookshelf.py", "--port=abc"]):
    with contextlib.redirect_stderr(io.StringIO()):          # silence the help text
        try:
            options.parse_command_line(argv)
        except (Error, ValueError) as exc:
            print(type(exc).__name__, "-", exc)

Error - Unrecognized command line option: 'no-such-flag'
Error - Option 'port' requires a value
ValueError - invalid literal for int() with base 10: 'abc'

import pathlib
import tempfile

with tempfile.TemporaryDirectory() as tmp:
    conf = pathlib.Path(tmp) / "bookshelf.conf"
    conf.write_text(
        "port = 9100\n"
        "origin = ['books.example', 'admin.example']\n"
        "shards = '1:2'\n"                 # strings are parsed like command-line values
        "request_timeout = '45s'\n"
        "here = __file__\n"                # not an option, so ignored
    )
    options.parse_config_file(str(conf))

assert options.port == 9100
assert options.origin == ["books.example", "admin.example"]
assert options.shards == [1, 2]
assert options.request_timeout == datetime.timedelta(seconds=45)

import os
import secrets

app_options = OptionParser()
app_options.define("port", default=8888, type=int)
app_options.define("debug", default=False)
app_options.define("origin", multiple=True)
app_options.define("config", type=str, help="optional config file")
app_options.define("cookie_secret", type=str, help="signing secret, from the environment only")


def load(argv, environ, prefix="BOOKSHELF_"):
    """Defaults < config file < environment < command line."""
    config_path = environ.get(prefix + "CONFIG")
    if config_path:
        app_options.parse_config_file(config_path, final=False)
    env_flags = [
        f"--{name}={environ[prefix + name.upper()]}"
        for name in app_options.as_dict()
        if name not in ("help", "config") and prefix + name.upper() in environ
    ]
    app_options.parse_command_line(["env", *env_flags], final=False)
    return app_options.parse_command_line(argv)       # the last call runs the callbacks


environ = {
    "BOOKSHELF_PORT": "9600",
    "BOOKSHELF_ORIGIN": "books.example,admin.example",
    "BOOKSHELF_COOKIE_SECRET": secrets.token_hex(32),   # injected by the platform, never committed
}
load(["bookshelf.py", "--port=9700"], environ)
assert app_options.port == 9700                         # the command line wins
assert app_options.origin == ["books.example", "admin.example"]
assert len(app_options.cookie_secret) == 64

SECRET_OPTIONS = {"cookie_secret"}


def effective_configuration(parser: OptionParser) -> dict:
    return {
        name: ("<hidden>" if name in SECRET_OPTIONS and value else value)
        for name, value in parser.as_dict().items()
        if name != "help"
    }


shown = effective_configuration(app_options)
assert shown["cookie_secret"] == "<hidden>" and shown["port"] == 9700

from unittest import mock

import tornado.options

tornado.options.define("shelf_size", default=100, type=int, help="books per shelf")
assert tornado.options.options.shelf_size == 100

with mock.patch.object(tornado.options.options.mockable(), "shelf_size", 3):
    assert tornado.options.options.shelf_size == 3
assert tornado.options.options.shelf_size == 100

import asyncio
import logging

from tornado import httpclient, httpserver, netutil, web
from tornado.log import access_log, app_log, gen_log


class Capture(logging.Handler):
    """Collect log records so the examples can print them."""

    def __init__(self):
        super().__init__()
        self.lines = []

    def emit(self, record):
        self.lines.append(f"{record.name} {record.levelname}: {record.getMessage()}")


capture = Capture()
tornado_logger = logging.getLogger("tornado")
tornado_logger.addHandler(capture)
tornado_logger.setLevel(logging.INFO)
tornado_logger.propagate = False

assert access_log.name == "tornado.access" and app_log.name == "tornado.application"
assert gen_log.name == "tornado.general" and access_log.parent is tornado_logger


class BooksHandler(web.RequestHandler):
    def get(self):
        app_log.info("listing books for %s", self.request.remote_ip)
        self.write({"books": []})


class BrokenHandler(web.RequestHandler):
    def get(self):
        raise RuntimeError("shelf database unavailable")


async def probe(app, *paths):
    server = httpserver.HTTPServer(app)
    sockets = netutil.bind_sockets(0, "127.0.0.1")
    server.add_sockets(sockets)
    base = f"http://127.0.0.1:{sockets[0].getsockname()[1]}"
    client = httpclient.AsyncHTTPClient()
    try:
        return [await client.fetch(base + path, raise_error=False) for path in paths]
    finally:
        server.stop()
        await server.close_all_connections()


app = web.Application([(r"/books", BooksHandler), (r"/broken", BrokenHandler)])
asyncio.run(probe(app, "/books", "/missing", "/broken"))
for line in capture.lines:
    print(line.splitlines()[0])

tornado.application INFO: listing books for 127.0.0.1
tornado.access INFO: 200 GET /books (127.0.0.1) 0.60ms
tornado.access WARNING: 404 GET /missing (127.0.0.1) 0.30ms
tornado.application ERROR: Uncaught exception GET /broken (127.0.0.1)
tornado.access ERROR: 500 GET /broken (127.0.0.1) 0.55ms

import json


def log_request(handler: web.RequestHandler) -> None:
    status = handler.get_status()
    level = logging.INFO if status < 400 else logging.WARNING if status < 500 else logging.ERROR
    access_log.log(level, json.dumps({
        "event": "request",
        "method": handler.request.method,
        "path": handler.request.path,          # the path only: query strings may carry tokens
        "status": status,
        "ms": round(1000 * handler.request.request_time(), 1),
        "ip": handler.request.remote_ip,
    }))


capture.lines.clear()
json_app = web.Application([(r"/books", BooksHandler)], log_function=log_request)
asyncio.run(probe(json_app, "/books?token=do-not-log"))
record = json.loads(next(line for line in capture.lines if line.startswith("tornado.access")).split(": ", 1)[1])
assert record["path"] == "/books" and record["status"] == 200 and "token" not in json.dumps(record)

from tornado.log import LogFormatter, define_logging_options, enable_pretty_logging

with tempfile.TemporaryDirectory() as tmp:
    logging_options = OptionParser()
    define_logging_options(logging_options)             # the same flags the global parser has
    logging_options.parse_command_line([
        "bookshelf.py",
        "--logging=debug",
        "--log-to-stderr=false",
        f"--log-file-prefix={tmp}/bookshelf-9000.log",
        "--log-file-max-size=1000000",
        "--log-file-num-backups=3",
    ], final=False)

    bookshelf_logger = logging.getLogger("bookshelf.demo")
    bookshelf_logger.propagate = False
    enable_pretty_logging(logging_options, logger=bookshelf_logger)   # applies the flags to one logger
    bookshelf_logger.info("rotating file handler is active")
    for handler in bookshelf_logger.handlers:
        handler.flush()

    files = sorted(path.name for path in pathlib.Path(tmp).iterdir())
    assert files == ["bookshelf-9000.log"]
    text = (pathlib.Path(tmp) / files[0]).read_text()
    assert text.startswith("[I ") and text.rstrip().endswith("rotating file handler is active")
    for handler in list(bookshelf_logger.handlers):
        handler.close()
        bookshelf_logger.removeHandler(handler)

formatter = LogFormatter(fmt="%(levelname)s %(name)s %(message)s", color=False)
assert formatter.format(logging.LogRecord("x", logging.INFO, "f", 1, "hi", None, None)) == "INFO x hi"

root_before = list(logging.getLogger().handlers)
quiet = OptionParser()
define_logging_options(quiet)
quiet.parse_command_line(["bookshelf.py", "--logging=none"])    # enable_pretty_logging returns immediately
assert quiet.logging == "none"
assert logging.getLogger().handlers == root_before               # nothing was added to the root logger