velocity-python

Overview

• 0.2.5Current release
• MITLicense
• 3.11 and upPython
• PostgreSQLDatabase
• January 2024On PyPI since
• 3,620Unit tests

velocity-python is a thin, opinionated layer between Python and a relational database, plus the pieces a serverless system on AWS needs around it. It sits between a heavy ORM, which hides the database behind classes and sessions, and a raw driver, which leaves you writing SQL strings and managing connections by hand.

The idea is simple. A table is a name. A row is a dictionary. Every read and write runs inside an explicit transaction that commits when your function returns and rolls back when it raises. Missing tables and columns are created as the application writes to them during development, and the schema is declared and locked for production.

The package is velocity-python on the Python Package Index; the import name is velocity. Its only required dependency is sqlparse; everything else is an optional extra.

Design principles

Install

pip install "velocity-python[postgres]"

Extras add what a project needs and nothing else:

Connecting

from velocity.db.servers.postgres import initialize

# Reads DBHost, DBDatabase, DBUser, DBPassword and DBPort
engine = initialize()

The engine is created once per process and holds a thread-safe connection pool (one to five connections by default), so a warm Lambda reuses its connections instead of opening new ones. Connections require TLS by default; only a loopback host may fall back to plain. Credentials are masked whenever configuration is logged.

Timeouts are set on every connection: connect, statement and idle-in-transaction, each configurable. Connections are validated and kept alive, and a dropped connection is retried with backoff.

Transactions

@engine.transaction
def create_order(tx, customer_email, items):
    customer = tx.table("customers").find({"email": customer_email})
    order = tx.table("orders").insert({
        "customer_id": customer["sys_id"],
        "status": "pending",
        "total": sum(i["price"] for i in items),
    })
    tx.table("order_items").insert_many([
        {"order_id": order["sys_id"], "product": i["name"], "price": i["price"]}
        for i in items
    ])
    return order

The decorator injects a transaction, or reuses one passed in, so functions compose: a helper called from inside a transaction joins it rather than opening a second one. The context manager form, with engine.transaction() as tx:, does the same for a block.

• Commit and rollback. Returning commits; raising rolls back. A finished transaction refuses further use.
• Retries. Deadlocks, transient connection errors and primary-key sequence collisions are retried automatically, classified by the database's own error codes, within a time budget.
• Side effects. Call tx.mark_external_side_effect() before charging a card or sending an email, and the transaction will never be retried after it.
• After commit. tx.on_commit(fn) runs work only once the data is safely written, which is how queue messages are sent.
• Savepoints and locks. Nested savepoints, row locks with lock=True or skip_locked=True, advisory locks, and timed leases with claim and release for work that several workers share.

Tables and rows

@engine.transaction
def demo(tx):
    users = tx.table("users")

    row = users.insert({"name": "Bob", "email": "bob@example.com"})

    user = users.row(row["sys_id"])                    # by primary key
    user = users.find({"email": "bob@example.com"})    # by lookup

    user["name"] = "Robert"                            # writes through immediately

    users.delete({"sys_id": row["sys_id"]})

Filters are dictionaries. The key names the column, with an optional operator in front; the value is always a bound parameter:

users.select(where={"status": "active", "role": "admin"})   # (status = %s) AND (role = %s)
users.select(where={"deleted_at": None})                     # IS NULL
users.select(where={"!deleted_at": None})                    # IS NOT NULL
users.select(where={"status": ["active", "pending"]})        # IN (...)
users.select(where={">=age": 18, "<created_at": "2024-01-01"})
users.select(where={"@@or": [{"status": "open"}, {"status": None}]})

Every predicate is wrapped in parentheses, and a key that doesn't parse raises an error rather than being guessed at. A key like owner_sys_id>name follows a foreign key and filters on the related row.

• Writes. insert, update, merge (an upsert, also called upsert), and the bulk forms insert_many, upsert_many and update_many.
• Rows. Assigning to a row writes through. with row.batch_update(): collects several changes into one statement.
• Results. all(), one(), scalar(), and conversions to dicts, tuples or row objects.
• JSON. JSON columns come back as live lists and dicts that write through when changed.

System columns

Every table velocity manages carries the same system columns, kept current by a database trigger, so every row can say when it was created, when and by whom it was last changed, and whether it is waiting to be processed.

Naming follows one rule throughout: sys_id is a row's identity, something_sys_id points at another row, and something_key is a business identifier.

The sys_dirty flag drives an asynchronous pipeline. A change marks a row dirty; a worker picks dirty rows up, does the slow work (sending, syncing, recalculating) and clears the flag. The work never blocks the person who made the change.

Schema management

velocity supports three ways of managing a schema, meant to be used together.

1. Create on write, in development. When a write names a table or column that doesn't exist, velocity creates it, under an advisory lock, and retries the write. A new field is a new key in a dictionary.
2. Declare it, at deploy time. ensure_schema declares a table's columns, indexes and foreign keys, and applies only what is missing. It runs the same way in every environment, so a fresh environment never depends on what development happened to create.
3. Lock it, in production. With the schema locked, a write to a missing column fails instead of altering the database. A lock setting that is on always locks; nothing can unlock it by accident.

@engine.transaction
def sync_schema(tx):
    tx.table("departments").ensure_schema(
        columns={"name": str},
    )

Declarations never drop, rename or retype a column. Those changes are deliberate and are written as migrations, one reviewed step at a time.

Safety model

Safety is built in at two levels: the database layer, which server code uses, and the client boundary, which every request from a browser passes through first.

In the database layer

• Every value is bound as a parameter. A value is never pasted into SQL.
• Write keys must be column names, and a value can never become a column type.
• A value starting with @@ is SQL only for a short fixed list of tokens, like @@CURRENT_TIMESTAMP. Any other @@ text is stored exactly as written.
• Real SQL from server code is marked explicitly, with Literal(...) or Expr(sql, params), so it can be found and reviewed.

sms.insert({"body": "@@(select password from users)"})   # stores that text, nothing more
sms.insert({"received_at": "@@CURRENT_TIMESTAMP"})       # one of the fixed SQL tokens
sms.update({"n": Literal("n + 1")}, where={"sys_id": 5})  # SQL, marked explicitly

At the client boundary

A browser can ask the data service to query and write tables, but it never sends SQL. The boundary checks every payload before anything runs:

• Filters must be structured data. SQL text in a filter, a sort or a grouping is refused, and so is any @@ token other than empty and null.
• Table names are checked against allow and deny lists, and row-scoped services add the owner's filter to every query and fail closed.
• Searches and groupings are capped (terms, fields, nesting depth, aggregates) so one request can't become an expensive one.
• Requests are signed in by default; a public action must be named as public.

Authorization is deny by default. Rules can permit or forbid; a forbid always wins, and a rule that fails can only take access away. Each user's effective permissions are compiled when their roles change, so checking them adds no query to a request.

Caching and performance

• Connection pool shared across warm invocations, with metrics.
• Bulk operations that send many rows in one statement instead of one at a time.
• Transaction cache. select(cache=True) remembers a result for the rest of the transaction.
• Read cache. cache_ttl= on a read keeps the result across requests. Entries are invalidated by writes to the table, published only when the writing transaction commits, and handed out as copies, so a cached row can't be changed by accident.
• N+1 detection. Repeating the same query in a loop is reported, with prefetch helpers to fix it.
• Slow-query log with a configurable threshold, plus prepared statements where they help.

Serverless on AWS

The velocity.aws package is the serverless half: the code that turns a Lambda function into an application back end.

from velocity.db.servers.postgres import initialize
from velocity.aws.handlers.lambda_handler import LambdaHandler

engine = initialize()

@engine.transaction
class HttpEventHandler(LambdaHandler):

    def OnActionGetUsers(self, tx, context):
        """POST /api  {"action": "get-users"}"""
        return tx.table("users").select(where={"active": True}).all()

def handler(event, context):
    return HttpEventHandler(event, context).serve()

• Actions. A request names an action, like get-users, and is routed to the matching OnActionGetUsers method, inside a transaction, with the signed-in user on the context. Errors map to clean responses: not allowed, bad request, not found.
• Queues. The SQS handler runs each message in its own transaction and reports only the failed ones back, so one bad message never fails the batch, and a message that keeps failing moves to a dead-letter queue.
• Enqueue on commit. context.enqueue(...) records the job with the caller's data and sends it only after commit, so a job never runs against data that was rolled back.
• Configuration. getenv(key) looks a setting up in AWS Systems Manager, first for the project and stage, then for every project in that stage, then in the environment. Results are cached for the life of the function.
• Capabilities. Optional features (scheduled jobs, retention, security headers, single sign-in sessions and more) are declared once, with their schema, defaults and schedules, and switched on per site.
• Operations. Job tracking, failure events sent to an alert topic, an operations dashboard, and deploy helpers for least-privilege access.

SQS batch: [record A, record B, record C]

record A → transaction A → commit
record B → transaction B → error → rollback
record C → transaction C → commit

returns {"batchItemFailures": [{"itemIdentifier": "msg-id-B"}]}

Beyond the database

• Payments. Stripe, Braintree and Authorize.Net behind one adapter. Money is integer cents, every charge carries an idempotency key, and the processor's record is written first, inside a savepoint, so a charge is never lost between the processor and the books. An outcome that can't be known is recorded as unknown and reconciled, never guessed.
• Per-user settings. One document per user for toolbar, grid layouts and preferences, changed by small patches under a row lock.
• Key-value and document stores. One tiered interface over Redis, DynamoDB, MongoDB, Elasticsearch and others, with a rule that no query silently scans a whole store.
• Utilities. Excel export, PDF rendering, email through SES or SMTP, S3 storage, and value conversion.

Configuration

Every setting velocity reads is listed in its configuration reference, and a test fails if the code reads one that isn't. The ones most projects touch:

Database support

PostgreSQL is the primary database, used in production and covered by the full test suite, through psycopg 3. SQLite works for tests and local development. Dialects for MySQL and SQL Server exist in the code, but they aren't tested to the same standard and aren't recommended for production.

An asynchronous API, engine.async_transaction(), is available as an opt-in experiment. Production code should use the synchronous API, which is what the library's own handlers use.

Testing and releases

• Tests. 3,620 unit tests with 80% line coverage as of release 0.2.5, plus integration tests that run against a real PostgreSQL database.
• Testing your code. velocity.testing.FakeTx is an in-memory transaction bound to the real method signatures. Anything it can't model raises an error instead of quietly pretending, so a test can't pass against behavior the real library doesn't have.
• Releases. Each release is cut after the unit suite, the formatter and a changelog check pass. The changelog records every release, with separate sections for breaking changes and for the order to deploy them in.
• Versions. The library is still 0.x: until 1.0, a release may change behavior callers rely on, and the changelog says so when it does. Pin the version you test with.
• Repository hygiene. A secret scanner runs before every commit.