How to Implement Dynamic Query Building with FastCRUD: A Complete Guide

FastCRUD enables dynamic query building by composing SQLAlchemy Select statements through a modular pipeline of builders, processors, and filters that handle joins, sorting, and pagination at runtime.

Dynamic query building with FastCRUD allows you to construct type-safe SQL queries based on runtime parameters without writing raw SQL. The benavlabs/fastcrud library provides a modular architecture where components like SQLQueryBuilder and FilterProcessor work together to generate complex database queries dynamically.

Architectural Overview of the Query Engine

FastCRUD implements a layered, protocol-driven architecture that separates concerns across distinct components. This design ensures the system remains ORM-agnostic while providing concrete SQLAlchemy implementations.

Protocol Layer. The abstract contracts defined in fastcrud/core/protocols.py establish interfaces such as QueryBuilder, CRUDInstance, and FilterProcessor. These protocols allow the rest of the codebase to interact with query components without depending on specific SQLAlchemy implementations.

Builder Layer. The SQLQueryBuilder class in fastcrud/core/query/builder.py provides the concrete implementation for constructing SQLAlchemy Select objects. It orchestrates the step-by-step assembly of queries.

Specialized Processors. Three dedicated processors handle specific query aspects:

• Filtering: FilterProcessor in fastcrud/core/filtering/processor.py parses flexible filter syntax like field__gt=5 or _or={...} into SQLAlchemy ColumnElement conditions
• Sorting: SortProcessor in fastcrud/core/query/sorting.py validates and generates ORDER BY clauses for single or multi-column sorting
• Joining: JoinBuilder in fastcrud/core/query/joins.py manages JOIN and OUTER JOIN clauses with automatic column selection from related models

Orchestration Functions. High-level helpers build_joined_query and execute_joined_query in fastcrud/core/query/builder.py coordinate the entire pipeline, integrating joins, counts, filters, sorting, and pagination into a single execution flow.

The Dynamic Query Building Pipeline

Implementing dynamic query building follows a seven-step composition pattern. Each step is optional; the builder returns the original statement when a step is omitted, enabling truly dynamic construction based on client-supplied parameters.

1. Instantiate SQLQueryBuilder with your model class
2. Build base select via builder.build_base_select() to create SELECT * FROM model
3. Prepare joins using builder.prepare_joins() with JoinBuilder to inject necessary table relationships
4. Apply filters by converting user parameters through FilterProcessor.parse_filters() and attaching them via builder.apply_filters()
5. Configure sorting through builder.apply_sorting(), which delegates to SortProcessor for direction validation
6. Add pagination via builder.apply_pagination(stmt, offset, limit)
7. Execute using execute_joined_query() to run the statement in an AsyncSession and return plain dictionary rows

Practical Implementation Examples

Simple List Endpoint with Optional Filters

This example demonstrates a flexible endpoint that accepts optional filtering, sorting, and pagination parameters:

from fastcrud.core.query import SQLQueryBuilder, build_joined_query, execute_joined_query
from fastcrud.core.filtering import FilterProcessor
from sqlalchemy.ext.asyncio import AsyncSession

async def list_users(
    db: AsyncSession,
    *,
    name: str | None = None,
    age__gt: int | None = None,
    sort: str | list[str] = None,
    order: str | list[str] = None,
    offset: int = 0,
    limit: int | None = None
):
    # Initialize builder and filter processor

    builder = SQLQueryBuilder(User)
    filter_proc = FilterProcessor(User)
    
    # Build the base query with filters

    stmt = build_joined_query(
        model=User,
        query_builder=builder,
        filter_processor=filter_proc,
        config={"join_definitions": [], "counts_config": []},
        **{"name": name, "age__gt": age__gt}
    )
    
    # Execute with optional sorting and pagination

    rows = await execute_joined_query(
        db=db,
        stmt=stmt,
        query_builder=builder,
        sort_columns=sort,
        sort_orders=order,
        offset=offset,
        limit=limit,
    )
    return rows

The config dictionary accepts join_definitions and counts_config; empty lists disable those features. Unrecognized filter keys are automatically ignored, making endpoints tolerant to missing parameters.

Querying Across Relationships with Joins

Dynamic query building supports dot-notation filtering across relationships. The following example joins User with Post and filters by post title:

from fastcrud.core.query import SQLQueryBuilder, build_joined_query, execute_joined_query
from fastcrud.core.filtering import FilterProcessor
from fastcrud.types import JoinConfig
from sqlalchemy.ext.asyncio import AsyncSession

async def users_with_posts(
    db: AsyncSession,
    *,
    post_title__ilike: str = None
):
    builder = SQLQueryBuilder(User)
    filter_proc = FilterProcessor(User)
    
    # Define join configuration

    post_join = JoinConfig(
        model=Post,
        join_on=User.id == Post.user_id,
        join_type="left",
        schema_to_select=None,
        alias=None,
    )
    
    # Build query with join and relationship filter

    stmt = build_joined_query(
        model=User,
        query_builder=builder,
        filter_processor=filter_proc,
        config={"join_definitions": [post_join], "counts_config": []},
        **{"posts.title__ilike": post_title__ilike}
    )
    
    rows = await execute_joined_query(db=db, stmt=stmt, query_builder=builder)
    return rows

The dot-notation (posts.title__ilike) triggers FilterProcessor._handle_joined_filter, which traverses the relationship chain (User → posts → title). Simultaneously, JoinBuilder.prepare_joins adds the LEFT OUTER JOIN and selects post columns automatically.

Custom Sorting with Multiple Columns

For explicit sorting control without the high-level executor:

stmt = builder.build_base_select()
sorted_stmt = builder.apply_sorting(
    stmt,
    sort_columns=["created_at", "name"],
    sort_orders=["desc", "asc"]
)

The SortProcessor validates column existence via get_model_column and ensures each order is either asc or desc, raising ValueError or ArgumentError for invalid inputs.

Key Source Files Reference

Understanding these core files is essential for advanced customization:

• fastcrud/core/protocols.py — Defines the QueryBuilder protocol and abstract contracts that enable ORM-agnostic operations
• fastcrud/core/query/builder.py — Contains SQLQueryBuilder and orchestration functions build_joined_query and execute_joined_query
• fastcrud/core/query/sorting.py — Implements SortProcessor for dynamic ORDER BY clause generation
• fastcrud/core/query/joins.py — Implements JoinBuilder for relationship handling and column selection
• fastcrud/core/filtering/processor.py — Parses flexible filter syntax including _or, _not, and dot-notation relationship filters
• fastcrud/types.py — Central type definitions including ModelType, JoinConfig, and SelectSchemaType

Summary

• FastCRUD uses a modular pipeline where SQLQueryBuilder, FilterProcessor, SortProcessor, and JoinBuilder collaborate to construct SQLAlchemy queries at runtime
• The protocol-based architecture in fastcrud/core/protocols.py ensures type safety and testability while remaining ORM-agnostic
• Dynamic composition allows optional application of filters, joins, sorting, and pagination based on runtime parameters
• Dot-notation filtering (posts.title__ilike) enables complex relationship queries without manual join syntax
• High-level functions build_joined_query and execute_joined_query in fastcrud/core/query/builder.py orchestrate the entire pipeline for common use cases

Frequently Asked Questions

How does FastCRUD handle security when building queries dynamically?

FastCRUD mitigates SQL injection risks by using SQLAlchemy’s expression language rather than string concatenation. The FilterProcessor in fastcrud/core/filtering/processor.py validates column names against the model’s actual fields and uses parameterized queries for all values, ensuring user input never reaches the database as raw SQL.

Can I use dynamic query building with complex multi-table joins?

Yes. The JoinConfig type in fastcrud/types.py supports complex join definitions including join types (left, inner, outer), custom join conditions, and nested relationships. Multiple JoinConfig objects can be passed in the join_definitions list to build_joined_query, enabling queries across several related tables with automatic column selection.

What performance considerations apply to dynamically built queries?

FastCRUD generates standard SQLAlchemy Select objects that benefit from SQLAlchemy’s caching and optimization. However, dynamic filters can sometimes prevent database query plan caching. For high-throughput applications, consider using the counts_config parameter in build_joined_query to include efficient COUNT queries alongside the main selection, or explicitly limit result sets using the limit parameter to avoid large dataset retrieval.

Is it possible to extend the filter syntax with custom operators?

The FilterProcessor class in fastcrud/core/filtering/processor.py supports standard operators like __gt, __lt, __eq, and __ilike out of the box. To add custom operators, you would subclass FilterProcessor and override the parsing methods, then inject your custom processor into the build_joined_query function via the filter_processor parameter.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}