🗃️ SQL Helpers

Comprehensive SQL helper library for database operations with multi-database support.

🎯 Overview

This module provides a batteries-included SQL toolkit for building database-heavy applications:

Category	Functions	Purpose
🔍 Query Registry	`run_id`, `validate_params`	Execute queries by ID from centralized registries
➕ Insert-Only	`insert_only`, `bulk_insert_only`	Insert new records, skip conflicts
🔄 Upsert	`upsert`, `bulk_upsert`	Insert or update existing records
📝 CRUD	`get_by_id`, `update_record`, `delete_record`, `bulk_delete`	Standard database operations
🔧 Utilities	`with_transaction`, `paginate_sql`, `batch_execute`	Transaction management & helpers
💰 Money	`to_cents`, `from_cents`	Currency conversion for `int`-only storage

🏗️ Architecture

┌─────────────────────────────────────────────────────────────────┐
│                     SQL Utilities Layer                         │
├─────────────────────────────────────────────────────────────────┤
│  🔍 Query Registry        │  Execute queries by ID              │
│  ➕ Insert Operations     │  insert_only, bulk_insert_only      │
│  🔄 Upsert Operations     │  upsert, bulk_upsert                │
│  📝 CRUD Operations       │  get, update, delete (single/bulk)  │
│  🔧 Utilities             │  transactions, pagination, batching │
│  💰 Money Helpers         │  cents ↔ dollars conversion         │
├─────────────────────────────────────────────────────────────────┤
│  DB_TYPE Detection: PostgreSQL / SQLite (auto-switch syntax)    │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│              fastsql Database Connection                        │
└─────────────────────────────────────────────────────────────────┘

📚 Quick Reference

Conflict Handling

Operation	On Conflict	Use Case
`insert_only`	Skip (DO NOTHING)	Append-only logs, idempotent imports
`upsert`	Update existing	Sync external data, refresh caches

Performance Tips

Scenario	Recommended	Why
Single record	`insert_only`, `upsert`	Simple, immediate
10+ records	`bulk_*` variants	10-100x faster
100+ records	`batch_execute`	Commits per batch, memory-safe
Multi-table changes	`with_transaction`	All-or-nothing commits

🔍 Database Type Detection

Automatically detects PostgreSQL vs SQLite from DB_TYPE environment variable to generate appropriate SQL syntax.

source

get_db_type


def get_db_type(
    
):

Get database type from environment variable

🎯 Query Registry

Execute queries by ID from a centralized registry with automatic parameter validation.

Function	Purpose
`run_id`	Execute a registered query by its ID
`validate_params`	Ensure all `:param` placeholders have values

source

run_id


def run_id(
    db:Database, registry:Dict, query_id:str, params:Optional=None
)->Any:

Execute a query by ID from a query registry.

source

validate_params


def validate_params(
    sql:str, params:Dict
)->None:

Validate that all required parameters are provided

➕ Insert-Only

Insert new records only if they don’t exist — conflicts are silently skipped.

Function	Records	SQL Generated
`insert_only`	Single	`ON CONFLICT DO NOTHING` (PG) / `INSERT OR IGNORE` (SQLite)
`bulk_insert_only`	Multiple	Same, with batch execution

💡 Use case: Idempotent imports, append-only audit logs, webhook deduplication

source

bulk_insert_only


def bulk_insert_only(
    db:Database, table_name:str, records:List, conflict_cols:List, auto_commit:bool=True
)->None:

Insert multiple records, skipping conflicts (optimized batch operation).

source

insert_only


def insert_only(
    db:Database, table_name:str, record:Dict, conflict_cols:List, auto_commit:bool=True
)->None:

Insert a single record only if it doesn’t exist (ignores conflicts).

🔄 Upsert

Insert new records or update existing ones on conflict.

Function	Records	SQL Generated
`upsert`	Single	`ON CONFLICT DO UPDATE` (PG) / `INSERT OR REPLACE` (SQLite)
`bulk_upsert`	Multiple	Same, with batch execution

💡 Use case: Syncing external API data, refreshing caches, user settings

source

bulk_upsert


def bulk_upsert(
    db:Database, table_name:str, records:List, conflict_cols:List, update_cols:Optional=None, auto_commit:bool=True
)->None:

Insert or update multiple records (optimized batch operation).

source

upsert


def upsert(
    db:Database, table_name:str, record:Dict, conflict_cols:List, update_cols:Optional=None, auto_commit:bool=True
)->None:

Insert a record or update if it exists (upsert).

📝 CRUD

Standard Create, Read, Update, Delete operations.

Function	Operation	Description
`get_by_id`	Read	Fetch single record by primary key
`update_record`	Update	Modify record fields by ID
`delete_record`	Delete	Remove single record by ID
`bulk_delete`	Delete	Remove multiple records by ID list

source

bulk_delete


def bulk_delete(
    db:Database, table_name:str, id_list:List, id_col:str='id', auto_commit:bool=True
)->None:

Delete multiple records by ID list.

source

delete_record


def delete_record(
    db:Database, table_name:str, id_value:Any, id_col:str='id', auto_commit:bool=True
)->None:

Delete a single record by ID.

source

update_record


def update_record(
    db:Database, table_name:str, id_value:Any, id_col:str='id', auto_commit:bool=True, updates:VAR_KEYWORD
)->None:

Update a single record by ID.

source

get_by_id


def get_by_id(
    db:Database, table_name:str, id_value:Any, id_col:str='id'
)->Any:

Get a single record by ID.

🔧 Utilities

Transaction management, pagination, and batch processing.

Function	Purpose
`with_transaction`	Context manager for atomic operations
`paginate_sql`	Add LIMIT/OFFSET to queries
`batch_execute`	Process large lists in memory-safe chunks

source

batch_execute


def batch_execute(
    db:Database, operation_func, items:List, batch_size:int=100
)->None:

Execute an operation on items in batches with commits after each batch.

source

paginate_sql


def paginate_sql(
    sql:str, page:int, page_size:int
)->str:

Add LIMIT/OFFSET pagination to a SQL query.

source

with_transaction


def with_transaction(
    db:Database
):

Context manager for safe transaction handling with auto-rollback on error.

💰 Money Helpers

Convert between dollar amounts and integer cents for database storage.

⚠️ Why cents? fastsql only supports int and str types — storing money as cents avoids floating-point precision issues.

Function	Direction	Example
`to_cents`	`"$150.00"` → `15000`	Store in DB
`from_cents`	`15000` → `"$150.00"`	Display to user

source

to_cents


def to_cents(
    dollars:str | float | None
)->int | None:

Convert dollar amount to integer cents for database storage.

source

from_cents


def from_cents(
    cents:int | None
)->str:

Convert integer cents to formatted dollar string for display.