sql-athame 0.4.0a15

Python tool for slicing and dicing SQL

sql-athame

Python tool for slicing and dicing SQL. Its intended target is Postgres with asyncpg, though it also includes support for rendering to a SQLAlchemy TextClause.

Note that current docstrings are autogenerated and have not been carefully reviewed; take them with a grain of salt!

Base query builder

Example

from sql_athame import sql


def get_orders(query):
    where = []

    if "id" in query:
        where.append(sql("id = {}", query["id"]))
    if "eventId" in query:
        where.append(sql("event_id = {}", query["eventId"]))
    if "startTime" in query:
        where.append(sql("start_time = {}", query["startTime"]))
    if "from" in query:
        where.append(sql("start_time >= {}", query["from"]))
    if "until" in query:
        where.append(sql("start_time < {}", query["until"]))

    return sql("SELECT * FROM orders WHERE {}", sql.all(where))


>>> list(get_orders({}))
['SELECT * FROM orders WHERE TRUE']

>>> list(get_orders({"id": "xyzzy"}))
['SELECT * FROM orders WHERE (id = $1)', 'xyzzy']

>>> list(get_orders({"eventId": "plugh", "from": "2019-05-01", "until": "2019-08-26"}))
['SELECT * FROM orders WHERE (event_id = $1) AND (start_time >= $2) AND (start_time < $3)',
 'plugh',
 '2019-05-01',
 '2019-08-26']


superquery = sql(
    """
    SELECT *
      FROM ({subquery}) sq
      JOIN other_table ot ON (ot.id = sq.id)
      WHERE ot.foo = {foo}
      LIMIT {limit}
    """,
    subquery=get_orders({"id": "xyzzy"}),
    foo="bork",
    limit=50,
)


>>> list(superquery)
['SELECT * FROM (SELECT * FROM orders WHERE (id = $1)) sq JOIN other_table ot ON (ot.id = sq.id) WHERE ot.foo = $2 LIMIT $3',
 'xyzzy',
 'bork',
 50]

API reference

from sql_athame import sql

sql(fmt: str, *args, **kwargs) -> Fragment

Creates a SQL Fragment from the fmt string. The fmt string contains literal SQL and may contain positional references, marked by {}, and named references, marked by {name}. Positional references must have a matching argument in *args. Named references may have a matching argument in **kwargs; if a named reference is not fullfilled by **kwargs it remains as a named slot to be filled later.

If a referenced argument is a Fragment, it is substituted into the SQL along with all of its embedded placeholders if any. Otherwise, it is treated as a placeholder value and substituted in place as a placeholder.

Fragment.query(self) -> Tuple[str, List[Any]]

Renders a SQL Fragment into a query string and list of placeholder parameters.

>>> q = sql("SELECT * FROM tbl WHERE qty > {qty}", qty=10)
>>> q.query()
('SELECT * FROM tbl WHERE qty > $1', [10])

If there are any unfilled slots ValueError will be raised.

>>> q = sql("SELECT * FROM tbl WHERE qty > {qty}")
>>> q.query()
ValueError: Unfilled slot: 'qty'
>>> q.fill(qty=10).query()
('SELECT * FROM tbl WHERE qty > $1', [10])

Fragment.__iter__(self) -> Iterator[Any]

A Fragment is an iterable which will return the query string followed by the placeholder parameters as returned by Fragment.query(self). This matches the (query, *args) argument pattern of the asyncpg API:

q = sql("SELECT * FROM tbl WHERE qty > {}", 10)
await conn.fetch(*q)

sql.list(parts: Iterable[Fragment]) -> Fragment

sql.list(*parts: Fragment) -> Fragment

Creates a SQL Fragment joining the fragments in parts together with commas.

>>> cols = [sql("a"), sql("b"), sql("c")]
>>> list(sql("SELECT {cols} FROM tbl", cols=sql.list(cols)))
['SELECT a, b, c FROM tbl']

sql.all(parts: Iterable[Fragment]) -> Fragment

sql.all(*parts: Fragment) -> Fragment

Creates a SQL Fragment joining the fragments in parts together with AND. If parts is empty, returns TRUE.

>>> where = [sql("a = {}", 42), sql("x <> {}", "foo")]
>>> list(sql("SELECT * FROM tbl WHERE {}", sql.all(where)))
['SELECT * FROM tbl WHERE (a = $1) AND (x <> $2)', 42, 'foo']
>>> list(sql("SELECT * FROM tbl WHERE {}", sql.all([])))
['SELECT * FROM tbl WHERE TRUE']

sql.any(parts: Iterable[Fragment]) -> Fragment

sql.any(*parts: Fragment) -> Fragment

Creates a SQL Fragment joining the fragments in parts together with OR. If parts is empty, returns FALSE.

>>> where = [sql("a = {}", 42), sql("x <> {}", "foo")]
>>> list(sql("SELECT * FROM tbl WHERE {}", sql.any(where)))
['SELECT * FROM tbl WHERE (a = $1) OR (x <> $2)', 42, 'foo']
>>> list(sql("SELECT * FROM tbl WHERE {}", sql.any([])))
['SELECT * FROM tbl WHERE FALSE']

Fragment.join(self, parts: Iterable[Fragment]) -> Fragment

Creates a SQL Fragment by joining the fragments in parts together with self.

>>> clauses = [sql("WHEN {} THEN {}", a, b) for a, b in ((sql("a"), 1), (sql("b"), 2))]
>>> case = sql("CASE {clauses} END", clauses=sql(" ").join(clauses))
>>> list(case)
['CASE WHEN a THEN $1 WHEN b THEN $2 END', 1, 2]

sql.literal(text: str) -> Fragment

Creates a SQL Fragment with the literal SQL text. No substitution of any kind is performed. Be very careful of SQL injection.

sql.identifier(name: str, prefix: Optional[str] = None) -> Fragment

Creates a SQL Fragment with a quoted identifier name, optionally with a dotted prefix.

>>> list(sql("SELECT {a} FROM tbl", a=sql.identifier("a", prefix="tbl")))
['SELECT "tbl"."a" FROM tbl']

sql.value(value: Any) -> Fragment

Creates a SQL Fragment with a single placeholder to value. Equivalent to:

sql("{}", value)

sql.escape(value: Any) -> Fragment

Creates a SQL Fragment with value escaped and embedded into the SQL. Types currently supported are strings, floats, ints, UUIDs, None, and sequences of the above.

>>> list(sql("SELECT * FROM tbl WHERE qty = ANY({})", sql.escape([1, 3, 5])))
['SELECT * FROM tbl WHERE qty = ANY(ARRAY[1, 3, 5])']

Compare to with a placeholder:

>>> list(sql("SELECT * FROM tbl WHERE qty = ANY({})", [1, 3, 5]))
['SELECT * FROM tbl WHERE qty = ANY($1)', [1, 3, 5]]

"Burning" an invariant value into the query can potentially help the query optimizer.

sql.slot(name: str) -> Fragment

Creates a SQL Fragment with a single empty slot named name. Equivalent to:

sql("{name}")

sql.unnest(data: Iterable[Sequence[Any]], types: Iterable[str]) -> Fragment

Creates a SQL Fragment containing an UNNEST expression with associated data.

The data is specified in tuples (in the "several database columns" sense, not necessarily the Python sense) in data, and the tuple Postgres types must be specified in types. The data is transposed into the correct form for UNNEST and embedded in placeholders in the Fragment.

>>> list(sql("SELECT * FROM {}", sql.unnest([("a", 1), ("b", 2), ("c", 3)], ["text", "integer"])))
['SELECT * FROM UNNEST($1::text[], $2::integer[])', ('a', 'b', 'c'), (1, 2, 3)]

Fragment.fill(self, **kwargs) -> Fragment

Creates a SQL Fragment by filling any empty slots in self with kwargs. Similar to sql subtitution, if a value is a Fragment it is substituted in-place, otherwise it is substituted as a placeholder.

Fragment.compile(self) -> Callable[..., Fragment]

Creates a function that when called with **kwargs will create a SQL Fragment equivalent to calling self.fill(**kwargs). This is optimized to do as much work as possible up front and can be considerably faster if repeated often.

Fragment.prepare(self) -> Tuple[str, Callable[..., List[Any]]]

Renders self into a SQL query string; returns that string and a function that when called with **kwargs containing the unfilled slots of self will return a list containing the placeholder values for self as filled with **kwargs.

>>> query, query_args = sql("UPDATE tbl SET foo={foo}, bar={bar} WHERE baz < {baz}", baz=10).prepare()
>>> query
'UPDATE tbl SET foo=$1, bar=$2 WHERE baz < $3'
>>> query_args(foo=1, bar=2)
[1, 2, 10]
>>> query_args(bar=42, foo=3)
[3, 42, 10]

As the name implies this is intended to be used in prepared statements:

query, query_args = sql("UPDATE tbl SET foo={foo}, bar={bar} WHERE baz < {baz}", baz=10).prepare()
stmt = await conn.prepare(query)
await stmt.execute(*query_args(foo=1, bar=2))
await stmt.execute(*query_args(bar=42, foo=3))

Fragment.sqlalchemy_text(self) -> sqlalchemy.sql.expression.TextClause

Renders self into a SQLAlchemy TextClause. Placeholder values will be bound with bindparams. Unfilled slots will be included as unbound parameters with their keys equal to the slot names. A placeholder value may be a SQLAlchemy BindParameter, in which case its value and type are used for the parameter created by this (and the key name is ignored); this allow control of the SQLAlchemy type used for the parameter.

Requires SQLAlchemy to be installed, otherwise raises ImportError.

>>> query = sql("SELECT * FROM tbl WHERE column = {}", 42)
>>> stmt = query.sqlalchemy_text()
>>> stmt._bindparams
{'_arg_0_140685932797232': BindParameter('_arg_0_140685932797232', 42, type_=Integer())}
>>> conn.execute(stmt).fetchall()

>>> query = sql("SELECT * FROM tbl WHERE column = {}", bindparam("ignored", 42, type_=Float()))
>>> stmt = query.sqlalchemy_text()
>>> stmt._bindparams
{'_arg_0_140294062155280': BindParameter('_arg_0_140294062155280', 42, type_=Float())}
>>> conn.execute(stmt).fetchall()

>>> query = sql("SELECT * FROM tbl WHERE column = {val}")
>>> stmt = query.sqlalchemy_text()
>>> stmt._bindparams
{'val': BindParameter('val', None, type_=NullType())}
>>> conn.execute(stmt, val=42).fetchall()

Dataclass helpers

Example

from dataclasses import dataclass, field
from datetime import date
from uuid import UUID, uuid4
from typing import Annotated, Optional

from sql_athame import ModelBase, ColumnInfo, sql


@dataclass
class Person(ModelBase, table_name="people", primary_key="id"):
    id: UUID
    name: str
    birthday: date
    title: Optional[str] = None
    extra: Optional[Annotated[dict, ColumnInfo(type="JSONB")]] = None


>>> list(Person.create_table_sql())
['CREATE TABLE IF NOT EXISTS "people" ('
 '"id" UUID NOT NULL, '
 '"name" TEXT NOT NULL, '
 '"birthday" DATE NOT NULL, '
 '"title" TEXT, '
 '"extra" JSONB, '
 'PRIMARY KEY ("id"))']

>>> list(Person.select_sql(where=sql("title = {}", "Director")))
['SELECT "id", "name", "birthday", "title", "extra" FROM "people" WHERE title = $1',
 'Director']

>>> people = [Person(uuid4(), "Bob", date(1974, 4, 3)), Person(uuid4(), "Anne", date(1985, 5, 4), extra={"awesome": "yes"})]
>>> people
[Person(id=UUID('8ecfe514-8fa3-48e5-8edb-fa810f5ed3f0'), name='Bob', birthday=datetime.date(1974, 4, 3), title=None, extra=None),
 Person(id=UUID('8d9bfadc-0026-4f6d-ae1b-ed89d1077f66'), name='Anne', birthday=datetime.date(1985, 5, 4), title=None, extra={'awesome': 'yes'})]

>>> list(Person.insert_multiple_sql(people))
['INSERT INTO "people" ("id", "name", "birthday", "title", "extra")'
 ' SELECT * FROM UNNEST($1::UUID[], $2::TEXT[], $3::DATE[], $4::TEXT[], $5::TEXT[]::JSONB[])',
 (UUID('8ecfe514-8fa3-48e5-8edb-fa810f5ed3f0'),
  UUID('8d9bfadc-0026-4f6d-ae1b-ed89d1077f66')),
 ('Bob', 'Anne'),
 (datetime.date(1974, 4, 3), datetime.date(1985, 5, 4)),
 (None, None),
 [None, '{"awesome": "yes"}']]

>>> list(Person.upsert_sql(Person.insert_multiple_sql(people)))
['INSERT INTO "people" ("id", "name", "birthday", "title", "extra")'
 ' SELECT * FROM UNNEST($1::UUID[], $2::TEXT[], $3::DATE[], $4::TEXT[], $5::TEXT[]::JSONB[])'
 ' ON CONFLICT ("id") DO UPDATE'
 ' SET "name"=EXCLUDED."name", "birthday"=EXCLUDED."birthday", "title"=EXCLUDED."title", "extra"=EXCLUDED."extra"',
 (UUID('8ecfe514-8fa3-48e5-8edb-fa810f5ed3f0'),
  UUID('8d9bfadc-0026-4f6d-ae1b-ed89d1077f66')),
 ('Bob', 'Anne'),
 (datetime.date(1974, 4, 3), datetime.date(1985, 5, 4)),
 (None, None),
 [None, '{"awesome": "yes"}']]

API reference

TODO, for now read the source.

License

MIT.

---

Copyright (c) 2019–2025 Brian Downing