Examples

Quick start

A whole web application of Pyjoyment.Lite:

from Pyjoyment.Lite import *


get('/', {'text': 'I ♥ Pyjoyment!'})

app.start()

Duct tape for the HTML5 web

A WebSocket service used by the template to extract the title from a website:

"""
__DATA__

@@ index.html.ep
% url = url_for('title')
<script>
  const ws = new WebSocket('<%= url.to_abs() %>');
  ws.onmessage = function (event) { document.body.innerHTML += event.data };
  ws.onopen    = function (event) { ws.send('https://mojolicious.org') };
</script>
"""

from Pyjoyment.Lite import *


# Render template "index.html.ep" from the DATA section
@get('/')
def index(c):
    c.render(template='index')


# WebSocket service used by the template to extract the title from a website
@websocket('/title')
def title(c):
    @c.on('message')
    def message(c, msg):
        c.ua().get_p(msg).then(lambda tx: c.send(tx.result.dom().at('title').text))


app.start()

Applications of other frameworks

A WSGI application, in a file of its own, under a path of an application of Pyjoyment.Lite:

from Pyjoyment.Lite import *


# The WSGI application of hello_wsgi.py, next to this script, under /wsgi
plugin('Mount', {'/wsgi': 'hello_wsgi:app'})

get('/', {'text': 'I ♥ Pyjoyment!'})

app.start()

The WSGI application, which also runs on its own, on the WSGI server of the standard library:

from wsgiref.simple_server import make_server


def app(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/plain')])
    return [f'Hello WSGI at {environ["SCRIPT_NAME"]} for {environ["PATH_INFO"]}!'.encode()]


if __name__ == '__main__':
    make_server('', 3000, app).serve_forever()

An application of Pyjoyment as the WSGI application, in the same file:

import Pyjo.Server.WSGI
import Pyjoyment

from Pyjoyment.Lite import *


# Hello world of Pyjoyment as a WSGI application, whose script name is the path it is mounted under
hello = Pyjoyment.new()
hello.routes.get('/').to(cb=lambda c: c.render(text=f'Hello Pyjoyment at {c.url_for("/")} on WSGI!'))
wsgi = Pyjo.Server.WSGI.new(app=hello).to_wsgi_app()

# The WSGI application under /wsgi of the application of Pyjoyment.Lite
plugin('Mount', {'/wsgi': wsgi})

get('/', {'text': 'I ♥ Pyjoyment!'})

# The application of a WSGI server too, which PYJO_WSGI tells: PYJO_WSGI=1 gunicorn mount_wsgi_inception:application
application = app.start()

An ASGI application, in a file of its own, under a path of an application of Pyjoyment.Lite, whose events stream at the pace of the client:

from Pyjoyment.Lite import *


# The ASGI application of hello_asgi.py, next to this script, under /asgi
plugin('Mount', {'/asgi': 'hello_asgi:app'})

get('/', {'text': 'I ♥ Pyjoyment!'})

app.start()

The ASGI application, which also runs on its own, on an ASGI server:

# On an ASGI server: uvicorn hello_asgi:app
async def app(scope, receive, send):
    if scope['type'] != 'http':
        return

    # Server-sent events, each sent once the one before has been written
    if scope['path'].endswith('/events'):
        await send({'type': 'http.response.start', 'status': 200, 'headers': [(b'content-type', b'text/event-stream')]})
        for n in range(3):
            await send({'type': 'http.response.body', 'body': f'data: {n}\n\n'.encode(), 'more_body': True})
        await send({'type': 'http.response.body', 'body': b''})
        return

    body = f'Hello ASGI at {scope["root_path"]} for {scope["path"]}!'.encode()
    await send({'type': 'http.response.start', 'status': 200, 'headers': [(b'content-type', b'text/plain')]})
    await send({'type': 'http.response.body', 'body': body})

An application of Pyjoyment as the ASGI application, in the same file:

import Pyjo.Server.ASGI
import Pyjoyment

from Pyjoyment.Lite import *


# Hello world of Pyjoyment as an ASGI application, whose root path is the path it is mounted under
hello = Pyjoyment.new()
hello.routes.get('/').to(cb=lambda c: c.render(text=f'Hello Pyjoyment at {c.url_for("/")} on ASGI!'))
asgi = Pyjo.Server.ASGI.new(app=hello).to_asgi_app()

# The ASGI application under /asgi of the application of Pyjoyment.Lite
plugin('Mount', {'/asgi': asgi})

get('/', {'text': 'I ♥ Pyjoyment!'})

# The application of an ASGI server too, which PYJO_ASGI tells: PYJO_ASGI=1 uvicorn mount_asgi_inception:application
application = app.start()

PHP on php-fpm, a FastCGI application, under a path of an application of Pyjoyment.Lite, with the scripts of its directory and the pool of php-fpm:

import Pyjo.App.FastCGI

from Pyjo.File import curfile
from Pyjoyment.Lite import *


# The scripts of PHP of the directory php under /php, on php-fpm with the pool of php/php-fpm.conf next to this script:
# php-fpm -n -y php/php-fpm.conf -F
fastcgi = Pyjo.App.FastCGI.new(url='fcgi://127.0.0.1:9000', document_root=curfile().sibling('php').to_str())
plugin('Mount', {'/php': fastcgi})

get('/', {'text': 'I ♥ Pyjoyment!'})

app.start()

<?php
// The script of the directory, and of every path under the mount when it is the only one, as a front controller is
if ($_SERVER['PATH_INFO'] === '/events') {
    header('Content-Type: text/event-stream');
    for ($n = 0; $n < 3; $n++) {
        echo "data: $n\n\n";
        flush();
    }
    exit;
}

header('Content-Type: text/plain');
echo "Hello PHP at {$_SERVER['SCRIPT_NAME']} for {$_SERVER['PATH_INFO']} ({$_SERVER['REQUEST_METHOD']} {$_SERVER['REQUEST_URI']})!\n";

<?php phpinfo(); ?>

; A pool of php-fpm in the foreground, for mount_fastcgi_php.py: php-fpm -n -y php/php-fpm.conf -F
[global]
error_log = /dev/stderr
daemonize = no

[www]
listen = 127.0.0.1:9000
pm = static
pm.max_children = 2

A CGI program of Python in the directory cgi-bin, under a path of an application of Pyjoyment.Lite, a process for each request:

import Pyjo.App.CGI

from Pyjo.File import curfile
from Pyjoyment.Lite import *


# The programs of the directory cgi-bin, next to this script, under /cgi-bin, a process for each request
plugin('Mount', {'/cgi-bin': Pyjo.App.CGI.new(document_root=curfile().sibling('cgi-bin').to_str())})

get('/', {'text': 'I ♥ Pyjoyment!'})

app.start()

#!/usr/bin/env python3
# A CGI program, which writes its response: the headers, an empty line and the body
import os
import urllib.parse


name = urllib.parse.parse_qs(os.environ.get('QUERY_STRING', '')).get('name', ['CGI'])[0]
print('Content-Type: text/plain; charset=UTF-8')
print()
print(f'Hello {name} at {os.environ["SCRIPT_NAME"]} for {os.environ["PATH_INFO"]}!')

PHP on php-cgi, with the scripts of the same directory as for php-fpm, a process of php-cgi for each request:

import Pyjo.App.CGI

from Pyjo.File import curfile
from Pyjoyment.Lite import *


# The scripts of PHP of the directory php under /php, on php-cgi of PATH, a process for each request
cgi = Pyjo.App.CGI.new(command=['php-cgi'], document_root=curfile().sibling('php').to_str(), script_extensions=['.php'])
plugin('Mount', {'/php': cgi})

get('/', {'text': 'I ♥ Pyjoyment!'})

app.start()

Servers

Pyjo.HelloWorld natively, on the web server of Pyjoyment:

import Pyjo.HelloWorld
import Pyjo.Server.Daemon


Pyjo.Server.Daemon.new(app=Pyjo.HelloWorld.new(), listen=['http://*:3000']).run()

On uvicorn, an ASGI server:

import Pyjo.HelloWorld
import Pyjo.Server.ASGI

import uvicorn


app = Pyjo.Server.ASGI.new(app=Pyjo.HelloWorld.new()).to_asgi_app()

uvicorn.run(app, port=3000)

On Werkzeug, a WSGI server:

import Pyjo.HelloWorld
import Pyjo.Server.WSGI

from werkzeug.serving import run_simple


app = Pyjo.Server.WSGI.new(app=Pyjo.HelloWorld.new()).to_wsgi_app()

run_simple('localhost', 3000, app)

As an event function of AWS Lambda:

import Pyjo.HelloWorld
import Pyjo.Server.Lambda


# The handler "hello_world_lambda.handler" of an event function of AWS Lambda
handler = Pyjo.Server.Lambda.new(app=Pyjo.HelloWorld.new()).to_lambda_app()

Requests on a timer

The user agent requests the server every second, in the same event loop, without blocking it:

import Pyjo.HelloWorld
import Pyjo.IOLoop
import Pyjo.Server.Daemon
import Pyjo.UserAgent


# The server and the user agent, in the same event loop, which close their sockets at the end of the block
with (
    Pyjo.Server.Daemon.new(app=Pyjo.HelloWorld.new(), listen=['http://*:3000']) as daemon,
    Pyjo.UserAgent.new() as ua,
):
    # A request every second, which does not block the event loop; a request that the end of the block interrupts gets
    # an error
    @Pyjo.IOLoop.recurring(1)
    def fetch(loop):
        def done(ua, tx):
            print(tx.error['message'] if tx.error else f'{tx.res.code} {tx.res.text}')

        ua.get('http://127.0.0.1:3000/', done)

    # The server runs the event loop until INT or TERM stops it
    daemon.run()

A request on a loopback

A blocking request of the user agent to the server, in the same event loop:

import Pyjo.HelloWorld
import Pyjo.IOLoop
import Pyjo.Server.Daemon
import Pyjo.UserAgent


# The server, on a port that the system picks, and the user agent, in the same event loop, which close their sockets at
# the end of the block
with (
    Pyjo.Server.Daemon.new(app=Pyjo.HelloWorld.new(), listen=['http://127.0.0.1'], silent=True) as daemon,
    Pyjo.UserAgent.new(ioloop=Pyjo.IOLoop.singleton) as ua,
):
    # A blocking request, which runs the event loop of the server until the response
    tx = ua.get(f'http://127.0.0.1:{daemon.start().ports()[0]}/')
    print(tx.result.code, tx.result.text)

Web scraping

import Pyjo.UserAgent

from Pyjo.String.Unicode import u


# Extract named character references from HTML spec
tx = Pyjo.UserAgent.new().get('https://html.spec.whatwg.org')
for n in tx.res.dom('#named-character-references-table tbody > tr').each():
    u(n.at('td > code').text + ' ' + n.children('td')[1].text).trim().say()

URL manipulation

import Pyjo.URL


# 'ssh+git://git@github.com/dex4er/Pyjoyment.git'
url = Pyjo.URL.new('https://github.com/dex4er/Pyjoyment')
print(url.set(scheme='ssh+git', userinfo='git', path=f'{url.path}.git').to_unsafe_str())

# 'https://pypi.org/search/?q=Pyjoyment&o=&c=Programming+Language+%3A%3A+Python'
url = Pyjo.URL.new('https://pypi.org/search/')
print(url.set(query=[('q', 'Pyjoyment'), ('o', ''), ('c', 'Programming Language :: Python')]))

Non-blocking TCP client/server

import Pyjo.IOLoop


# Listen on port 3000
@Pyjo.IOLoop.server(port=3000)
def server(loop, stream, cid):

    @stream.on
    def read(stream, chunk):
        # Process input chunk
        print(f'Server: {chunk.decode("utf-8")}')

        # Write response
        stream.write(b'HTTP/1.1 200 OK\x0d\x0a\x0d\x0a')

        # Disconnect client
        stream.close_gracefully()


# Connect to port 3000
@Pyjo.IOLoop.client(port=3000)
def client(loop, err, stream):

    @stream.on
    def read(stream, chunk):
        # Process input
        print(f'Client: {chunk.decode("utf-8")}')

    # Write request
    stream.write(b'GET / HTTP/1.1\x0d\x0a\x0d\x0a')


# Add a timer
@Pyjo.IOLoop.timer(3)
def timeouter(loop):
    print('Timeout')
    # Shutdown server
    loop.remove(server)


# Start event loop
Pyjo.IOLoop.start()

Standalone HTTP server serving embedded template file

"""
__DATA__

@@ index.html.ep
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>Pyjoyment</title>
</head>
<body>
<h1>♥ Pyjoyment ♥</h1>
<h2>This page is served by Pyjoyment framework.</h2>
<p><%= method %> request for <%= path %></p>
</body>
</html>
"""

import Pyjo.App
import Pyjo.GetOpt
import Pyjo.Server.Daemon
import Pyjo.Template
import Pyjo.URL

from Pyjo.Loader import data_section
from Pyjo.Util import die


class MyApp(Pyjo.App.object):
    # Embedded template file
    template = Pyjo.Template.new(auto_escape=True, vars=True).parse(data_section(__name__, 'index.html.ep'))

    # All the complexities of CGI, WSGI, ASGI, HTTP and WebSockets get reduced to a single method call!
    def handler(self, tx):
        # Resume transaction at the end of the block
        with tx:
            # Request
            method = tx.req.method
            path = tx.req.url.path

            # Response
            tx.res.code = 200
            tx.res.headers.content_type = 'text/html; charset=utf-8'
            tx.res.body = self.template.process({'method': method, 'path': path}).encode()


# Options --address and --port
getopt = Pyjo.GetOpt.new().option('a|address=s', default='0.0.0.0').option('p|port=i', default=3000)
opts = getopt.parse() or die('Usage: http_server.py [--address ADDRESS] [--port PORT]')

url = Pyjo.URL.new().set(scheme='http', host=opts.address, port=opts.port)

Pyjo.Server.Daemon.new(app=MyApp.new(), listen=[url]).run()