Quick Start

Query Builder Initialization

$ pip install david8

♻️ Most examples in the documentation will use this configuration to avoid unnecessary qb initialization / imports / prints etc

Default query builder initialization:

from logging.config import dictConfig
from david8 import get_default_qb


# enable logging
dictConfig({
    'version': 1,
    'formatters': {'standard': {'format': '[%(levelname)s]: %(message)s'}},
    'handlers': {
        'default': {
            'level': 'INFO',
            'formatter': 'standard',
            'class': 'logging.StreamHandler',
            'stream': 'ext://sys.stdout',
        },
    },
    'loggers': {
        '': {'handlers': ['default'], 'level': 'INFO'},
        'david8': {'propagate': True},
    }
})

qb = get_default_qb()

Placeholders / Parameter Styles and Quote mode

By default david8 will use PyFormatParamStyle with disabled quote_mode. Use param_styles.py to change query parameter / placeholder style or enable quote_mode:

from david8 import get_default_qb
from david8.expressions import param
from david8.param_styles import *

for _qb in [
    get_default_qb(),
    get_default_qb(is_quote_mode=True),
    get_default_qb(QMarkParamStyle()),
    get_default_qb(NumericParamStyle()),
    get_default_qb(NamedParamStyle()),
    get_default_qb(FormatParamStyle()),
]:
    query = (
        _qb.select(
            param('one').as_('first_param'),
            param(2).as_('second_param'),
            'column_name'
        )
        .from_table('table_name')
    )

    # to see logging info and generate query parameters
    query.get_sql()
    # to get Query parameters
    # print(query.get_parameters()) will return dict from log
    print(query.get_tuple_parameters())
    print(query.get_list_parameters())

# [INFO]: SELECT %(p1)s AS first_param, %(p2)s AS second_param, column_name FROM table_name
# {'p1': 'one', 'p2': 2}
# ('one', 2)
# ['one', 2]
# [INFO]: SELECT %(p1)s AS "first_param", %(p2)s AS "second_param", "column_name" FROM "table_name"
# {'p1': 'one', 'p2': 2}
# ('one', 2)
# ['one', 2]
# [INFO]: SELECT ? AS first_param, ? AS second_param, column_name FROM table_name
# {'1': 'one', '2': 2}
# ('one', 2)
# ['one', 2]
# [INFO]: SELECT $1 AS first_param, $2 AS second_param, column_name FROM table_name
# {'1': 'one', '2': 2}
# ('one', 2)
# ['one', 2]
# [INFO]: SELECT :p1 AS first_param, :p2 AS second_param, column_name FROM table_name
# {'p1': 'one', 'p2': 2}
# ('one', 2)
# ['one', 2]
# [INFO]: SELECT %s AS first_param, %s AS second_param, column_name FROM table_name
# {'1': 'one', '2': 2}
# ('one', 2)
# ['one', 2]

See examples and how to use

• SELECT
• DML
• DDL
• test_cast and other tests

Below are examples using specific databases and drivers

ClickHouse

$ docker run -d --name david8_clickhouse -p 9000:9000 -e CLICKHOUSE_USER=david8 -e CLICKHOUSE_PASSWORD=david8 -d clickhouse/clickhouse-server:25.9.5.21
$ pip install clickhouse-driver==0.2.10

💡 See david8_clickhouse

from david8 import get_default_qb
from david8.functions import count
from david8.predicates import ne
from clickhouse_driver import Client

client = Client('localhost', user='david8', password='david8',)
query = (
    get_default_qb()
    .select(
        'database',
        count('').as_('tables'),
    )
    .from_table('tables', db_name='system')
    .where(ne('database', 'INFORMATION_SCHEMA'))
    .group_by('database')
)

print(client.execute(query.get_sql(), query.get_parameters()))
# [INFO]: SELECT database, count() AS tables FROM system.tables
# WHERE database != %(p1)s GROUP BY database
# {'p1': 'INFORMATION_SCHEMA'}
# [('system', 113), ('information_schema', 20)]

PostgreSQL

docker run --name david8_postgres -e POSTGRES_PASSWORD=david8 -e POSTGRES_DB=david8 -e POSTGRES_USER=david8 -p 5432:5432 -d postgres:16

💡 See david8_postgresql

psycopg2 example(pip install psycopg2==2.9.11):

import psycopg2
from david8 import get_default_qb
from david8.functions import count
from david8.predicates import ne

connection = psycopg2.connect(user='david8', password='david8', host='127.0.0.1', port='5432', database='david8')
query = (
    get_default_qb().
    select(
        'relkind',
        count('*').as_('obj_counter'),
    )
    .from_table('pg_class', db_name='pg_catalog')
    .where(ne('relkind', 't'))
    .group_by('relkind')
    .order_by('relkind')
)

cursor = connection.cursor()
cursor.execute(query.get_sql(), query.get_parameters())
print(cursor.fetchall())

psycopg3 example(pip install psycopg==3.2.12):

import psycopg
from david8 import get_default_qb
from david8.param_styles import FormatParamStyle
from david8.functions import count
from david8.predicates import ne
conn = psycopg.connect('dbname=david8 user=david8 password=david8 host=127.0.0.1 port=5432')
conn.autocommit = True

query = (
    get_default_qb(FormatParamStyle())
    .select(
        'relkind',
        count('*').as_('obj_counter'),
    )
    .from_table('pg_class', db_name='pg_catalog')
    .where(ne('relkind', 't'))
    .group_by('relkind')
    .order_by('relkind')
)

with conn.cursor() as cursor:
    cursor.execute(query.get_sql(), query.get_tuple_parameters())
    print(cursor.fetchall())
# SELECT relkind, count(*) AS obj_counter FROM pg_catalog.pg_class
# WHERE relkind != %s GROUP BY relkind ORDER BY relkind
# [('i', 164), ('r', 68), ('v', 141)]

asyncpg example(pip install asyncio==4.0.0 asyncpg==0.31.0):

from david8.param_styles import NumericParamStyle
from david8 import get_default_qb
from david8.predicates import ne
from david8.functions import count
import asyncio
import asyncpg

async def run():
    conn = await asyncpg.connect(user='david8', password='david8', database='david8', host='127.0.0.1')
    query = (
        get_default_qb(NumericParamStyle())
        .select(
            'relkind',
            count('*').as_('obj_counter'),
        )
        .from_table('pg_class', db_name='pg_catalog')
        .where(ne('relkind', "t".encode("utf-8")))
        .group_by('relkind')
        .order_by('relkind')
    )

    values = await conn.fetchrow(
        query.get_sql(),
        *query.get_tuple_parameters(),
    )
    print(values)

asyncio.run(run())
# [INFO]: SELECT relkind, count(*) AS obj_counter FROM pg_catalog.pg_class WHERE relkind != $1 GROUP BY relkind ORDER BY relkind
# {'1': b't'}
# <Record relkind=b'i' obj_counter=164>

DuckDB

An example of how to merge PostgreSQL data and local csv

$ pip install duckdb==1.4.2
# create a csv
$ echo "relkind,name" > names.csv && echo "r,ordinary table" >> names.csv && echo "i,index" >> names.csv && echo "v,view" >> names.csv

import duckdb
from david8.joins import left
from david8 import get_default_qb
from david8.predicates import ne
from david8.functions import count
from david8.param_styles import QMarkParamStyle

con = duckdb.connect()
con.execute('INSTALL postgres')
con.execute('LOAD postgres')
con.execute("""
    ATTACH 'host=127.0.0.1 port=5432 user=david8 password=david8 dbname=david8'
    AS pg
    (TYPE postgres);
""")

con.execute("CREATE TABLE names AS SELECT * FROM read_csv('names.csv')")
qb = get_default_qb(QMarkParamStyle())
query = (
    qb.select(
        'relkind',
        'name',
        count('*').as_('obj_counter'),
    )
    .from_table('pg_class', db_name='pg_catalog')
    .join(left().table('names').using('relkind'))
    .where(ne('relkind', 't'))
    .group_by('relkind', 'name')
    .order_by('relkind')
)

print(con.execute(query.get_sql(), query.get_tuple_parameters()).fetchall())
# [INFO]: SELECT relkind, name, count(*) AS obj_counter FROM pg_catalog.pg_class
# LEFT JOIN names USING (relkind) WHERE relkind != ? GROUP BY relkind, name ORDER BY relkind
# {'1': 't'}
# [('i', 'index', 124), ('r', 'ordinary table', 210), ('v', 'view', 46)]

MySQL

$ docker run --name david8_msql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=david8 -e MYSQL_DATABASE=david8 -e MYSQL_USER=david8 -e MYSQL_PASSWORD=david8 -d mysql:8.0
$ pip install mysql-connector-python==9.5.0

import mysql.connector
from david8 import get_default_qb
from david8.functions import count
from david8.predicates import ne

cnx = mysql.connector.connect(host='127.0.0.1', port=3306, user='david8', password='david8', database='david8')
query = (
    get_default_qb()
    .select(
        'data_type',
        count('*').as_('data_types_without_json'),
    )
    .from_table('COLUMNS', db_name='information_schema')
    .where(ne('data_type', 'json'))
    .group_by('data_type')
    .order_by_desc('data_types_without_json')
)

cursor = cnx.cursor()
cursor.execute(query.get_sql(), query.get_parameters())
print(cursor.fetchall())
# [INFO]: SELECT data_type, count(*) AS data_types_without_json FROM information_schema.COLUMNS
# WHERE data_type != %(p1)s GROUP BY data_type ORDER BY data_types_without_json DESC
# {'p1': 'json'}
# [('varchar', 339), ('bigint', 150), ('int', 104), ('binary', 26), ('enum', 19), ('char', 18), ('longtext', 13), ('datetime', 13), ('float', 9), ('text', 8), ('timestamp', 7), ('set', 6), ('varbinary', 4), ('decimal', 3), ('mediumtext', 2), ('tinyint', 2), ('blob', 1), ('double', 1)]

SQLite

import sqlite3
from david8 import get_default_qb
from david8.param_styles import QMarkParamStyle
from david8.predicates import ne
from david8.functions import count


cursor = sqlite3.connect('david8.db').cursor()
query = (
    get_default_qb(QMarkParamStyle())
    .select(
        'type',
        count('type').as_('types'),
    )
    .from_table('sqlite_master')
    .where(ne('type', 'view'))
    .group_by('type')
)

cursor.execute('DROP TABLE IF EXISTS david8')
cursor.execute('CREATE TABLE david8 (id INTEGER)')
cursor.execute(query.get_sql(), query.get_tuple_parameters())

print(cursor.fetchall())
# [INFO]: SELECT type, count(type) AS types FROM sqlite_master WHERE type != ? GROUP BY type {'1': 'view'}
# [('table', 1)]