Tanka
Objects form a tree
Each small part has its own place
Nothing owns the whole
Mount and catch and route and wrap
The composition is all
True Object-Oriented Python Web Framework.
Designed for humans by human.
- Elegant
- Fully object oriented
- Embraces declarative style
- Extremely maintainable
A Tanka application is a single tree of objects. There are no decorators on functions, no global registries, no configuration files. You declare what your application is by composing objects, and the framework runs that declaration.
import asyncio
from tanka import Endpoint, Get, Html, Response, Route, Routes, Tanka, Uvicorn
class Index(Endpoint):
async def response(self, request):
return Response(Html("<h1>Hello, world</h1>"))
asyncio.run(
Tanka(
Routes(
Route(Get(), "/", Index()),
),
).run(Uvicorn("127.0.0.1", 8080)),
)
Table of Contents
- Installation
- Core Concepts
- Application
- Servers
- Endpoint
- Routing
- Request
- Response
- Bodies
- Templates
- Static Files
- Cookies
- Flash Messages
- Authentication and Authorization
- Error Handling
- CORS
- Compression
- OpenAPI
- Logging
- Testing Your Application
- How It All Fits Together
- Development
- How to Report Issues
1. Installation
With uv:
uv add tanka
With pip:
pip install tanka
Optional extras add integrations, for example uv add "tanka[uvicorn]" or
pip install "tanka[uvicorn]":
| Extra | Adds |
|---|---|
tanka[uvicorn] |
Uvicorn server |
tanka[hypercorn] |
Hypercorn server |
tanka[reload] |
Reload() hot reload through watchfiles |
tanka[openapi] |
OpenApi validation through openapi-core |
tanka[jinja] |
Jinja templates through jinja2 |
tanka[all] |
Everything above |
2. Core Concepts
| Concept | What it is | Examples |
|---|---|---|
| Tanka | The complete runnable application | Tanka(Routes(...)) |
| Server | Something that serves the application | Uvicorn(...), Hypercorn(...) |
| Endpoint | Anything that turns a Request into a Reply |
Index(), Routes(...), Authenticated(...), Cors(...) |
| Route | Binds an HTTP method and a path pattern to an endpoint | Route(Get(), "/users/{id}", UserPage(...)) |
| Request | The incoming HTTP message | request.target().path(), await request.body().json() |
| Reply | The outgoing HTTP message interface | Response(...), Redirect(...), WithCookie(...) |
| Body | The payload of a message | Html(...), Json(...), File(...), Stream(...) |
| Template | A named template that renders a Context into Html |
Template(Jinja("./templates"), "index.html") |
| Middleware | An endpoint that wraps another endpoint | Authenticated(...), Catch(...), Compressed(...) |
| Identity | Who is making the request | Principal("42", "admin"), Anonymous() |
| Abort | The one exception that ends a request with a status | raise Abort(404, "User not found") |
Every building block is small and immutable. Behaviour is added by wrapping one object into another, never by modifying an existing one.
3. Application
Tanka wraps a single endpoint. Usually that endpoint is a Routes
collection, possibly wrapped in middleware.
from tanka import Delete, Get, Post, Route, Routes, Tanka
app = Tanka(
Routes(
Route(Get(), "/", IndexPage(...)),
Route(Get(), "/users", UsersPage(...)),
Route(Post(), "/users", CreateUser(...)),
Route(Delete(), "/users/{id}", DeleteUser(...)),
),
)
Running with a built-in server
await app.run(Uvicorn("127.0.0.1", 8080))
With no argument, run() starts Uvicorn on 127.0.0.1:8000.
Running as ASGI
asgi() exposes a standard ASGI callable, so any ASGI server can host it.
# myapp.py
asgi = app.asgi()
uvicorn myapp:asgi
4. Servers
Server is an interface with two implementations.
from tanka import Hypercorn, Reload, Uvicorn
Uvicorn("127.0.0.1", 8080)
Hypercorn("0.0.0.0", 8443)
Uvicorn("127.0.0.1", 8080, Reload())
Reload() watches the current directory and restarts the process when a
file changes. Run the script directly for it to work:
python myapp.py
5. Endpoint
Endpoint is the one interface every request handler implements. It has a
single method, response, which receives a Request and returns a Reply.
from tanka import Endpoint, Json, Reply, Request, Response
class UserPage(Endpoint):
def __init__(self, db):
self.db = db
async def response(self, request: Request) -> Reply:
user = await PgUsers(self.db).user(
request.target().path().parameter("id")
)
return Response(Json({"id": user.id(), "name": user.name()}))
Pages, API handlers, static file servers, middleware, CORS, error catching and OpenAPI validation are all endpoints. Because they share one interface, any of them can wrap any other.
6. Routing
Route
A Route maps an HTTP method and a path to an endpoint.
Route(Get(), "/", IndexPage())
Route(Post(), "/users", CreateUser(...))
Available methods: Get, Post, Put, Patch, Delete, Head,
Options, and Verb("PURGE") for anything else.
Several methods on one route
Route(Methods(Get(), Head()), "/users", UsersPage(...))
Path parameters
Paths declare parameters with curly braces. Parsing is delegated to the parse library, so its type suffixes work too.
Route(Get(), "/users/{id}", UserPage(...))
Route(Get(), "/posts/{year:d}/{slug}", PostPage(...))
request.target().path().parameter("id") # "42"
request.target().path().parameter("year") # 2024
Mount
Mount attaches an endpoint under a path prefix. The prefix is stripped
before the inner endpoint sees the request. This is how versioned or grouped
APIs are built.
Routes(
Mount(
"/v1",
Routes(
Route(Get(), "/users", UsersV1(...)),
Route(Post(), "/users", CreateUserV1(...)),
),
),
Mount(
"/v2",
Routes(
Route(Get(), "/users", UsersV2(...)),
),
),
)
Routes are tried in order. The first one that matches wins. A request that
matches nothing ends with 404.
7. Request
A Request is the incoming HTTP message.
request.method().names() # ["GET"]
request.target().path() # Path, str() gives "/users"
request.target().path().parameter("id") # path parameter
request.target().query().parameter("page") # first value, fails if absent
request.target().query().values("tag") # all values, maybe empty
request.headers().header("accept") # first value, fails if absent
request.headers().values("accept") # all values, maybe empty
request.cookies().cookie("session") # cookie value, fails if absent
request.identity() # see Authentication
The body is decoded on demand.
raw = await request.body().bytes()
text = await request.body().text()
data = await request.body().json()
You can build a Request yourself, for example in tests.
from tanka import Empty, Get, Headers, Request, Text
Request(Get(), "/users?page=2", Headers({"accept": "text/html"}), Empty())
Request(Post(), "/users", Headers(), Text('{"name": "Ann"}'))
8. Response
Reply is the interface of an outgoing message: a status, headers and a
body. Response is the plain implementation.
Response(Html("<h1>Hello</h1>")) # status 200
Response(201, Json({"id": 7}))
Response(200, Headers({"cache-control": "no-store"}), Text("fresh"))
Redirect
Redirect("/login") # 302
Redirect("/users", 303)
Adding headers
Replies are wrapped, never modified.
WithHeaders(
Response(Json(items)),
Headers({"x-total-count": str(len(items))}),
)
Cookies and flash messages are added the same way, see below.
9. Bodies
| Body | Purpose |
|---|---|
Html("<p>hi</p>") |
HTML document |
Json({"ok": True}) |
JSON payload |
Text("plain") |
Plain text |
Empty() |
No body, for 204 and redirects |
Raw(b"...", "image/png") |
Bytes with an explicit content type |
File("./report.pdf") |
A single file, content type guessed |
Stream(chunks, "text/csv") |
Chunks from any async iterable |
Streaming a large export without buffering it:
class Export(Endpoint):
async def response(self, request):
return Response(Stream(self.rows(), "text/csv"))
async def rows(self):
yield b"id,name\n"
async for user in PgUsers(self.db).users():
yield f"{user.id()},{user.name()}\n".encode()
Every body contributes its own headers, such as content-type and
content-length. Headers given to Response take precedence.
10. Templates
Template renders a named template into an Html body. The engine behind
it implements Templates. Jinja is the engine that ships with Tanka. It
needs the jinja extra.
from tanka import Context, Jinja, Pair, Template
templates = Jinja("./templates")
Context is an immutable collection of named values. Pair names one of
them.
Response(
200,
await Template(templates, "account.html").html(
Context(
Pair("user", user),
Pair("title", "Account"),
Pair("page", 1),
)
),
)
Domain objects
Tanka does not hand arbitrary objects to the engine. A value with an
async def json(self) -> dict is replaced by that dict before rendering,
so account.html reads {{ user.name }}. Plain values such as str,
int and dict pass through unchanged. The capability is structural: your
own JsonReadable base class works without inheriting anything from Tanka.
class User:
def __init__(self, id: str, name: str):
self.id = id
self.name = name
async def json(self) -> dict:
return {"id": self.id, "name": self.name}
Example endpoint
from tanka import (
Context,
Endpoint,
Get,
Jinja,
Pair,
Reply,
Request,
Response,
Route,
Routes,
Tanka,
Template,
Templates,
)
class Index(Endpoint):
def __init__(self, templates: Templates):
self.templates = templates
async def response(self, request: Request) -> Reply:
return Response(
200,
await Template(self.templates, "index.html").html(
Context(
Pair("title", "Welcome"),
Pair("page", 1),
)
),
)
templates = Jinja("./templates")
app = Tanka(
Routes(
Route(Get(), "/", Index(templates)),
),
)
With templates/index.html:
<h1>{{ title }}</h1>
<p>Page {{ page }}</p>
GET / returns 200 with content-type: text/html; charset=utf-8 and the
rendered markup as the body.
Jinja
Jinja(directory) loads templates from a folder and escapes values in
.html, .htm and .xml templates. A missing or broken template raises an
Exception chained from the engine error, which ends the request with
500. Pass a ready jinja2.Environment instead of a directory when you
need custom filters or another loader.
Jinja(Environment(loader=PackageLoader("myapp"), autoescape=True))
Other engines implement Templates without touching Template.
class Mako(Templates):
async def markup(self, name: str, values: dict) -> str:
...
11. Static Files
Static serves files from a Files source. Directory is the source for a
local folder. Combine it with Mount so paths are relative to the folder.
Routes(
Mount("/static", Static(Directory("./public"))),
)
GET /static/css/app.css serves ./public/css/app.css. A folder serves its
index.html. Paths that escape the root or point to nothing end with 404.
Other sources are added by implementing Files, without touching Static.
class S3Bucket(Files):
def __init__(self, bucket: str):
self.bucket = bucket
def file(self, path: str) -> Body:
...
12. Cookies
Reading
session = request.cookies().cookie("session")
Writing
Wrap the reply in WithCookie.
WithCookie(
Redirect("/"),
Cookie(
"session",
token,
HttpOnly(),
Secure(),
SameSite("Lax"),
CookiePath("/"),
Lifetime(3600),
Domain("example.com"),
),
)
Deleting
WithCookie(Redirect("/login"), ForgetCookie("session", CookiePath("/")))
Several cookies are set by nesting WithCookie.
13. Flash Messages
A flash message travels to the next page through a cookie.
WithFlash(
Redirect("/users"),
Flash("User created", Success()),
)
Kinds: Success(), Failure(), Notice(), Alert().
The next page reads and clears it.
class UsersPage(Endpoint):
async def response(self, request):
notes = [
f"<p class='{flash.kind().name()}'>{flash.text()}</p>"
for flash in Flashes(request.cookies())
]
return WithCookie(
Response(Html("".join(notes) + await self.table())),
ForgetCookie("flash", CookiePath("/")),
)
14. Authentication and Authorization
Both are middleware: endpoints that wrap another endpoint.
Identity source
Tanka does not ship identity sources. Your application implements
IdentitySource: given a request, return an Identity or raise Abort.
from tanka import Abort, Identity, IdentitySource, Principal
class SessionIdentity(IdentitySource):
def __init__(self, db):
self.db = db
async def identity(self, request) -> Identity:
try:
token = request.cookies().cookie("session")
except Exception as error:
raise Abort(401, "Please log in") from error
user = await PgSessions(self.db).user(token)
return Principal(user.id(), *user.roles())
Authenticated
Authenticated asks the source for an identity, attaches it to the request
and passes the request on.
Route(
Get(),
"/profile",
Authenticated(ProfilePage(...), SessionIdentity(...)),
)
Inside the endpoint:
request.identity().id() # "42"
request.identity().roles() # ["admin", "editor"]
A request that did not pass through Authenticated carries Anonymous():
id() fails fast and roles() is empty.
Authorized
Authorized checks a requirement against the identity and ends with 403
when it is not met. Place it inside Authenticated.
Authenticated(
Authorized(AdminPage(...), Role("admin")),
SessionIdentity(...),
)
Requirements compose:
Role("admin") # exactly this role
Roles("admin", "editor") # any of these roles
AnyOf(Role("owner"), Roles("admin", "root")) # any requirement
AllOf(Role("staff"), Role("verified")) # every requirement
Custom rules implement Requirement:
class Verified(Requirement):
def matches(self, identity: Identity) -> bool:
return "unverified" not in identity.roles()
15. Error Handling
Abort
Abort is the single exception for ending a request with an HTTP error. It
is never caught inside endpoints; it propagates up to the application, which
turns it into a reply.
class UserPage(Endpoint):
async def response(self, request):
try:
user = await PgUsers(self.db).user(
request.target().path().parameter("id")
)
except Exception as error:
raise Abort(404, "User not found") from error
return Response(Html(user.page()))
Abort accepts 4xx and 5xx codes from the standard table (400, 401,
403, 404, 405, ..., 500, 501, 502, 503, 504, 505). Any
other code breaks the request and ends with 500.
Default behaviour
- Any exception that is not
Abortends with500. - A request that matches no route ends with
404. - Without a custom page, the framework answers with a small HTML page
naming the error:
Not Found,Internal Server Error, and so on.
Catch and On
Catch maps status codes to your own error endpoints.
Tanka(
Catch(
Routes(
Route(Get(), "/", IndexPage(...)),
),
On(404, NotFoundPage()),
On((401, 403), AccessDeniedPage()),
On(Range(500, 599), ServerErrorPage()),
On(..., AnyErrorPage()),
),
)
On accepts a single code, a tuple of codes, a Range, or ... for
everything. The first matching On answers. Codes nothing matches fall
through to the default pages.
Fallback
An error page that needs to know what went wrong implements Fallback
instead of Endpoint. Its response receives the request and an Abort
carrying the status and the message, which is how a JSON API turns every
failure into one error format. A plain exception arrives as Abort(500, message).
from tanka import Abort, Catch, Fallback, Json, On, Response
class Enveloped(Fallback):
async def response(self, request, error: Abort):
return Response(error.status(), Json({"error": str(error)}))
Catch(routes, On(..., Enveloped()))
A plain Endpoint given to On is wrapped in Indifferent, which answers
the same way whatever the error was.
16. CORS
Cors wraps an endpoint with a set of policies. Preflight requests are
answered without reaching the wrapped endpoint.
Tanka(
Cors(
Routes(
Route(Get(), "/users", Users(...)),
Route(Post(), "/users", CreateUser(...)),
),
AllowOrigins("https://app.example.com"),
AllowMethods(Get(), Post()),
AllowHeaders("Content-Type", "Authorization"),
AllowCredentials(),
ExposeHeaders("X-Total-Count"),
MaxAge(600),
),
)
AllowOrigins("*") allows every origin. Browsers never accept a wildcard
origin together with credentials, so Cors refuses that combination: a
cross-origin request answered with both AllowOrigins("*") and
AllowCredentials() raises an Exception instead of sending headers that
would fail silently in the browser. List the allowed origins explicitly when
credentials are needed.
17. Compression
Compressed wraps an endpoint and gzips every reply the client is willing
to accept. Wrap a single route or a whole subtree; the endpoints inside
know nothing about it.
Tanka(
Compressed(
Routes(
Route(Get(), "/", Index()),
Route(Get(), "/report", Report(...)),
),
),
)
A reply is compressed when the request carries accept-encoding with
gzip (or *), the body is at least 500 bytes long and no
content-encoding is set already. Every other reply passes through
untouched. The second argument changes the minimum size:
Compressed(Routes(...), 2048)
A compressed reply sends content-encoding: gzip and vary: accept-encoding, drops content-length and streams compressed chunks, so
large files and streams are never buffered in memory. Gzipped is the
body behind it and can be used on its own.
18. OpenAPI
OpenApi wraps an endpoint with an OpenAPI specification file (YAML or
JSON). It needs the openapi extra.
from tanka import OpenApi
Tanka(
OpenApi(
"openapi.yaml",
Routes(
Route(Get(), "/users", Users(...)),
Route(Post(), "/users", CreateUser(...)),
),
),
)
It provides three things:
- Documentation at
/docs, rendered with Stoplight Elements, and the specification itself at/openapi.json. Its "Try It" panel keeps and sends cookies for same-origin calls, so cookie-based flows work from the page. - Request validation: a request that violates its operation schema ends
with
400before reaching the endpoint; one that misses a declared security requirement ends with401. - Response validation: a reply that violates the schema ends with
500.
Paths the specification does not describe pass through untouched. The endpoint itself stays unaware of validation.
19. Logging
Unhandled exceptions are logged with their traceback before the 500 page
is sent. Tanka logs through the standard logging module under the
tanka logger by default. Pass a Log to change that.
Tanka(routes, Logging("myapp.http")) # a named stdlib logger
Tanka(routes, Silence()) # nothing, handy in tests
20. Testing Your Application
Endpoints are plain objects. Call them with a hand-made Request.
async def test_greets_by_name():
reply = await Route(Get(), "/hello/{name}", Greeting()).response(
Request(Get(), "/hello/Ann", Headers(), Empty())
)
assert_that(await Body.Smart(reply.body()).text(), equal_to("Hi, Ann"))
For the whole tree, drive the ASGI application with httpx.
async def test_serves_index():
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=Tanka(routes, Silence()).asgi()),
base_url="http://test",
) as http:
assert_that((await http.get("/")).status_code, equal_to(200))
Replace real dependencies with fake objects that implement the same
interfaces, such as a fake IdentitySource or a fake Files.
21. How It All Fits Together
Every piece of Tanka is an Endpoint or wraps one. An application is a tree
where each layer adds one concern and delegates the rest inward.
Tanka
└── Catch error pages
└── Cors cross-origin policy
└── Compressed gzip bodies
└── OpenApi documentation + validation
└── Routes dispatch by method + path
├── Route → Authenticated → Authorized → AdminPage
├── Mount("/static") → Static(Directory("./public"))
└── Mount("/v1") → Routes → ...
Replies follow the same idea. A plain Response is wrapped to add cookies,
headers or flash messages.
WithFlash
└── WithCookie
└── Redirect("/users")
The result is an application you can read top-down as a single expression, and change by swapping or wrapping one object at a time.
22. Development
uv sync
make help
| Command | Purpose |
|---|---|
make unit |
Unit tests with coverage |
make deep |
Integration tests against live servers |
make lint |
black, flake8 and ruff |
23. How to Report Issues
Enhancements
Open a GitHub issue and label it enhancement. Describe the desired
behaviour and why it is useful. No code is required.
Bugs in Code
Open a pull request, not an issue. The pull request must contain a test that
reproduces the bug and fails against the current code. Mark the test as
disabled with pytest.mark.skip and a short reason, so CI stays green while
the failing case is on record:
@pytest.mark.skip(reason="Reproduces #6, unskip once fixed")
async def test_passes_value_with_plain_json_method_through_unchanged():
...
The fix, if any, can arrive in the same or a follow-up pull request, which removes the skip. Contributors without push rights fork the repository first.
Bugs Outside Code
If the bug cannot be reproduced with a test (documentation, packaging, CI
configuration, and so on), open a GitHub issue and label it bug. Describe
the expected and actual behaviour and how to observe it.
Release files for tanka 0.0.3
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| tanka-0.0.3.tar.gz | 149.2 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| tanka-0.0.3-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 180.0 kB
Release files / tanka-0.0.3.tar.gz
| Download URL | tanka-0.0.3.tar.gz |
|---|---|
| Size | 149.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
88c5beb80f3e432cf507cb02acc4a0248fc9dd83acf3374312041bd2a72a8428
|
|
BLAKE2b-256 checksum How to use checksums |
60ad2d7c5f0f3553cd6df986d9c27d1061e648b034ee7f98f684d94b5f9ceab3
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.12.16 {"installer":{"name":"uv","version":"0.12.16","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / tanka-0.0.3-py3-none-any.whl
| Download URL | tanka-0.0.3-py3-none-any.whl |
|---|---|
| Size | 30.9 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
f8265a61e041658fa57f63d63bc2dceba8a51fc2b61f53306f28ebe3695fa89a
|
|
BLAKE2b-256 checksum How to use checksums |
eebe13f5fbc29493de466acc2fc406b1055ce8fc5293cf06d0c0795fdc904d08
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.12.16 {"installer":{"name":"uv","version":"0.12.16","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|