filtersql

A lightweight Python library that translates structured JSON into safe, parameterized SQL.

Frontends, REST APIs and LLMs naturally produce structured JSON - not SQL. filtersql is a Python package that defines a declarative JSON query language and compiles it into safe, parameterized SQL.

JSON payload (or Python dicts) → filtersql → SQL string + values list

It is intentionally not an ORM or a connection manager. It builds the query and returns it. Query execution remains the responsibility of the caller, making filtersql compatible with any Python DB driver (psycopg2, sqlite3, asyncpg, etc.).

Designed for three use cases:

• DataTables server-side backends
• Cursor-based pagination (Access-style, no OFFSET)
• Deterministic LLM pipelines - the LLM generates JSON, filtersql compiles it to safe SQL

Supports PostgreSQL, SQLite, MySQL, DuckDB and Oracle.

Features

• Python-native & Lightweight - No ORM required. Works seamlessly with your favorite Python database driver
• Secure by default - Fully parameterized queries, protected against SQL injection
• Multi-database support - PostgreSQL, SQLite, MySQL, DuckDB and Oracle
• Language-agnostic - Clean JSON protocol, perfect for REST APIs and frontend applications
• High-performance pagination - Keyset (cursor-based) pagination, avoiding slow OFFSET queries
• AI/LLM friendly - Designed for structured output from large language models
• Server-side allowlists - constrain which columns and sources a client can name, exact-match, no normalization

Why filtersql?

filtersql is intentionally:

• stateless
• deterministic
• driver agnostic
• parameterized
• declarative
• portable
• JSON-first
• LLM friendly

Applications describe what they want. filtersql decides how to express it in SQL.

Example

From JSON:

{
  "action": "select",
  "source": "users",
  "filters": [
    {
      "field": "name",
      "operator": "icontains",
      "value": "john"
    }
  ]
}

to SQL:

select
  *
from
  "users"
where
  ("name" like ? escape '\')

with values:

["%john%"]

Install

pip install filtersql

Quick start

import filtersql

ds = filtersql.Datasource(
    source='users',
    dbms='SQLite',
    placeholder='?',
)

query, values = ds.select(
    columns=['id', 'first_name', 'last_name'],
    filters=[
        {'field': 'first_name', 'operator': 'icontains', 'value': 'John'},
        {'field': 'last_name',  'operator': 'icontains', 'value': 'Smith'},
    ],
    order=[{'field': 'first_name', 'order': 'asc'}],
    limit={'start': 0, 'length': 10},
)

print(query)
# select
#   "id",
#   "first_name",
#   "last_name"
# from
#   "users"
# where
#   ("first_name" like ? escape '\' and "last_name" like ? escape '\')
# order by
#   "first_name" asc
# limit 0, 10

print(values)
# ['%John%', '%Smith%']

Filters

Filter format

Every filter is a dict with three keys:

{'field': 'status', 'operator': '=', 'value': 'active'}

• field - column name, JSONB path (attributes->>amount), or schema-prefixed (m.doc_type)
• operator - the comparison operator (see table below)
• value - the value to compare against

A flat list of filters is joined with AND:

filters = [
    {'field': 'status',   'operator': '=',         'value': 'active'},
    {'field': 'doc_date', 'operator': '>=',        'value': '2025-01-01'},
    {'field': 'title',    'operator': 'icontains', 'value': 'oxygen'},
]

Some operators don't need a value:

{'field': 'deleted_at', 'operator': 'null'}
{'field': 'deleted_at', 'operator': 'notnull'}

Some operators take a list:

{'field': 'status',   'operator': 'in',      'value': ['active', 'pending']}
{'field': 'doc_date', 'operator': 'between', 'value': ['2025-01-01', '2025-12-31']}

OR groups

Wrap filters in a dict with key 'or':

filters = [
    {'field': 'status', 'operator': '=', 'value': 'active'},
    {'or': [
        {'field': 'first_name', 'operator': 'icontains', 'value': 'john'},
        {'field': 'last_name',  'operator': 'icontains', 'value': 'john'},
    ]},
]
# → "status" = ? AND ("first_name" like ? escape '\' OR "last_name" like ? escape '\')

Nesting

OR groups can contain AND groups and vice versa:

filters = [
    {'or': [
        {'field': 'doc_type', 'operator': '=', 'value': 'CONTRACT'},
        {'and': [
            {'field': 'doc_type', 'operator': '=',  'value': 'ORDER'},
            {'field': 'amount',   'operator': '>=', 'value': '10000'},
        ]},
    ]},
]
# → ("doc_type" = ? OR ("doc_type" = ? AND "amount" >= ?))

scope

scope is a dict of fixed field: value pairs applied as = filters to every operation on the Datasource - select, where, update, delete, and insert.

ds = filtersql.Datasource(
    source      = 'documents',
    dbms        = 'Pg',
    placeholder = '%s',
    scope       = {'tenant_id': 42},
)

# scope is always added
query, values = ds.select(columns=columns, filters=user_filters)
# → ... WHERE "tenant_id" = %s AND ...

query, values = ds.delete(id={'id': 99})
# → ... WHERE "tenant_id" = %s AND "id" = %s

query, values = ds.insert(values={'title': 'New doc'})
# → INSERT INTO "documents" ("title", "tenant_id") VALUES (%s, %s)

Note on column names: the keys of values (in insert and update) and of id (in update and delete) must be plain column names - no dots, no JSONB paths, no quoting. This is what makes the scope collision check reliable across dialects: a client can't reference a scoped column under a different spelling (e.g. users.tenant_id, or tenant_id with trailing whitespace) and slip past the check.

Allowlists (allowed_columns, allowed_sources)

scope protects rows; the allowlists protect what a client may name. Both are exact-match sets of strings, set server-side when you build the Datasource:

ds = filtersql.Datasource(
    source          = 'users',
    dbms            = 'Pg',
    scope           = {'tenant_id': 42},
    allowed_columns = {'id', 'name', 'status', 'created_at'},
    allowed_sources = {'users', 'orders'},
)

With allowed_columns set, every column a client can reference is checked against it - columns, filters and having (including nested and/or groups), order, group_by, cursor, returning, and the keys of values and id. A name that is not listed raises ValidationError.

• Exact match, nothing is normalized. Name, name , users.name and a Unicode look-alike are all different strings, and none of them matches name. List a column exactly as the client will send it, including a JSONB path like attributes->>amount.
• SELECT * is refused: columns must be given explicitly.
• raw columns and filters are refused, even with allow_raw_fields=True: a raw string is SQL, not a column name, so it cannot be allowlisted.
• scope columns do not have to be listed. Scope is added by the server after the check, so a client that is not allowed to name tenant_id can neither filter on it nor set it.
• An alias is an output label, not a column reference, and is not checked.
• None (the default) keeps the old behaviour. An empty collection is not "no restriction": it allows nothing.

allowed_sources is checked the same way against source on every action, and applies to a raw source too (the whole string must be listed). Neither parameter can be supplied in the payload of filtersql().

Use the allowlist as the primary control; identifier validation (no dots, quotes, JSONB paths in write keys) stays underneath it as a safety net for callers that forget to set one.

Columns

Columns are passed per call to select(), not at construction time - they can be computed at runtime.

Each column is a dict with a field key. Any extra keys (label, visible, editable etc.) are ignored by filtersql and can be used by your frontend layer:

columns = [
    {'field': 'id',         'label': 'ID',         'visible': True,  'editable': False},
    {'field': 'first_name', 'label': 'First Name', 'visible': True,  'editable': True},
    {'field': 'last_name',  'label': 'Last Name',  'visible': True,  'editable': True},
]

Plain strings also work:

columns = ['id', 'first_name', 'last_name']

Raw expressions with raw=True, but since this bypasses quoting/escaping entirely never pass untrusted input as field when set, and remember to set allow_raw_fields=True on the corresponding Datasource:

ds = filtersql.Datasource(source='users', dbms='Pg', allow_raw_fields=True)

columns = [
    {'field': 'COUNT(*) as total', 'raw': True},
    {'field': 'MAX(age) as max_age', 'raw': True},
]

Note: raw=True bypasses quoting/escaping entirely — never pass untrusted input as field when set.

Raw source (raw_source)

By default, source is always treated as an identifier and quoted accordingly. If you need to pass a subquery instead of a plain table name (e.g. "(SELECT * FROM users WHERE active = true) AS active_users"), set raw_source=True — this tells filtersql to use the source string directly in the SQL, unquoted.

Because raw_source=True disables quoting entirely, it also requires an explicit allow_raw_source=True on the Datasource — this two-flag requirement exists specifically so this feature can't be turned on by accident or by a value that came from user input.

ds = filtersql.Datasource(
    source="(SELECT * FROM users WHERE active = true) AS active_users",
    raw_source=True,
    allow_raw_source=True,
    dbms='Pg'
)

Note: raw_source=True bypasses quoting entirely — only use it with trusted, hardcoded values. Never build source from user input:

# UNSAFE — never do this
ds = filtersql.Datasource(
    source=request.GET.get('table'),
    raw_source=True,
    allow_raw_source=True,
    dbms='Pg'
)

Insert, update, delete

All three return (query, values) like select().

# insert
query, values = ds.insert(
    values = {'first_name': 'John', 'last_name': 'Smith'},
)

# insert with returning (PostgreSQL only)
query, values = ds.insert(
    values    = {'first_name': 'John'},
    returning = 'id',
)

# update - id can be single or composite key
query, values = ds.update(
    id     = {'id': 42},
    values = {'first_name': 'John', 'status': 'active'},
)
# values order: SET values first, then WHERE values
# → ['John', 'active', 42]

# delete
query, values = ds.delete(id={'id': 42})

Keys in values and id must be plain column names. Qualified names (users.id) and JSONB paths (data->>x) are rejected — use filters in select() for non-key filtering.

where() - just the WHERE clause

When you want to inject filters into your own handwritten query:

ds = filtersql.Datasource(source='documents', dbms='Pg', placeholder='%s')

where_clause, values = ds.where(filters=[
    {'field': 'doc_type', 'operator': '=',  'value': 'CONTRACT'},
    {'field': 'doc_date', 'operator': '>=', 'value': '2025-01-01'},
])
# → '("doc_type" = %s and "doc_date" >= %s)'
# → ['CONTRACT', '2025-01-01']

query = f"""
    SELECT v.chunk, m.doc_type
    FROM file_vectors v
    JOIN file_metadata m ON v.sha256 = m.sha256
    WHERE v.model_name = %s
    AND {where_clause}
    ORDER BY v.embedding <=> %s
"""

Group By and Having

filtersql now supports GROUP BY and HAVING for aggregations and analytics.

query, values = ds.select(
    columns=[
        {'field': 'status'},
        {'field': 'count(*)', 'raw': True, 'alias': 'total'}
    ],
    filters=[{'field': 'age', 'operator': '>=', 'value': 18}],
    group_by=['status'],
    having=[{'field': 'count(*)', 'raw': True, 'operator': '>', 'value': 5}],
    order=[{'field': 'total', 'order': 'desc'}]
)

filtersql() - convenience function

Builds any query from a single payload dict. Useful for REST APIs and AI-generated actions:

from filtersql import filtersql

query, values = filtersql(
    payload = {
        'action':  'select',
        'source':  'users',
        'columns': [{'field': 'id'}, {'field': 'first_name'}],
        'filters': [{'field': 'status', 'operator': '=', 'value': 'active'}],
        'order':   [{'field': 'id', 'order': 'asc'}],
        'limit':   {'start': 0, 'length': 10},
    },
    dbms        = 'Pg',
    placeholder = '%s',
)

Supported actions: select, insert, update, delete.

debug()

Replaces placeholders with actual values for logging. For security reasons, never use the output as real SQL.

query, values = ds.select(columns=columns, filters=filters)
print(ds.debug(query, values))
# select
#   "id",
#   "first_name"
# from
#   "users"
# where
#   ("first_name" ilike '%John%' escape '\')

Disabled by default — set FILTERSQL_DEBUG=1 in your environment to enable, since this should never run in production

Wildcard Escape in Pattern Matching

filtersql automatically escapes the wildcard characters that matter for whichever mechanism is in play - % and _ for like/ilike, or *, ? and [ for SQLite's glob - so searching for "100%" finds the literal string "100%", not "100" followed by anything.

The escaping is applied to the value before it is bound: the generated SQL contains no wildcard markers, and the pattern itself (e.g. '%100\%%') is what the driver binds. This only affects operators that use patterns (contains, starts_with, etc.). Other operators (=, >, in, ...) are unchanged.

AI / LLM pipelines

LLMs should never generate SQL directly. They should generate structured intent. filtersql validates that intent and compiles it into parameterized SQL. The values are always parameterized - no SQL injection risk even if the model produces unexpected output.

user question → LLM → JSON → filtersql → parameterized SQL

# DANGEROUS!
# the naive approach everyone starts with
query = f"SELECT * FROM users WHERE {llm_output}"  # SQL injection waiting to happen

# with filtersql
payload = json.loads(llm_response)
query, values = filtersql(payload, dbms='Pg', placeholder='%s')
cursor.execute(query, values)  # always parameterized, always safe

The simplest version without Pydantic:

Note: the schema below restricts value to a plain string, so it only works with scalar operators (=, icontains, etc.). Operators that need a list (in, between) require widening value's schema type accordingly.

import json
from google import genai
from filtersql import filtersql

SCHEMA = {
    "type": "OBJECT",
    "properties": {
        "source": {"type": "STRING", "enum": ["users", "contracts"]},
        "filters": {
            "type": "ARRAY",
            "items": {
                "type": "OBJECT",
                "properties": {
                    "field":    {"type": "STRING"},
                    "operator": {"type": "STRING", "enum": ["=", "!=", ">", ">=", "<", "<=", "icontains"]},
                    "value":    {"type": "STRING"}
                },
                "required": ["field", "operator", "value"]
            }
        }
    },
    "required": ["source", "filters"]
}

response = client.models.generate_content(
    model='gemini-3.1-flash-lite',
    contents=user_question,
    config=genai.types.GenerateContentConfig(
        system_instruction="Convert user questions into SQL filter payloads.",
        response_mime_type="application/json",
        response_schema=SCHEMA,
        temperature=0.0,
    )
)

payload = json.loads(response.text)
query, values = filtersql(payload, action='select', dbms='SQLite', placeholder='?')

With Pydantic for stricter validation:

from pydantic import BaseModel, Field
from typing import List, Literal

class SQLFilter(BaseModel):
    field: str
    operator: Literal[
        '=', '!=', '>', '>=', '<', '<=', 
        'contains', 'not_contains', 'icontains', 'not_icontains',
        'starts_with', 'not_starts_with', 'istarts_with', 'not_istarts_with',
        'ends_with', 'not_ends_with', 'iends_with', 'not_iends_with',
        'regexp', 'iregexp', 'not_regexp', 'not_iregexp',
        'null', 'notnull', 'fts', 'fts_query'
    ]
    value: str

class QuerySchema(BaseModel):
    semantic_query: str
    sql_filters: List[SQLFilter]

parsed = QuerySchema.model_validate_json(response.text)
filters = [f.model_dump() for f in parsed.sql_filters]

ds = filtersql.Datasource(source='documents', dbms='Pg', placeholder='%s')
where_clause, values = ds.where(filters=filters)

Empty list handling (in / notin)

If a frontend or LLM passes an empty array as a filter value (e.g. unselected checkboxes), filtersql handles it gracefully without breaking SQL syntax:

{'field': 'category_id', 'operator': 'in',    'value': []} # → 1 = 0
{'field': 'category_id', 'operator': 'notin', 'value': []} # → 1 = 1

JSONB and value_type (PostgreSQL)

{'field': 'attributes->>amount',  'operator': '>=', 'value': '10000',     'value_type': 'numeric'}
# → ("attributes"->>'amount')::numeric >= %s::numeric

{'field': 'attributes->>expiry',  'operator': '<=', 'value': '2025-12-31', 'value_type': 'date'}
# → ("attributes"->>'expiry')::date <= %s::date

Without the cast, '9' > '10' as text gives wrong results for numeric ranges.

Cursor-based pagination

Navigates large tables without OFFSET - no performance degradation.

ds = filtersql.Datasource(
    source = 'documents',
    dbms   = 'Pg',
    order  = [{'field': 'id', 'order': 'asc'}],
)

# forward
query, values = ds.select(
    columns   = columns,
    cursor    = {'id': last_seen_id},
    direction = 'next',
)

# backward
query, values = ds.select(
    columns   = columns,
    cursor    = {'id': last_seen_id},
    direction = 'prev',
)

# jump to specific record
query, values = ds.select(
    columns   = columns,
    cursor    = {'id': 42},
    direction = 'seek',
)

Multi-column cursors work too:

ds = filtersql.Datasource(
    source = 'documents',
    dbms   = 'Pg',
    order  = [
        {'field': 'last_name',  'order': 'asc'},
        {'field': 'first_name', 'order': 'asc'},
    ],
)

query, values = ds.select(
    columns   = columns,
    cursor    = {'first_name': last_first, 'last_name': last_last},
    direction = 'next',
)

Full-text search

PostgreSQL

# pre-built tsvector column
{'field': 'tsv_content', 'operator': 'fts', 'value': 'oxygen supply'}
# → "tsv_content" @@ websearch_to_tsquery('english', %s)

# on the fly from a text column
{'field': 'description', 'operator': 'fts_query', 'value': 'oxygen supply'}
# → to_tsvector('english', coalesce("description", '')) @@ websearch_to_tsquery('english', %s)

ds = filtersql.Datasource(..., fts_language='italian')

MySQL

{'field': 'content', 'operator': 'fts', 'value': 'oxygen supply'}
# → match("content") against(? in boolean mode)

Filter operators

Supported databases

dbms is case-insensitive - 'pg', 'PG', 'Pg' are all equivalent. Unrecognized names (e.g. 'postgres') still raise a ConfigurationError listing the valid names above.

Override with any placeholder your driver expects:

ds = filtersql.Datasource(..., placeholder='%s')
ds = filtersql.Datasource(..., placeholder='$1')

Examples

Working examples are in the /examples folder:

• examples/datatables/ - Flask + DataTables server-side with column filters and global search
• examples/pagination/ - REST API gateway with security whitelist for insert/update/delete
• examples/ai/ - Gemini structured output → filtersql → SQLite
• examples/pandas_duckdb/ - Query Pandas DataFrames using DuckDB + filtersql

License

MIT