pip install sqlalchemy-aip160from sqlalchemy import select
from sqlalchemy_aip160 import apply_filter, parse_filter, FilterBuilderquery = select(MyModel)
filtered = apply_filter(query, MyModel, 'status = "active"')filtered = apply_filter(query, MyModel, 'status = "active" AND priority > 3')filtered = apply_filter(query, MyModel, 'status = "active" OR status = "pending"')filtered = apply_filter(query, MyModel, 'NOT status = "inactive"')filtered = apply_filter(query, MyModel, 'name = "Widget*"')filtered = apply_filter(query, MyModel, 'category:*')filtered = apply_filter(
query, MyModel,
'(status = "active" OR status = "pending") AND priority >= 3'
)filtered = apply_filter(
query, MyModel, 'status = "active"',
allowed_fields={"status", "priority"} # Only these fields can be filtered
)filtered = apply_filter(
query, MyModel,
'department = "Engineering"',
field_aliases={"department": "department.name"}
)parse_filter returns a structured FilterExpression that can be inspected,
manipulated, serialized back to a string, or passed directly to apply_filter.
expr = parse_filter('status = "active" AND priority > 3')
expr.get_fields() # {'status', 'priority'}expr = parse_filter('kind = "issue" AND source_id = "abc"')
expr.rename_field("kind", "kind_str")
str(expr) # 'kind_str = "issue" AND source_id = "abc"'expr = parse_filter('status = "active" AND priority > 3')
expr.remove("priority")
str(expr) # 'status = "active"'expr = parse_filter('kind = "acc:issue" AND source_id = "abc"')
kind_clauses = expr.extract("kind") # Removes and returns kind comparisons
replacement = FilterBuilder().add("kind", "=", "acc:meeting").build()
combined = expr & replacement
str(combined) # 'source_id = "abc" AND kind = "acc:meeting"'user_filter = parse_filter('status = "active"')
server_filter = parse_filter('org_id = "org-123"')
combined = user_filter & server_filter # AND
either = user_filter | server_filter # OR
negated = ~user_filter # NOTf = FilterBuilder()
f.add("kind", "=", "issue").add("priority", ">", 3)
str(f) # 'kind = "issue" AND priority > 3'
# Pass directly to apply_filter
query = apply_filter(select(MyModel), MyModel, f.build())expr = parse_filter('status = "active"')
expr.rename_field("status", "state")
filtered = apply_filter(query, MyModel, expr)| Feature | Example | Notes |
|---|---|---|
| Equality | status = "active" | String, int, float, bool, UUID |
| Not equals | status != "inactive" | |
| Comparisons | priority > 3, score <= 4.5 | <, >, <=, >= |
| AND | a = 1 AND b = 2 | Explicit |
| Implicit AND | a = 1 b = 2 | Adjacent terms |
| OR | a = 1 OR a = 2 | Higher precedence than AND per AIP-160 |
| NOT | NOT status = "active" | Also -status = "active" |
| Parentheses | (a OR b) AND c | |
| Wildcards | name = "*.txt" | Converted to SQL LIKE |
| Presence | field:* | Field is not null |
| Has value | field:value | Field equals value |