querky

Turn your SQL queries into type annotated Python functions and autogenerated types with a single decorator.

Showcase

This example shows what querky SQL functions look like.

Consider this PostgreSQL database schema:

CREATE TABLE account (
    id BIGSERIAL PRIMARY KEY,
    username TEXT UNIQUE NOT NULL,
    first_name TEXT NOT NULL,
    last_name TEXT,
    phone_number TEXT,
    balance BIGINT NOT NULL DEFAULT 0,
    join_ts TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    referred_by_account_id BIGINT REFERENCES account (id)
);

CREATE TABLE post (
    id BIGSERIAL PRIMARY KEY,
    poster_id BIGINT NOT NULL REFERENCES account (id),
    message TEXT NOT NULL,
    ts TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE post_comment (
    id BIGSERIAL PRIMARY KEY,
    post_id BIGINT NOT NULL REFERENCES post (id),
    commenter_id BIGINT NOT NULL REFERENCES account (id),
    message TEXT NOT NULL,
    ts TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

And these are the queries defined on it:

from querky_def import qrk


# an UPDATE query: no value returned
@qrk.query  # or @qrk.query(shape='status')
def update_account_phone_number(account_id, new_phone_number):
    return f'''
        UPDATE
            account
        SET
            phone_number = {+new_phone_number}
        WHERE
            id = {+account_id}
        '''


# an INSERT query to always return a single value
@qrk.query(shape='value', optional=False)
def insert_account(username, first_name, last_name, phone_number, balance, referred_by_account_id):
    return f'''
        INSERT INTO
            account
            (
                username,
                first_name,
                last_name,
                phone_number,
                balance,
                referred_by_account_id
            )
        VALUES
            (
                {+username},
                {+first_name},
                {+last_name},
                {+phone_number},
                {+balance},
                {+referred_by_account_id}
            )
        RETURNING
            id
        '''


# a SELECT query to return an array of single values
@qrk.query(shape='column')
def select_top_largest_balances(limit):
    return f'''
        SELECT
            balance
        FROM
            account
        ORDER BY
            balance DESC
        LIMIT
            {+limit}
        '''


# now for the most interesting part: fetching rows
# a SELECT query to return a single (one) AccountReferrer row or None (optional)
@qrk.query('AccountReferrer', shape='one', optional=True)
def get_account_referrer(account_id):
    return f'''
        SELECT
            referrer.id,
            referrer.username,
            referrer.first_name,
            referrer.last_name,
            referrer.join_ts

        FROM 
            account

        INNER JOIN
            account AS referrer
        ON
            account.referred_by_account_id = referrer.id

        WHERE
            account.id = {+account_id}
        '''


# a SELECT query to return many (an array of) AccountPostComment rows
@qrk.query('AccountPostComment', shape='many')
def select_last_post_comments(post_id, limit):
    return f'''
        SELECT 
            account.first_name,
            account.last_name,
            post_comment.id,
            post_comment.message

        FROM
            post_comment

        INNER JOIN
            account
        ON
            post_comment.commenter_id = account.id

        WHERE
            post_comment.post_id = {+post_id}

        ORDER BY
            post_comment.ts DESC

        LIMIT
            {+limit}
        '''

So, as you can see, all you need is 3 simple steps:

Write a Python function returning the desired SQL query.
Insert the arguments exactly where you want them to be. Don't forget to prepend your arguments with a plus sign (+). Even though it is a regular Python format string, the resulting query is not SQL-injectable, as you'll later see.
Add the @qrk.query decorator using arguments to describe the expected shape and type of result set.

Before you can use this code, you'll need the qrk object.

Bear with me, I'll show the full configuration in the next section, but, firstly, I would like to show the results of running querky's code generator. Here it is:

# ~ AUTOGENERATED BY QUERKY ~ #
import datetime
from asyncpg import Connection
from dataclasses import dataclass
from sql.example import update_account_phone_number as _q0
from sql.example import insert_account as _q1
from sql.example import select_top_largest_balances as _q2
from sql.example import get_account_referrer as _q3
from sql.example import select_last_post_comments as _q4


async def update_account_phone_number(__conn: Connection, /, account_id: int, new_phone_number: str) -> str:
    return await _q0.execute(__conn, account_id, new_phone_number)


async def insert_account(__conn: Connection, /, username: str, first_name: str, last_name: str, phone_number: str, balance: int, referred_by_account_id: int) -> int:
    return await _q1.execute(__conn, username, first_name, last_name, phone_number, balance, referred_by_account_id)


async def select_top_largest_balances(__conn: Connection, /, limit: int) -> list[int]:
    return await _q2.execute(__conn, limit)


@dataclass(slots=True)
class AccountReferrer:
    id: int
    username: str
    first_name: str
    last_name: str
    join_ts: datetime.datetime


async def get_account_referrer(__conn: Connection, /, account_id: int) -> AccountReferrer | None:
    return await _q3.execute(__conn, account_id)

_q3.bind_type(AccountReferrer)


@dataclass(slots=True)
class AccountPostComment:
    first_name: str
    last_name: str
    id: int
    message: str


async def select_last_post_comments(__conn: Connection, /, post_id: int, limit: int) -> list[AccountPostComment]:
    return await _q4.execute(__conn, post_id, limit)

_q4.bind_type(AccountPostComment)


__all__ = [
    "select_last_post_comments",
    "AccountPostComment",
    "AccountReferrer",
    "insert_account",
    "update_account_phone_number",
    "get_account_referrer",
    "select_top_largest_balances",
]

So, let's analyze what we got:

We have all of our input and output types defined. The linter can now help us whenever we use any of these functions and types in our code.
Whenever the database schema changes, the types and function arguments will accommodate automatically: just run the generation script again - and you're set.
All the types were inferred from a live database connection, because your database is the single source of truth for your data, not the application.
Our "models" are database rows. At last.

Do not be discouraged, if you don't like using dataclasses in your projects, as this is just an example!

So, if you like what you're seeing, let's configure your project!

Basic Configuration

asyncpg

To install, run

pip install querky[asyncpg]

Consider this project structure:

src
|__ querky_def.py
|__ querky_gen.py
|__ sql
    |__ example.py

sql folder contains .py files with the query functions. Generated code will be placed in the sql/queries folder under the same name as the inital script (example.py in this case).

querky_gen.py file is the code generation script. You run it when you want to regenerate the query functions:

import asyncio

from querky.presets.asyncpg import generate

from querky_def import qrk
import sql
from env import CONNECTION_STRING


if __name__ == "__main__":
    asyncio.run(generate(qrk, CONNECTION_STRING, base_modules=(sql, )))

querky_def.py is code generator configuration. We'll use a preset for the sake of simplicity.

import os
from querky.presets.asyncpg import use_preset


qrk = use_preset(os.path.dirname(__file__), type_factory='dataclass+slots')

The first argument should be the path to the root directory of your project.

If you'd like more fine-grained control over the Querky object, there will be an explaination in the later sections.

After the configuration of the qrk object it's time to run the querky_gen.py script.

Each of your queries will become type hinted, each of them will return a real Python object, and you can call these queries as regular Python functions.

Every time you change your database schema or queries, you can now expect the changes to propagate throughout your code. Because of that, refactoring SQL-dependent code has never been easier. This time the linter is on your side.

Do not change the generated files, as they are transient and will be overwritten. If you need to modify the generated code, consider using on_before_func_code_emit and on_before_type_code_emit hooks passed in to the Querky object constructor.

Type Hinting Extensions

Arguments

Optionals

A careful reader might have noticed, that the generated argument types are never optional. If we go back to the database schema, we will notice, that some of the columns are NULLABLE. And yet, in insert_account query the respective arguments are not Optional.

Why is that?

Unfortunately, there is no straightforward way for the library to automate this process, because SQL-wise these are constraints, and not data types.

So, it's our job to hint the library to do the right thing.

Let's look at the signature of this function again:

@qrk.query(shape='value', optional=False)
def insert_account(
        username, 
        first_name, 
        last_name, 
        phone_number, 
        balance, 
        referred_by_account_id
):
    ...

We know that phone_number, last_name and referred_by_account_id are optional. So we just hint them like this:

import typing

@qrk.query(shape='value', optional=False)
def insert_account(

Querky

Install / Use

README