DatabaseDataSource

The DatabaseDataSource uses SQLAlchemy statements to fetch data from the database. This is the most common data source for production use.

Basic Usage

from sqlalchemy import select

from ckan import model

from ckanext.tables.shared import DatabaseDataSource


data_source = DatabaseDataSource(
    stmt=select(
        model.User.id,
        model.User.email,
        model.User.name,
        model.User.state,
    ).order_by(model.User.created.desc()),
)

Definition

A data source that uses a SQLAlchemy statement as the data source.

PARAMETER	DESCRIPTION
stmt	The SQLAlchemy statement to use as the data source TYPE: Select[Any]

Source code in ckanext/tables/data_sources.py

class DatabaseDataSource(BaseDataSource):
    """A data source that uses a SQLAlchemy statement as the data source.

    Args:
        stmt: The SQLAlchemy statement to use as the data source
    """

    def __init__(self, stmt: Select[Any]):
        self.base_stmt = stmt
        self.stmt = stmt

    def filter(self, filters: list[FilterItem]) -> Self:
        self.stmt = self.base_stmt

        for filter_item in filters:
            if not hasattr(self.stmt.selected_columns, filter_item.field):
                continue

            col = getattr(self.stmt.selected_columns, filter_item.field)
            expr = self.build_filter(col, filter_item.operator, filter_item.value)

            if expr is not None:
                self.stmt = self.stmt.where(expr)

        return self

    def build_filter(self, column: ColumnElement[Any], operator: str, value: str) -> ColumnElement[bool] | None:
        try:
            if isinstance(column.type, Boolean):
                casted_value = str(value).lower() in ("true", "1", "yes", "y")
            elif isinstance(column.type, Integer):
                casted_value = int(value)
            elif isinstance(column.type, DateTime):
                casted_value = datetime.fromisoformat(value)
            else:
                casted_value = str(value)
        except ValueError:
            return None

        operators: dict[
            str,
            Callable[[ColumnElement[Any], Any], ColumnElement[bool] | None],
        ] = {
            "=": lambda col, val: col == val,
            "<": lambda col, val: col < val,
            "<=": lambda col, val: col <= val,
            ">": lambda col, val: col > val,
            ">=": lambda col, val: col >= val,
            "!=": lambda col, val: col != val,
            "like": lambda col, val: col.ilike(f"%{val}%") if isinstance(val, str) else None,
        }

        func = operators.get(operator)
        return func(column, casted_value) if func else None

    def sort(self, sort_by: str | None, sort_order: str | None) -> Self:
        if not sort_by or not hasattr(self.stmt.selected_columns, sort_by):
            return self

        col = getattr(self.stmt.selected_columns, sort_by)

        # Clear existing order_by clauses
        self.stmt = self.stmt.order_by(None)

        if sort_order and sort_order.lower() == "desc":
            self.stmt = self.stmt.order_by(col.desc())
        else:
            self.stmt = self.stmt.order_by(col.asc())

        return self

    def paginate(self, page: int, size: int) -> Self:
        if page and size:
            self.stmt = self.stmt.limit(size).offset((page - 1) * size)

        return self

    def all(self) -> list[dict[str, Any]]:
        return [self.serialize_row(row) for row in model.Session.execute(self.stmt).mappings().all()]

    def serialize_row(self, row: RowMapping) -> dict[str, Any]:
        return dict(row)

    def count(self) -> int:
        return model.Session.execute(select(func.count()).select_from(self.stmt.subquery())).scalar_one()

    def get_columns(self) -> list[str]:
        return [c.name for c in self.stmt.selected_columns]

Best Practices

• Use selective queries
• Add database indexes for commonly filtered/sorted columns

# Good: Select only needed columns
stmt = select(model.User.id, model.User.name, model.User.email)

# Avoid: Selecting everything
stmt = select(model.User)