🧩 recordsql

recordsql is a Python-based SQL query builder that offers a fluent and composable interface for generating complex SQL statements — safely and efficiently.

📦 Installation

Install the required dependency and the pip package

pip install expressql
pip install recordsql

🚀 Features

• Fluent, chainable API for building SQL queries
• Support for SELECT, INSERT, UPDATE, DELETE, WITH, JOIN, COUNT, and EXISTS
• Parameterized placeholders for SQL injection safety
• Full support for multiple JOIN types (INNER, LEFT, RIGHT, FULL)
• Modular, composable components for query reuse

🛠️ Usage

Here are some examples of how to use recordsql to build SQL queries.

1. SELECT Query

from recordsql import SELECT, cols, text, col

# Define columns
name, age, email, total_purchases, signup_date, infractions = cols(
    "name", "age", "email", "total_purchases", "signup_date", "infractions"
)

# Build a SELECT query
select_query = SELECT(name, age, email, total_purchases).FROM("customers").WHERE(
    ((signup_date - col("CURRENT_TIMESTAMP")) > (text("1 year").DATETIME())) &
    (total_purchases > 1000) &
    (infractions == 0)
).ORDER_BY(total_purchases, "DESC", (signup_date - col("CURRENT_TIMESTAMP"), "ASC")).LIMIT(10).OFFSET(1)

print(*select_query.placeholder_pair(), sep="\n")

2. WITH Query and JOINs

from recordsql import WITH, JoinQuery, cols, num, col

# Define columns and values
name, age, email, total_purchases = cols("name", "age", "email", "total_purchases")
current_store_id = num(1275682)

# Build a WITH query
with_query = WITH(select_query.AS("customer_data")).SELECT(
    name, age, email, total_purchases
).FROM("customer_data").WHERE(
    (total_purchases > 1000) & (infractions == 0)
).ORDER_BY(total_purchases, "DESC").LIMIT(10).OFFSET(1)

# Add JOINs
with_query.INNER_JOIN(
    table_name="prices",
    on=(current_store_id == col("store_id"))
).LEFT_JOIN(
    table_name="orders",
    on=(current_store_id == col("store_id"))
)

print(*with_query.placeholder_pair(), sep="\n")

Click to reveal output

WITH customer_data AS (SELECT name, age, email, total_purchases FROM "customers" WHERE ((signup_date-CURRENT_TIMESTAMP) > DATETIME(?)) AND (total_purchases > ?) AND (infractions = ?) ORDER BY total_purchases DESC, (signup_date-CURRENT_TIMESTAMP) ASC LIMIT 10 OFFSET 1) SELECT name, age, email, total_purchases FROM "customer_data" INNER JOIN "prices" ON ? = store_id LEFT JOIN "orders" ON ? = store_id WHERE (total_purchases > ?) AND (infractions = ?) ORDER BY total_purchases DESC LIMIT 10 OFFSET 1

['1 year', 1000, 0, 1000, 0, 1275682, 1275682]

3. UPDATE Query

from recordsql import UPDATE, col

# Build an UPDATE query
update_query = UPDATE("customers").SET(
    name="John Doe",
    age=30,
    email="johndoe@gmail.com"
).WHERE(col("customer_id") == 12345).RETURNING("name", "age", "email")

print(*update_query.placeholder_pair(), sep="\n")

4. INSERT Query

from recordsql import INSERT

# Build an INSERT query
insert_query = INSERT("col1", "col2").INTO("table_name").VALUES(
    (1, 2),
    (3, 4),
    (5, 6)
).ON_CONFLICT(
    do="UPDATE",
    conflict_cols=["col1"],
    set={"col2": 10},
    where=col("col1") == 1
).RETURNING("col1", "col2")

print(*insert_query.placeholder_pair(), sep="\n")

5. COUNT and EXISTS Queries

from recordsql import COUNT, EXISTS, col, text

# Build a COUNT query
count_query = COUNT().FROM("customers").WHERE(
    ((signup_date - col("CURRENT_TIMESTAMP")) > (text("1 year").DATETIME())) &
    (total_purchases > 1000) &
    (infractions == 0)
).GROUP_BY(name).HAVING(total_purchases > 1000)

print(*count_query.placeholder_pair(), sep="\n")

# Build an EXISTS query
exists_query = EXISTS().FROM("customers").WHERE(
    ((signup_date - col("CURRENT_TIMESTAMP")) > (text("1 year").DATETIME())) &
    (total_purchases > 1000) &
    (infractions == 0)
)

print(*exists_query.placeholder_pair(), sep="\n")

6. DELETE Query

from recordsql import DELETE, col, text

# Build a DELETE query
delete_query = DELETE().FROM("customers").WHERE(
    ((signup_date - col("CURRENT_TIMESTAMP")) > (text("1 year").DATETIME())) &
    (total_purchases > 1000) &
    (infractions == 0)
)

print(*delete_query.placeholder_pair(), sep="\n")

📝 Output

The queries generated by recordsql are parameterized and safe for execution. Here’s an example output:

WITH customer_data AS (
    SELECT * FROM "customers"
    WHERE ((signup_date - CURRENT_TIMESTAMP) > DATETIME(?))
    AND (total_purchases > ?)
    AND (infractions = ?)
    ORDER BY total_purchases DESC, (signup_date - CURRENT_TIMESTAMP) ASC
    LIMIT 10 OFFSET 1
)
SELECT * FROM "customer_data"
WHERE (total_purchases > ?)
AND (infractions = ?)
ORDER BY total_purchases DESC, (signup_date - CURRENT_TIMESTAMP) ASC
LIMIT 10 OFFSET 1

Placeholders:

['1 year', 1000, 0, 1000, 0]

📖 Documentation

Full documentation is available and includes:

• Installation Guide: Detailed installation instructions
• Quick Start: Get started with recordsql in minutes
• Advanced Examples: Complex query patterns and best practices
• API Reference: Complete API documentation with type hints

Building Documentation Locally

To build the documentation locally:

cd docs
poetry run sphinx-build -b html . _build/html

Or using make:

cd docs
make html  # On Unix/Linux/macOS
make.bat html  # On Windows

The generated documentation will be in docs/_build/html/. Open index.html in your browser to view it.

For more details on contributing to documentation, see docs/README.md.