[FEAT] Different dependencies per function

[DorskFR]

Hello,

I am looking at modifying your crudrouter to work with Marmelab's react-admin.

To work with their Listview, I added pagination, sort column and order, column filter and result count to the get_all function.

To work with their Edit and Create views, I have permissions rules on the front end but I would like to have some on the backend too.

I added a dependency to the router and check there the authenticated status.
I suppose I could extend it to check the method and path and allow / restrict access based on this and permissions.
However I thought it would also be convenient to have different dependencies per route/function as these rules might be different from collection to collection.

Something like this:

users_router = SQLAlchemyCRUDRouter(
    schema=User,
    create_schema=UserCreate,
    db_model=UserModel,
    db=get_db,
    prefix='users',
    dependencies=[Depends(token_auth)],
    delete_one_route_dependencies=[Depends(must_be_admin)],
    delete_all_route_dependencies=[Depends(must_be_admin)],
)

I am not sure this is a good idea but if it seems interesting, where would you suggest I look at implementing this?
I looked in your code but was not sure if this should be in CRUDGenerator or in add_api_route.

In any case, great project, thank you !

[awtkns]

Hey thanks for the feedback! Definitely excited to see crud router in action with the dashboard.

To work with their Listview, I added pagination, sort column and order, column filter and result count to the get_all function.

I actually added pagination support today with limit and skip to the get_all function (docs). Curious to see your implementation though; I did not add support for filtering yet.

With regards to what you are proposing with the dependencies, I think that would make sense. Especially for routes like delete_all as you mentioned.

Implementation wise, the only file you would need to change is /core/_base.py. You can probably get away with just tacking on the dependencies at the end of each super().add_api_route(....) call in the constructor.

if get_all_route:
    super().add_api_route('', self._get_all(), methods=['GET'], response_model=Optional[List[self.schema]], summary='Get All')

if create_route:
    super().add_api_route('', self._create(), methods=['POST'], response_model=self.schema, summary='Create One')

if delete_all_route:
    super().add_api_route('', self._delete_all(), methods=['DELETE'], response_model=Optional[List[self.schema]], summary='Delete All')

if get_one_route:
    super().add_api_route('/{item_id}', self._get_one(), methods=['GET'], response_model=self.schema, summary='Get One')

if update_route:
    super().add_api_route('/{item_id}', self._update(), methods=['PUT'], response_model=self.schema, summary='Update One')

if delete_one_route:
    super().add_api_route('/{item_id}', self._delete_one(), methods=['DELETE'], response_model=self.schema, summary='Delete One')

Let me know if you had any other questions! PR's Always welcome 😄

[DorskFR]

Thanks for your answer

For now I cheated using the Request object in a dependency:
if request.method in ["PUT", "DELETE"] and not "admin" in permissions:
but will definitely look where you indicated, thank you.

With react-admin the path params are like this:
"GET /products?filter=%7B%7D&range=%5B0%2C4%5D&sort=%5B%22code%22%2C%22ASC%22%5D HTTP/1.1" 200 OK

so:

'filter': | {}
'range': | [0,4]
'sort': | ["code","ASC"]

React-admin also needs to receive a "content-range" header with the total count of items.

Therefore I did the following.

Imports:

from typing import Callable, Optional
from fastapi import Depends, HTTPException, Query, Response
from re import sub
from json import loads

try:
    from sqlalchemy.orm import Session
    from sqlalchemy.ext.declarative import DeclarativeMeta
    from sqlalchemy.exc import IntegrityError
    from sqlalchemy.inspection import inspect
    from sqlalchemy.sql import text, desc

Then, I had to use some aliases because filter, sort, range cannot be used in Python (at least I had some errors).

def _get_all(self) -> Callable:
        def route(
            response: Response,
            db: Session = Depends(self.db_func),
            q_range: Optional[str] = Query(None, alias="range"),
            q_filter: Optional[str] = Query(None, alias="filter"),
            q_sort: Optional[str] = Query(None, alias="sort"),
            pagination: dict = self.pagination,

        ):

            q = db.query(self.db_model)

            count = q.count()
            response.headers["Content-Range"] = str(count)

            # sort column and order
            if q_sort:
                q_sort = sub('[\[\"\'\]]*', '', q_sort).split(',')
                q_order_by = q_sort[0]
                q_sort = q_sort[1]
                if q_sort == "DESC":
                    q = q.order_by(desc(text(q_order_by)))
                else:
                    q = q.order_by(text(q_order_by))

            # pagination
            if q_range:
                q_range = sub('[\[\]]*', '', q_range).split(',')
                q_skip = int(q_range[0])
                q_limit = int(q_range[1]) - int(q_range[0]) + 1
            else:
                q_skip, q_limit = pagination.get(
                    'skip'), pagination.get('limit')

            return q.offset(q_skip).limit(q_limit).all()

        return route

For filtering I sort of got one column filtering.
But not yet with multiple columns at once or with a tablewide search ("q" in react-admin).

'filter': | {"name":"Pencil"}

if q_filter:
    q_filter = loads(q_filter)
    if "q" in list(q_filter):
        del q_filter["q"]
    q = [q.filter(getattr(self.db_model, col).ilike(f"%%{val}%%"))
         for col, val in q_filter.items()]
    q = [item for item in q if item][0]

I have no doubts there's cleaner solutions than what I did.

[awtkns]

Thanks for taking the time to share! Its always cool to see how people are using the package. I am hoping to add native filter and sort support in the next big feature release. (Probably in early April).

Do let me know once you get it all working, It would be interesting to see them both working together!

[DorskFR]

This time working good for filtering across all columns, column by column, multi columns and array of values.

filter={}
filter={"customer_id":"47NX1F8TF"}
filter={"id":["47NX1HNUK","47NX1HP5V","47NX1HPX8"]}
filter={"email":"john@gmail.com","first_name":"johnny","last_name":"cash","search":"blue"}&range=[0,9]&sort=["id","ASC"]

def _get_all(self) -> Callable:
    def route(
        response: Response,
        db: Session = Depends(self.db_func),
        q_range: Optional[str] = Query(None, alias="range"),
        q_filter: Optional[str] = Query(None, alias="filter"),
        q_sort: Optional[str] = Query(None, alias="sort"),
        pagination: dict = self.pagination,
    ):

        q = db.query(self.db_model)

        q_filter = loads(q_filter)

        for key, value in q_filter.items():

            if key in ["q", "search"]:
                cols = [column.name for column in inspect(
                    self.db_model).c]
                q = q.filter(or_(
                    *[cast(getattr(self.db_model, col), String)
                        .ilike(f'%{value}%') for col in cols])
                )

            elif isinstance(value, list):
                q = q.filter(
                    cast(getattr(self.db_model, key), String)
                    .in_(value)
                )

            else:
                q = q.filter(
                    cast(getattr(self.db_model, key), String)
                    .ilike(f"%{value}%")
                )

        count = q.count()
        response.headers["Content-Range"] = str(count)

        # sort column and order
        if q_sort:
            q_sort = sub('[\[\"\'\]]*', '', q_sort).split(',')
            q_order_by = q_sort[0]
            q_sort = q_sort[1]
            if q_sort == "DESC":
                q = q.order_by(desc(text(q_order_by)))
            else:
                q = q.order_by(text(q_order_by))

        # pagination
        if q_range:
            q_range = sub('[\[\]]*', '', q_range).split(',')
            q_skip = int(q_range[0])
            q_limit = int(q_range[1]) - int(q_range[0]) + 1
        else:
            q_skip, q_limit = pagination.get(
                'skip'), pagination.get('limit')

        return q.offset(q_skip).limit(q_limit).all()

    return route

If that can be of some use.

Thanks again for Crudrouter, its making me gain a lot of time!

[awtkns]

Thanks man! Happy to hear it is working well 😄

[awtkns]

@DorskFR is this a PR you are still interested in? Otherwise I might start on the implementation.

[DorskFR]

Hello @awtkns ,

Thank you for your message.

My reasoning is the following:

If we have different resources and different users with different privilege levels.

1. Some resources can be CRUD by anyone:
   This works perfectly with the current implementation with dependencies that apply to all CRUD of a resource / router.

2. Some resources can be Read but should not be CUD by simple users.
   With the current implementation, from what I understood and did, this requires having an admin dependency on the CRUDRouter and then overloading the Read route at the App level (as discussed in another ticket) as otherwise the admin dependency will still be applied if overloading on the router.

3. Some resources can be CRU but should not be Deleted by users who are not admin.
   Same concept as 2. but easier to implement since the CRUDRouter can have the lower privilege dependency and only the overloaded route Delete has the extra dependency to be admin so it can be done by overloading at the router level.

4. We want to replace or add a dependency only for specific resources and specific routes.
   For example updating and deleting specific important resources might require checking for freshness in an auth token.

I think two things might be worth considering:

A- Is it intentional that an overloaded route still has applied to it the dependency of the CRUDRouter ?
It does make sense because this overloaded route is overloaded in its functional content but since we are overloading the router then it is still a child of it, so it makes sense it inherits the prefix and the dependencies.
However, since we overload it would also make sense that we can change all the parameters of it and therefore do not inherit anything.
But since we have the alternative to overload at the App level, we still have both options.
I mention this because for example in case 2. above, we would only need to overload one route with a lower privilege auth Dependency and it would not require this Dependency per route feature.

B- Does different dependencies per route makes sense in the design of the universal profile of the CRUDRouter or is this in the domain of custom, overloaded routes?

I mention both those possibilities because, in the situation where I use the normal implementation of the CRUDRouter but only want a specific Dependency on a route, I think B would be really useful and avoid creating overloaded routes only to add (router level) or remove a Dependency (App level).

However in the situation where I also want to do different things or trigger side effects in the CRUD routes then this different dependency feature could also be implemented directly in the overload functionality by disabling the routers default dependency and only considering the ones of the overloaded route.

I am not sure if the more general usecase would be extra or different dependencies while using the normal CRUD routes or almost always together with custom routes.

What I ended up doing is creating all my overloaded routes at the App level to prevent the dependency from the router to be applied and to handle my side effects for those situations where I have custom functions to handle the CRUD event.

For the other normal routes I added the request.method conditional test in my auth Dependency as mentioned above to disallow the Delete call if not admin.

While I do believe it would make the general readability easier and cleaner, I think the worth of implementing this depends on your opinion of A and B above and the general design you want to have for CRUDRouter.

In any case, for my current use case I find it already flawless in the current implementation and I worked my way around my specific needs doing what I described here.

Thank you!

[awtkns]

Thanks for the insightful feedback. Glad you hacked together an implementation on your end.

Is it intentional that an overloaded route still has applied to it the dependency of the CRUDRouter?

It is intentional that the dependencies stick with the overloaded router as this mimics how dependencies are used when added at the router level. As you mentioned above, it is possible to get around this at the app level.

Does different dependencies per route makes sense in the design of the universal profile of the CRUDRouter or is this in the domain of custom, overloaded routes?

There has been significant interest amongst other users in the implementation of this feature (eg: #47). Going forward I think the addition of this is well within the goals and motivations of this package of making CRUD routes easy and simple to implement.

users_router = SQLAlchemyCRUDRouter(
    ...
    dependencies=[Depends(token_auth)],
    update_route_dependencies=[Depends(must_be_owner)],
    delete_one_route_dependencies=[Depends(must_be_admin)],
    delete_all_route_dependencies=[Depends(must_be_admin)],
)

Based on all this, I think an ideal solution would be to provide an interface (that you mentioned) to add such dependencies. Thoughts? Is there a cleaner interface that you would prefer?

[DorskFR]

Hello,
Thank you for your answer.

I have been coding for just a few months so not sure I can come up with any clean or elegant suggestion.
Maybe it can be added directly to the conditionals you use for the routes?

if get_all_route:
if create_route:
if delete_all_route:
# etc.

So they would be either: True by default, same as now but with an optional list of Dependencies or False ?

[awtkns]

@DorskFR is this something you would be interested in Coding / PRing in the next couple weeks? If not, I would be happy to implement it.

[DorskFR]

Hello @awtkns ,

I never did a PR but I can give it a try in the coming week if you do not mind.
Worst case you can always refuse what I tried to code :)

[jm-moreau]

Based on all this, I think an ideal solution would be to provide an interface (that you mentioned) to add such dependencies. Thoughts? Is there a cleaner interface that you would prefer?

For my use case, I'd like to be able to access these dependencies from within a subclass of SQLAlchemyCRUDRouter. So maybe a dictionary of dependencies could work like in the example below?

Basically, for my application, I need to check that the current user has access rights for whatever they are requesting.

So I planned to do something like:

class CheckingSQACRUDRouter(SQLAlchemyCRUDRouter):
    def _get_one(self, *args, **kwargs):
        def route(item_id: self._pk_type, db: Session = Depends(self.db_func), dependencies):
            model: Model = db.query(self.db_model).get(item_id)

            if model:
                if model.customer_id != dependencies['user'].customer_id:
                  raise NO_ACCESS
                else:
                  return model
            else:
                raise NOT_FOUND

[awtkns]

So maybe a dictionary of dependencies could work like in the example below?

Yes, I think that would be a good idea to provide an interface to do that. That being said I am curious why one would choose to overload the route like that. It might be simpler to over load the route as shown below for your use case.

router = SQLAlchemyCRUDRouter(myModel, ...)

@router.get("/{item_id}", reponseModel=MyModel)
def custom_get_one_route(item_id: int, db: Session = Depends(self.db_func), user = Depends(UserDep))
   # Custom logic

That being said, your custom CheckingSQACRUDRouter would be more reusable throughout the application.

[jm-moreau]

It might be simpler to over load the route as shown below for your use case.

That's a nice clear way to do it.

I am curious why one would choose to overload the route like that

Probably there is a cleaner way to do this.... In my application nearly every route needs this kind of "do you have rights for that" checking. A subclass seemed DRYer. In my application, I am using inspection to investigate a models.py module with all the sqlalchemy ORM objects. After filtering out the "special case" classes, I just loop through and generate a few hundred routes using fastapi-crudrouter. Now, I'm thinking about how to add dependency support for this checking.

Would this be a good idea? Refactor SQLAlchemyCRUDRouter such that every instance of getting an item like:
model: Model = db.query(self.db_model).get(item_id) in _get_one() and
db_model: Model = self._get_one()(item_id, db) in _update()

got refactored to a _get_item() function who also got whichever dependencies passed along

Then others could just overload the getter if need be to add any logic related to the whatever Depends() returns.

[DorskFR]

Hello,

Sorry I got a bit caught up with a weird bug with Depends locking up sqlalchemy (discussed here : tiangolo/full-stack-fastapi-template#104).

On that note I had to modify my local copy of crudrouter to use a context manager instead of Depends for the database in the different routes.

Anyway, I tried to modify to pass the dependencies in the routes like this :

app.include_router(
    CRUDRouter(
        schema=Potato,
        get_all_route=[Depends(say_hello)],
        get_one_route=[Depends(say_hi)],
        delete_all_route=[Depends(say_hello), Depends(say_hi)],
    )
)

and in _base.py and all the router instances they are typed like this:

get_all_route: Union[bool, Sequence[params.Depends]] = True,

which seems to match what is expected in add_api_route

the condition is then tested like this:

if get_all_route:
    if isinstance(get_all_route, bool):
        dependencies = None
    else:
        dependencies = get_all_route
    super().add_api_route(
        "",
        self._get_all(),
        methods=["GET"],
        response_model=Optional[List[self.schema]],  # type: ignore
        summary="Get All",
        dependencies=dependencies,
    )

It works fine with my simple example.

INFO:     Application startup complete.
Hello
INFO:     127.0.0.1:58758 - "GET /potato?skip=0 HTTP/1.1" 200 OK
hi
INFO:     127.0.0.1:58760 - "GET /potato/0 HTTP/1.1" 404 Not Found
Hello
hi
INFO:     127.0.0.1:58762 - "DELETE /potato HTTP/1.1" 200 OK

I ran pytest, got the same result 361 passed, 58 warnings.
I am still trying to understand how to write tests and make a pull request but maybe it's not worth it since I'm late.

Let me know!
Thank you