Frontends, REST APIs and LLMs naturally produce structured JSON - not SQL.
filtersql
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.
Designed for three use cases:
DataTables server-side backendsCursor-based pagination(Access-style, no OFFSET)** Deterministic LLM pipelines**- the LLM generates JSON, filtersql compiles it to safe SQL
Supports PostgreSQL, SQLite, MySQL, and Oracle.
Secure by default- Fully parameterized queries, protected against SQL injection** Multi-database support**- PostgreSQL, SQLite, MySQL, and Oracle** Lightweight**- No ORM required. Works with any database driver (psycopg2, sqlite3, mysql-connector, cx_Oracle, etc.)** Language-agnostic**- Clean JSON protocol, perfect for REST APIs and frontend applications** High-performance pagination**- Keyset (cursor-based) pagination, avoiding slowOFFSET
queriesAI/LLM friendly- Designed for structured output from large language models
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.
From JSON:
{
"action": "select",
"source": "users",
"filters": [
{
"field": "name",
"operator": "icontains",
"value": "john"
}
]
}
to SQL:
select
*
from
"users"
where
"name" ilike '%' || ? || '%'
with values:
["john"]
pip install filtersql
python
import filtersql
ds = filtersql.Datasource(
source = 'users',
dbms = 'Pg',
placeholder = '%s',
)
query, values = ds.select(
columns = [
{'field': 'id'},
{'field': 'first_name'},
{'field': '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)
print(values)
cursor.execute(query, values)
rows = cursor.fetchall()
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']}
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'},
]},
]
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'},
]},
]},
]
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},
)
query, values = ds.select(columns=columns, filters=user_filters)
query, values = ds.delete(id={'id': 99})
query, values = ds.insert(values={'title': 'New doc'})
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
:
columns = [
{'field': 'COUNT(*) as total', 'raw': True},
{'field': 'MAX(age) as max_age', 'raw': True},
]
All three return (query, values)
like select()
.
query, values = ds.insert(
values = {'first_name': 'John', 'last_name': 'Smith'},
)
query, values = ds.insert(
values = {'first_name': 'John'},
returning = 'id',
)
query, values = ds.update(
id = {'id': 42},
values = {'first_name': 'John', 'status': 'active'},
)
query, values = ds.delete(id={'id': 42})
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'},
])
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
"""
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
.
Replaces placeholders with actual values for logging. Never use the output as real SQL.
query, values = ds.select(columns=columns, filters=filters)
print(ds.debug(query, values))
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
query = f"SELECT * FROM users WHERE {llm_output}" # SQL injection waiting to happen
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:
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", "in"]},
"value": {"type": "STRING"}
},
"required": ["field", "operator", "value"]
}
}
},
"required": ["source", "filters"]
}
response = client.models.generate_content(
model='gemini-flash',
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: Literal['doc_type', 'doc_date', 'author', 'title']
operator: Literal['=', '!=', '>', '>=', '<', '<=', 'icontains', 'between', 'in', 'notin']
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)
php
{'field': 'attributes->>amount', 'operator': '>=', 'value': '10000', 'value_type': 'numeric'}
{'field': 'attributes->>expiry', 'operator': '<=', 'value': '2025-12-31', 'value_type': 'date'}
Without the cast, '9' > '10'
as text gives wrong results for numeric ranges.
Navigates large tables without OFFSET - no performance degradation.
ds = filtersql.Datasource(
source = 'documents',
dbms = 'Pg',
order = [{'field': 'id', 'order': 'asc'}],
)
query, values = ds.select(
columns = columns,
cursor = {'id': last_seen_id},
direction = 'next',
)
query, values = ds.select(
columns = columns,
cursor = {'id': last_seen_id},
direction = 'prev',
)
query, values = ds.select(
columns = columns,
cursor = {'id': 42},
direction = 'seek',
)
direction |
SQL | Use case |
|---|---|---|
'next' |
field > last_value |
Forward navigation |
'prev' |
field < last_value |
Backward navigation |
'seek' |
field = value |
Jump to specific record |
Multi-column cursors work too:
query, values = ds.select(
columns = columns,
cursor = {'first_name': last_first, 'last_name': last_last},
direction = 'next',
)
{'field': 'tsv_content', 'operator': 'fts', 'value': 'oxygen supply'}
{'field': 'description', 'operator': 'fts_query', 'value': 'oxygen supply'}
ds = filtersql.Datasource(..., fts_language='italian')
{'field': 'content', 'operator': 'fts', 'value': 'oxygen supply'}
| Operator | Description |
|---|---|
= != > >= < <= |
|
| Comparison | |
between |
|
Range - value must be [low, high] |
|
contains |
|
| Case-sensitive substring | |
starts_with |
|
| Case-sensitive prefix | |
ends_with |
|
| Case-sensitive suffix | |
not_contains |
|
| Case-sensitive substring exclusion | |
not_starts_with |
|
| Case-sensitive prefix exclusion | |
icontains |
|
| Case-insensitive substring | |
istarts_with |
|
| Case-insensitive prefix | |
iends_with |
|
| Case-insensitive suffix | |
not_icontains |
|
| Case-insensitive substring exclusion | |
null / notnull |
|
| NULL checks - no value needed | |
in / notin |
|
| List membership | |
reverse_in |
|
Value in a set of columns - ? IN (col1, col2) |
|
regexp |
|
| Case-sensitive regular expression | |
iregexp |
|
Case-insensitive regexp (Pg: ~* , Oracle: regexp_like with i ) |
|
fts |
|
| Full-text search on indexed column (Pg, MySQL) | |
fts_query |
|
| Full-text search on text column (Pg only) |
| DBMS | dbms value |
Default placeholder |
|---|---|---|
| PostgreSQL | 'Pg' |
%s |
| SQLite | 'SQLite' |
? |
| MySQL | 'mysql' |
? |
| Oracle | 'Oracle' |
? |
Override with any placeholder your driver expects:
ds = filtersql.Datasource(..., placeholder='%s')
ds = filtersql.Datasource(..., placeholder=':val')
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
MIT