Drivers

The sql_driver option picks which database library the generated code targets. It must match your engine. Nine drivers are supported:

Every generated query function takes the connection as its first argument, so you open and manage the connection yourself and pass it in.

All PostgreSQL drivers produce the same models and type contract. Pick asyncpg for raw throughput, or a psycopg driver to stay in the psycopg ecosystem (libpq, pipeline mode, PgBouncer) - psycopg_async for asyncio, psycopg_sync for synchronous code.

asyncpg (PostgreSQL)

import asyncio

import asyncpg

from app.db import queries


async def main() -> None:
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    user = await queries.get_field_naming(conn, id_=1)


asyncio.run(main())

asyncpg supports :copyfrom (bulk insert via copy_records_to_table).

psycopg_async (PostgreSQL)

import asyncio

import psycopg

from app.db import queries


async def main() -> None:
    conn = await psycopg.AsyncConnection.connect("postgresql://user:pass@localhost/db")
    user = await queries.get_field_naming(conn, id_=1)


asyncio.run(main())

The generated code targets Psycopg 3 (3.2 or newer) with its default tuple rows - the connection annotation is psycopg.AsyncConnection[psycopg.rows.TupleRow], so a connection configured with another row factory is rejected by pyright. :copyfrom streams rows through cursor.copy().

psycopg_sync (PostgreSQL)

import psycopg

from app.db import queries

with psycopg.connect("postgresql://user:pass@localhost/db") as conn:
    user = queries.get_field_naming(conn, id_=1)

Same contract as psycopg_async, emitted as plain functions without async/await. The connection annotation is psycopg.Connection[psycopg.rows.TupleRow]. :many returns the same QueryResults helper: call it (queries.list_x(conn)()) to fetch every row, or iterate it with a plain for loop. The json/jsonb loader registration is identical; the Windows event-loop caveat does not apply.

aiosqlite (async SQLite)

import asyncio
import sqlite3

import aiosqlite

from app.db import queries


async def main() -> None:
    async with aiosqlite.connect("app.db", detect_types=sqlite3.PARSE_DECLTYPES) as conn:
        user = await queries.get_field_naming(conn, id_=1)


asyncio.run(main())

sqlite3 (sync SQLite)

import sqlite3

from app.db import queries

conn = sqlite3.connect("app.db", detect_types=sqlite3.PARSE_DECLTYPES)
user = queries.get_field_naming(conn, id_=1)

asyncmy (async MySQL)

import asyncio

import asyncmy

from app.db import queries


async def main() -> None:
    conn = await asyncmy.connect(host="localhost", user="user", password="pass", database="db")
    user = await queries.get_field_naming(conn, id_=1)


asyncio.run(main())

The generated code targets asyncmy with its default tuple cursors; the connection annotation is asyncmy.Connection. No connection flags are needed - values convert inline, so date/datetime/decimal columns round-trip out of the box.

pymysql (sync MySQL)

import pymysql

from app.db import queries

conn = pymysql.connect(host="localhost", user="user", password="pass", database="db")
user = queries.get_field_naming(conn, id_=1)

Same contract as asyncmy, emitted as plain functions without async/await. The connection annotation is pymysql.Connection.

Behavior shared by both MySQL drivers:

• Queries are written with ? placeholders like on SQLite; the generated SQL constants hold the pyformat %s form the drivers expect, with literal % doubled. The rewrite happens at generation time - your .sql files stay plain MySQL.
• time columns map to datetime.timedelta and tinyint(1) to bool, matching what the drivers return. The full table is in the type mappings reference.
• memoryview parameters bind as bytes automatically - the PyMySQL family cannot encode a raw memoryview.
• Inline ENUM (and SET) columns generate enum classes.
• :execlastid returns cursor.lastrowid; :copyfrom is not supported.

turso_sync / turso_async (Turso)

Turso is an SQLite-compatible database engine; its pyturso package (uv add pyturso) mirrors the sqlite3 module’s API and adds a native asyncio variant.

• turso_sync targets the turso module, turso_async the turso.aio module.
• Both use engine: "sqlite" and the same ? placeholders and queries as the SQLite drivers.
• Unlike the sqlite3 module, no connection flags are needed - the generated code converts values inline, and the observable Python types match the SQLite drivers exactly.

import turso

from app.db import queries

conn = turso.connect("app.db")
user = queries.get_field_naming(conn, id_=1)

import asyncio

import turso.aio

from app.db import queries


async def main() -> None:
    conn = await turso.aio.connect("app.db")
    user = await queries.get_field_naming(conn, id_=1)


asyncio.run(main())

Command support

Not every query command works on every driver - for example :copyfrom is PostgreSQL-only and :execlastid is limited to the SQLite-engine and MySQL drivers. The full matrix is in the feature support reference.