Feature: Support simple type field matching query

[AngusWG]

• Convert sqlalchemy-model to pydantic-model using pydantic_sqlalchemy
• Support simple type field matching query

[vercel]

This pull request is being automatically deployed with Vercel (learn more).
To see the status of your deployment, click below or on the icon next to each commit.

🔍 Inspect: https://vercel.com/flortz/fastapi-crudrouter/FyVsT7CkSt7jX93gFWS5n1LoNwpj
✅ Preview: https://fastapi-crudrouter-git-fork-anguswg-supportmatchingquery-flortz.vercel.app

[awtkns]

Hi @AngusWG. Thanks for the PR.

Query filter parameters has been a feature that I've been looking to implement. I think this is a good first attempt, however I am not sure if relying on pydantic_sqlalchemy is the right move as this must be extendable to all the other crudrouters.

Ideally what were the features you were hoping for?

[AngusWG]

@awtkns I need to query the table form various dimensions.
I used a ternary operator like pymongo to design the interface in another project.

# Refer mongo  https://docs.mongodb.com/manual/reference/operator/query-comparison/
_operator_map = {
    None: lambda attr, v: attr.op("=")(v),  # k:v - k=v
    "eq": lambda attr, v: attr.op("=")(v),  # k__eq:v - k=v
    "gt": lambda attr, v: attr.op(">")(v),  # k__gt:v - k>v
    "gte": lambda attr, v: attr.op(">=")(v),  # k__gte:v - k>=v
    "in": lambda attr, v: attr.in_(json.loads(v)),  # k__in:v - k in v
    "lt": lambda attr, v: attr.op("<")(v),  # k__lt:v - k<v
    "lte": lambda attr, v: attr.op("<=")(v),  # k__lte:v - k<=v
    "ne": lambda attr, v: attr.op("!=")(v),  # k__ne:v - k!=v
    "nin": lambda attr, v: attr.notin_(json.loads(v)),  # k__nin:v k not in v
    "regex": lambda attr, v: attr.like("%{}%".format(v)),  # k__regex:v k contains v
}


def parse_operator(class_obj, key: str, value: str):
    """将查询请求转换mysql识别的 BinaryExpression
    eg。 x__gt:30 -> filter(x >= 30 )
    eg。 x__in:[1,2,3] -> filter(x.in_([1,2,3]))
    """
    if "__" in key:
        attr_name, _operator = key.split("__")
    else:
        _operator = None
        attr_name = key
    attr = class_obj.columns.get(attr_name)
    if attr is None:
        raise GrpcError("unknown field {}".format(attr_name))
    if _operator not in _operator_map.keys():
        raise GrpcError("unknown operator {}".format(key))
    return _operator_map[_operator](attr, value)
	
query_field: Dict[str, Union[str,int]]  = {"column_a__gt":20}
for field, value in query_field.items():
	query = query.filter(parse_operator(self.table, field, value))

hope this code will help you to make fastapi-crudrouter become more convenient

[awtkns]

Thanks @AngusWG. Based off of your first implementation I was able to implement query filters as a Fastapi Dependency

def query_factory(schema: Type[T]) -> Any:
    field_names = schema.__fields__.keys()

    _str = "{}: Optional[{}] = None"
    args_str = ", ".join(
        [
            _str.format(name, field.type_.__name__)
            for name, field in schema.__fields__.items()
            if field.type_ in FILTER_TYPES
        ]
    )

    _str = "{}={}"
    return_str = ", ".join(
        [
            _str.format(name, field.name)
            for name, field in schema.__fields__.items()
            if field.type_ in FILTER_TYPES
        ]
    )

    filter_func_src = f"""
def filter_func({args_str}):
    ret = dict({return_str})
    return {{k:v for k, v in ret.items() if v is not None}}
"""

    exec(filter_func_src, globals(), locals())
    return Depends(locals().get("filter_func"))

Now for all the routers it is as simple as doing:

 def route(
            db: Session = Depends(self.db_func),
            pagination: PAGINATION = self.pagination,
            filter_: FILTER = self.filter,
        ) -> List[Model]:
            skip, limit = pagination.get("skip"), pagination.get("limit")
            query = db.query(self.db_model).filter_by(**filter_)
            db_models: List[Model] = query.limit(limit).offset(skip).all()

I just need to finish it for the Ormar / memory router and then we should be good to merge

[awtkns]

Basic query filtering is good to go. Needs docs and then it can be released. Looking to implement more powerful query features in the future.

[fsecada01]

Any movement on this? I know that querying would make my life a lot easier.