Allowing for advanced FilterConfigs in crud_router

[Simon128]

Describe the bug or question I'm not sure whether this is a bug or intentional, but the _validate_filter_config method of EndpointCreator does not allow for more advanced query options as filters for the filter_config argument of the crud_router function.

To Reproduce Postgres database required (can probably be easily changed though)

import os
from typing import AsyncGenerator
from fastapi import FastAPI
from fastcrud import FilterConfig, crud_router
from functools import partial
import uvicorn
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from uuid import UUID, uuid4
from sqlalchemy import String
from sqlalchemy.dialects.postgresql import UUID as SQLUUID
from sqlalchemy.orm import mapped_column, Mapped
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import DateTime
from sqlalchemy.sql import func
from datetime import datetime
from pydantic import BaseModel

class Base(DeclarativeBase):
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=func.now())

class Dataset(Base):
    """
    Dataset Model
    """
    __tablename__ = 'dataset'

    id: Mapped[UUID] = mapped_column(SQLUUID(as_uuid=True), primary_key=True, default=uuid4)
    name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    directory_path: Mapped[str] = mapped_column(String, unique=True, nullable=False)

class DatasetCreateSchema(BaseModel):
    name: str
    directory_path: str
class DatasetUpdateSchema(BaseModel):
    id: UUID
    name: str
    directory_path: str

async def get_session_async(async_session) -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

def create_app():
    pg_host = os.environ.get("POSTGIS_HOST", "localhost")
    assert pg_host
    pg_user = os.environ.get("POSTGIS_USER", "postgis")
    assert pg_user
    pg_pw = os.environ.get("POSTGIS_PASSWORD", "postgis")
    assert pg_pw
    pg_db = os.environ.get("POSTGIS_DB", "rsdatval")
    assert pg_db

    engine = create_async_engine(f"postgresql+asyncpg://{pg_user}:{pg_pw}@{pg_host}/{pg_db}", echo=False)
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) # type:ignore
    get_session = partial(get_session_async, async_session)

    app = FastAPI()
    dataset_router = crud_router(
        session=get_session,
        model=Dataset,
        create_schema=DatasetCreateSchema,
        update_schema=DatasetUpdateSchema,
        path="/dataset",
        tags=["Dataset"],
        filter_config=FilterConfig(filters={"id": None, "name": None, "directory_path": None, "name__startswith": None}),
    )
    app.include_router(dataset_router)
    return app

if __name__ == "__main__":
    app = create_app()
    uvicorn.run(app, host="0.0.0.0", port=4000)

throws:

nvim-dap: Thread stopped due to exception of type ValueError (unhandled)
Description: Invalid filter column 'name__startswith': not found in model 'Dataset' columns
Stack trace:
  File "/home/simon/playground/test.py", line 59, in create_app
    dataset_router = crud_router(
                     ^^^^^^^^^^^^
  File "/home/simon/playground/test.py", line 72, in <module>
    app = create_app()
          ^^^^^^^^^^^^
ValueError: Invalid filter column 'name__startswith': not found in model 'Dataset' columns

Description This might be an easy fix, by just changing the _validate_filter_config method to:

def _validate_filter_config(self, filter_config: FilterConfig) -> None:
    model_columns = self.crud.model_col_names
    for key in filter_config.filters.keys():
        if "__" in key:
            field_name, op = key.rsplit("__", 1)
            if op not in self.crud._SUPPORTED_FILTERS:
                raise ValueError(
                    f"Invalid filter op '{op}': following filter ops are allowed: {self.crud._SUPPORTED_FILTERS.keys()}"
                )
        else:
            field_name = key

        if field_name not in model_columns:
            raise ValueError(
                f"Invalid filter column '{key}': not found in model '{self.model.__name__}' columns"
            )

I have not tested this thoroughly as It is my first day dealing with this repo and I'm not sure whether this would violate some concept/idea.

[Simon128]

Edit: My proposed fix is basically just mirroring the behaviour of the _parse_filter method.

A current possible workaround consists of creating a custom EndpointCreator. E.g.

class CustomEndpointCreator(EndpointCreator):
    def _validate_filter_config(self, filter_config: FilterConfig) -> None:
        model_columns = self.crud.model_col_names
        for key in filter_config.filters.keys():
            if "__" in key:
                field_name, op = key.rsplit("__", 1)
                if op not in self.crud._SUPPORTED_FILTERS:
                    raise ValueError(
                        f"Invalid filter op '{op}': following filter ops are allowed: {self.crud._SUPPORTED_FILTERS.keys()}"
                    )
            else:
                field_name = key

            if field_name not in model_columns:
                raise ValueError(
                    f"Invalid filter column '{key}': not found in model '{self.model.__name__}' columns"
                )

One could also have like a default filter configs generator, which automatically allows for all advanced queries:

from pydantic import BaseModel
from typing import Dict, Type

def generate_filter_configs(pydantic_model: Type[BaseModel]) -> Dict[str, str]:
    filter_config = {}

    for field_name, field in pydantic_model.model_fields.items():
        field_type = field.annotation
        filter_config[field_name] = None

        if field_type == str:
            filter_config[f"{field_name}__like"] = None
            filter_config[f"{field_name}__notlike"] = None
            filter_config[f"{field_name}__ilike"] = None
            filter_config[f"{field_name}__notilike"] = None
            filter_config[f"{field_name}__startswith"] = None
            filter_config[f"{field_name}__endswith"] = None
            filter_config[f"{field_name}__contains"] = None

        elif field_type in {int, float}:
            filter_config[f"{field_name}__gt"] = None
            filter_config[f"{field_name}__lt"] = None
            filter_config[f"{field_name}__gte"] = None
            filter_config[f"{field_name}__lte"] = None
            filter_config[f"{field_name}__ne"] = None
            filter_config[f"{field_name}__between"] = None

        elif field_type == bool:
            filter_config[f"{field_name}__is"] = None
            filter_config[f"{field_name}__is_not"] = None

        elif field_type.__name__ == list.__name__:
            filter_config[f"{field_name}__in"] = None
            filter_config[f"{field_name}__not_in"] = None

    return filter_config

and then:

    dataset_router = crud_router(
        session=get_session,
        model=Dataset,
        create_schema=DatasetCreateSchema,
        update_schema=DatasetUpdateSchema,
        endpoint_creator=CustomEndpointCreator,
        path="/dataset",
        tags=["Dataset"],
        filter_config=FilterConfig(filters=generate_filter_configs(DatasetFilterSchema))
    )

Just some ideas I have.

[doubledare704]

@Simon128 does this changes will solve an issue? https://github.com/igorbenav/fastcrud/pull/204

[doubledare704]

@Simon128 #204 was accepted, could you try this feature from main branch ?

[igorbenav]

Fixed by #204