Server & Routing
Routing, request parsing, responses, cookies, file serving, file uploads, mailing and WebSockets.
Routing
Decorate any method with @Server.expose (or @BaseServer.expose)
to make it a public HTTP endpoint. The method name determines the URL path.
| Method name | URL path |
|---|---|
index | / |
default | Catch-all — handles any unmatched path |
api_posts | /api_posts |
user_profile | /user_profile |
from vesta import BaseServer
class App(BaseServer):
@BaseServer.expose
def index(self):
return "<h1>Home</h1>"
@BaseServer.expose
def api_ping(self):
return "pong"
@BaseServer.expose
def default(self):
self.response.code = 404
return "Not found"
method parameter in the body to
distinguish them if needed.
Request parameters
Parameters are injected as keyword arguments. Vesta reads them from query strings, JSON bodies, and multipart form data automatically.
@Server.expose
def api_search(self, query="", page="1", limit="20"):
# GET /api_search?query=python&page=2
results = self.db.getFilters("articles", ["title", "LIKE", f"%{query}%"])
return json.dumps(results)
Cookie access
@Server.expose
def api_whoami(self):
theme = self.response.cookies.get("theme", "light")
return f"Your theme: {theme}"
Raw request body
For JSON payloads sent as the body (not query params), declare a parameter with the same key as the JSON field:
# POST /api_create {"title": "Hello", "body": "World"}
@Server.expose
def api_create(self, title="", body=""):
...
Response
Configure the response via self.response before returning.
Status code
@Server.expose
def api_item(self, id=""):
item = self.db.getSomething("items", id)
if not item:
self.response.code = 404
return "Not found"
return json.dumps(item)
Content type
| Value | Content-Type header |
|---|---|
"html" (default) | text/html; charset=utf-8 |
"json" | application/json |
"plain" | text/plain; charset=utf-8 |
@Server.expose
def api_data(self):
self.response.type = "json"
return json.dumps({"ok": True})
Custom headers
self.response.headers.append(("X-My-Header", "value"))
Security headers
Vesta sets these on every response by default:
X-Content-Type-Options, X-Frame-Options,
X-XSS-Protection, Referrer-Policy,
Permissions-Policy.
Errors & redirects
code is an integer (400, 403, 404, 500…). message is sent as the response body.target (an absolute path or full URL).from vesta import Server, HTTPError, HTTPRedirect
@Server.expose
def api_private(self):
uid = self.getUser() # raises HTTPRedirect to /auth if not logged in
item = self.db.getSomething("items", self.request_id)
if not item:
raise HTTPError(self.response, 404, "item not found")
if item["owner"] != uid:
raise HTTPError(self.response, 403, "forbidden")
return json.dumps(item)
@Server.expose
def old_url(self):
raise HTTPRedirect(self.response, "/new_url")
Custom error pages
Map status codes to static HTML files in the features dict:
class App(Server):
features = {
"errors": {
404: "/static/404.html",
500: "/static/500.html",
}
}
Cookies
| Parameter | Type | Description |
|---|---|---|
name | str | Cookie name |
value | str | Cookie value |
exp | dict | {"value": 30, "unit": "days"} or "minutes". Omit for session cookie. |
httponly | bool | Prevent JS access. Default True. |
samesite | str | "Strict", "Lax" or "None". |
secure | bool | HTTPS only. Default True. |
@Server.expose
def api_set_theme(self, theme="light"):
self.response.set_cookie(
"theme", theme,
exp={"value": 365, "unit": "days"},
httponly=False, # JS needs to read this
samesite="Lax",
secure=False, # set True in production
)
self.response.type = "json"
return json.dumps({"ok": True})
File serving
PATH.from os.path import abspath, dirname
PATH = dirname(abspath(__file__))
class App(Server):
@Server.expose
def index(self):
return self.file(PATH + "/static/index.html")
@Server.expose
def about(self):
return self.file(PATH + "/static/about.html")
File uploads
"avatars/abc123.png"static/attachments/.
Returns the path relative to static/attachments/
which you can store in the DB and serve directly.
| Parameter | Description |
|---|---|
content | Data URI string: data:image/png;base64,iVBOR… |
name | Optional filename. Auto-generates SHA256 hash if empty. |
ext | Force a file extension, e.g. "jpg". Inferred from the data URI if omitted. |
category | Subdirectory under attachments/, e.g. "avatars". |
@Server.expose
def api_upload_avatar(self, image=""):
uid = self.getUser()
# `image` is the data URI from a FileReader on the frontend
path = self.saveFile(image, category="avatars")
self.db.edit("users", uid, "avatar", path)
self.response.type = "json"
return json.dumps({"path": f"/static/attachments/{path}"})
/, \ and
.. are stripped. A ValueError is raised if the sanitised name
would be empty.
Mailing
Available on Server (not BaseServer). Configure
SMTP credentials in server.ini [MAILING]. Emails are signed
with DKIM using the private key at mailing/dkim.txt.
DEBUG = true the mailing service is not initialised — OTP codes and reset
links are printed to the console instead. Set DEBUG = false to send real emails.
[MAILING]
MAILING_HOST = smtp.example.com
MAILING_PORT = 587
NOREPLY_PASSWORD = s3cr3t
template is a filename in mailing/.
The template is a plain HTML file — values is a tuple or list
passed to Python's str.format(), so use {} or
{0} placeholders.
mailVerif.html. Called internally by UniAuth.mailInvite.html.@Server.expose
def api_contact(self, email="", message=""):
self.sendTemplate(
template = "mailContact.html",
target = "support@myapp.com",
subject = "New contact message",
text = message, # plain-text fallback
values = (email, message)
)
self.response.type = "json"
return json.dumps({"ok": True})
Example mailing/mailContact.html:
<p>From: <strong>{0}</strong></p>
<p>{1}</p>
WebSockets
Enable WebSockets by setting features = {"websockets": True}
and adding a [NOTIFICATION] section to server.ini.
[NOTIFICATION]
PORT = 9877
URL = wss://myapp.com/ws
DEBUG_URL = ws://localhost:9877
import json
from vesta import Server
class App(Server):
features = {"websockets": True}
# Vesta auto-exposes /config — returns the WS URL as a JS module
# Vesta auto-exposes /authWS — authenticates a connection by connectionId
async def handle_message(self, websocket):
"""Override this to process incoming WS messages."""
async for raw in websocket:
data = json.loads(raw)
match data.get("type"):
case "ping":
await websocket.send(json.dumps({"type": "pong"}))
case "broadcast":
uid = self.getWsUser(websocket)
await self.sendNotificationAsync(uid, data["content"])
exclude to skip the sender.
initWebSockets() from
static/framework/websockets.mjs. The URL is fetched from
the /config endpoint automatically.
Configuration reference
[server]
IP = 0.0.0.0 # bind address (127.0.0.1 for local only)
PORT = 9876 # HTTP port
DEBUG = false # true → verbose errors, reload-friendly
DEFAULT_ENDPOINT = /dashboard # where to redirect after login
SERVICE_NAME = myapp # used by vesta service setup
[security]
SECRET_KEY = change-me-in-production # signs JWT tokens
[DB]
DB_USER = app_user
DB_PASSWORD = secret
DB_HOST = localhost
DB_PORT = 5432
DB_NAME = app_db
[UNIAUTH]
DB_NAME = uniauth
DB_HOST = localhost
DB_PORT = 5432
[MAILING]
MAILING_HOST = smtp.example.com
MAILING_PORT = 587
NOREPLY_PASSWORD = smtp-password
[NOTIFICATION]
PORT = 9877
URL = wss://myapp.com/ws
DEBUG_URL = ws://localhost:9877
server.ini to version control — it contains
database credentials and the JWT secret key. Add it to .gitignore.