Query Builder — Vesta Docs

Setup

Server always connects to the database at startup — just add a [DB] section to server.ini and self.db is available in every exposed method.

If you're using BaseServer instead, opt in explicitly with features = {"orm": True} in your class definition.

[DB]
DB_USER     = app_user
DB_PASSWORD = secret
DB_HOST     = localhost
DB_PORT     = 5432
DB_NAME     = app_db

from vesta import Server

class App(Server):
    @Server.expose
    def api_test(self):
        row = self.db.getSomething("users", 1)
        return str(row)

Note The database connection is created once at startup and reused across requests. Commits happen automatically after every INSERT, UPDATE and DELETE.

Insert

db.insertDict(table, dict, getId=False)

int | None

Insert a row from a Python dictionary. Column names are the dict keys. If the row already exists (primary key or unique constraint conflict) it is silently skipped. Returns the new row's ID when getId=True.

# Simple insert
self.db.insertDict("tags", {"name": "python", "color": "#3776ab"})

# Get the new ID back
post_id = self.db.insertDict("posts", {
    "title":   title,
    "content": content,
    "user_id": uid,
}, getId=True)

return json.dumps({"id": post_id})

db.insertReplaceDict(table, dict)

None

Upsert by id: inserts the row if it doesn't exist, or updates all fields if a row with the same id already exists. The dict must include an id key.

self.db.insertReplaceDict("settings", {
    "id":     uid,
    "theme":  "dark",
    "locale": "fr",
})

Select

db.getSomething(table, id, selector='id')

dict | [] — empty list when not found

Fetch a single row where selector = id. Returns a dict of column→value, or an empty list [] if not found. Always check with if row:, not if row is not None:.

# Get by primary key
user = self.db.getSomething("users", 42)

# Get by a different column
user = self.db.getSomething("users", "alice@example.com", selector="email")

db.getAll(table, id, selector='id')

list[dict] | [] — empty list when no rows match

Fetch all rows where selector = id. Useful for one-to-many relationships.

# All posts by user 42
posts = self.db.getAll("posts", 42, selector="user_id")

# All items in a category
items = self.db.getAll("items", "books", selector="category")

Filtered queries

db.getFilters(table, filter, basis=None)

list[dict] | []

Execute a WHERE clause built from a filter list. At least one condition is required. basis is an optional raw SQL string prepended to the WHERE clause (e.g. basis="TRUE" to match all rows).

The filter argument is a flat list of alternating conditions and logical operators:

filter = [
    column,    # column name
    operation, # =, !=, >, <, >=, <=, IN, LIKE, IS NULL, IS NOT NULL
    value,     # value to compare against
    "AND",     # or "OR" — links to next condition
    column2,
    operation2,
    value2,
    # ... repeat
]

Note Pass the string "true" or "false" (not Python booleans) for boolean columns. Pass Python None for NULL checks.

# Active users — boolean value as string "true"
users = self.db.getFilters("users", ["active", "=", "true"])

# Posts from 2024 with more than 10 likes
posts = self.db.getFilters("posts", [
    "year",  "=", 2024, "AND",
    "likes", ">", 10,
])

# Users whose name is NULL — "IS" with Python None maps to SQL NULL
nulls = self.db.getFilters("users", ["name", "IS", None])

# All rows — an empty filter selects everything
everything = self.db.getFilters("items", [])

IN operator

# Rows where status is one of several values
rows = self.db.getFilters("orders", ["status", "IN", ["pending", "processing"]])

Many-to-many (proxied queries)

db.getSomethingProxied(table, proxy, commonTable, id)

list[dict]

Follow a foreign key through a proxy table to fetch related rows. Useful for many-to-many relationships.

Given a schema like: users ← user_tags → tags

# Get all tags for user 42
# proxy = "user_tags", commonTable = "tags"
tags = self.db.getSomethingProxied("users", "user_tags", "tags", 42)

Update

db.edit(table, id, element, value, selector='id')

None

Update a single field on a row identified by selector = id.

# Update a field by primary key
self.db.edit("users", uid, "bio", "New bio text")

# Update by a different selector
self.db.edit("sessions", token, "last_seen", datetime.now(), selector="token")

Delete

db.deleteSomething(table, id, selector='id')

None

Delete all rows where selector = id.

self.db.deleteSomething("posts", post_id)

# Delete by a non-primary column
self.db.deleteSomething("sessions", uid, selector="user_id")

Table management

db.createTable(name, schema)

None

Create a table with the given SQL schema string. Typically called from db/initDB.py, not in route handlers.

db.resetTable(table)

None

Delete all rows from a table and reset its auto-increment sequence. Useful in tests.

self.db.createTable("posts", """
    id         SERIAL PRIMARY KEY,
    user_id    INTEGER REFERENCES users(id),
    title      TEXT NOT NULL,
    content    TEXT,
    created_at TIMESTAMP DEFAULT NOW()
""")

Common patterns

Paginated list

@Server.expose
def api_posts(self, page="1", limit="20"):
    uid = self.getUser()
    posts = self.db.getFilters("posts", ["user_id", "=", uid])
    page, limit = int(page), int(limit)
    start = (page - 1) * limit
    self.response.type = "json"
    return json.dumps(posts[start:start + limit])

Create and return

@Server.expose
def api_create_post(self, title="", content=""):
    uid = self.getUser()
    if not title:
        raise HTTPError(self.response, 400, "title required")
    post_id = self.db.insertDict("posts", {
        "title":   title,
        "content": content,
        "user_id": uid,
    }, getId=True)
    post = self.db.getSomething("posts", post_id)
    self.response.type = "json"
    return json.dumps(post)

Ownership check before update

@Server.expose
def api_edit_post(self, id="", title=""):
    uid = self.getUser()
    post = self.db.getSomething("posts", id)
    if not post:
        raise HTTPError(self.response, 404, "not found")
    if post["user_id"] != uid:
        raise HTTPError(self.response, 403, "forbidden")
    self.db.edit("posts", id, "title", title)
    self.response.type = "json"
    return json.dumps({"ok": True})