Examples

The rules language of Turq is documented only by example.

Basics

# This is a comment. Normal Python syntax.
if path == '/hello':
    header('Content-Type', 'text/plain')
    body('Hello world!\r\n')
else:
    error(404)

“RESTful” routing

if route('/v1/products/:product_id'):
    if GET or HEAD:
        json({'id': int(product_id),
              'inStock': True})
    elif PUT:
        # Pretend that we saved it
        json(request.json)
    elif DELETE:
        status(204)   # No Content

HTML pages

To get a simple page:

html()

If you want to change the contents of the page, the full Dominate library is at your service (dominate.tags imported as H):

with html():
    H.h1('Welcome to our site')
    H.p('Have a look at our ',
        H.a('products', href='/products'))

To change the <head>:

with html() as document:
    with document.head:
        H.style('h1 {color: red}')
    H.h1('Welcome to our site')

Request details

if request.json:     # parsed JSON body
    name = request.json['name']
elif request.form:   # URL-encoded or multipart
    name = request.form['name']
elif query:     # query string parameters
    name = query['name']
else:
    raw_name = request.body     # raw bytes
    name = raw_name.decode('utf-8')

# Header names are case-insensitive
if 'json' in request.headers['Accept']:
    json({'hello': name})
else:
    text('Hello %s!\r\n' % name)

Response headers

header() replaces the given header, so this will send only max-age:

header('Cache-Control', 'public')
header('Cache-Control', 'max-age=3600')

To add a header instead:

add_header('Set-Cookie', 'sessionid=123456')
add_header('Set-Cookie', '__adtrack=abcdef')

Custom status code and reason

status(567, 'Server Fell Over')
text('Server crashed, sorry!\r\n')

Redirection

if path == '/':
    redirect('/index.html')
elif path == '/index.html':
    html()

redirect() sends 302 (Found) by default, but you can override:

redirect('/index.html', 307)

Authentication

To demand basic authentication:

basic_auth()
with html():
    H.h1('Super-secret page!')

This sends 401 (Unauthorized) unless the request had Authorization with the Basic scheme (credentials are ignored).

Similarly for digest:

digest_auth()

And for bearer:

bearer_auth()

Body from file

text(open('/etc/services'))

Inspecting requests

To see what the client sends, including headers (but not the raw body), put debug() somewhere early in your rules:

debug()

and watch the console output. Alternatively, for even more diagnostics, run Turq with the --verbose option.

Or use mitmproxy.

Forwarding requests

Turq can act as a gateway or “reverse proxy”:

forward('httpbin.org', 80,  # host, port
        target)             # path + query string
# At this point, response from httpbin.org:80
# has been copied to Turq, and can be tweaked:
delete_header('Server')
add_header('Cache-Control', 'max-age=86400')

Turq uses TLS when connecting to port 443, but ignores certificates. You can override TLS like this:

forward('develop1.example', 8765,
        '/v1/articles', tls=True)

Cross-origin resource sharing

cors() adds the right Access-Control-* headers, and handles preflight requests automatically:

cors()
json({'some': 'data'})

For legacy systems, JSONP is also supported, reacting automatically to a callback query parameter:

json({'some': 'data'}, jsonp=True)

Compression

Call gzip() after setting the body:

with html():
    # 100 paragraphs of text
    for i in range(100):
        H.p(lorem_ipsum())
gzip()

Random responses

if maybe(0.1):    # 10% probability
    error(503)
else:
    html()

Response framing

By default, if the client supports it, Turq uses Transfer-Encoding: chunked and keeps the connection alive.

To use Content-Length instead of Transfer-Encoding, call content_length() after you’ve set the body:

text('Hello world!\r\n')
content_length()

To close the connection after sending the response:

add_header('Connection', 'close')
text('Hello world!\r\n')

Streaming responses

header('Content-Type', 'text/event-stream')
sleep(1)        # 1 second delay
chunk('data: my event 1\r\n\r\n')
sleep(1)
chunk('data: my event 2\r\n\r\n')
sleep(1)
chunk('data: my event 3\r\n\r\n')

Once you call chunk(), the response begins streaming. Any headers you set after that will be sent in the trailer part:

header('Content-Type', 'text/plain')
header('Trailer', 'Content-MD5')
chunk('Hello, ')
chunk('world!\n')
header('Content-MD5', '746308829575e17c3331bbcb00c0898b')

Handling Expect: 100-continue

with interim():
    status(100)

text('Resource updated OK')

In the above example, 100 (Continue) is sent immediately after the interim() block, but the final 200 (OK) response is sent only after reading the full request body.

If instead you want to send a response before reading the request body:

error(403)      # Forbidden
flush()

Custom methods

if method != 'FROBNICATE':
    error(405)   # Method Not Allowed
    header('Allow', 'FROBNICATE')

Switching protocols

if request.headers['Upgrade'] == 'QXTP':
    with interim():
        status(101)   # Switching Protocols
        header('Upgrade', 'QXTP')
        header('Connection', 'upgrade')
    send_raw('This is no longer HTTP!\r\n')
    send_raw('This is QXTP now!\r\n')

Anything else

In the end, Turq rules are just Python code that is not sandboxed, so you can import and use anything you like. For example, to send random binary data:

import os
header('Content-Type', 'application/octet-stream')
body(os.urandom(128))