Support for FastAPI dependencies

[heliumbrain]

Hi!

Working more with FastAPI makes it clear how big of a role the dependency injection system (https://fastapi.tiangolo.com/tutorial/dependencies/) plays. It would make a lot of sense to implement the support for this in Piccolo for a smooth experience. Afterall, FastAPI seems to be winning the ASGI framework race right now. The goal is to be able to do something like this:

app = FastAPI()

@app.get("/me/", response_model=BaseUserModel")
async def get_currently_authenticated_user(current_user: BaseUser = Depends(get_current_admin_user)):
    return current_user

Where get_current_admin_user would have to inspect the token and return the BaseUser object. I still haven't wrapped my around the dependcy injection system nor do I know where this should be implemented. Just want to put this here to kick off the discussion.

[heliumbrain]

See below for an example of where I'd like to use a dependency... I'm sure there's already parts of the piccolo ecosystem I could use for this, but I mostly did it like this to understand the flow. Please keep in mind that all of this is coming from someone who is not working as a programmer and is also VERY new to Python. Maybe this is really obvious to someone with more experience.

This is my first attempt at something that works exactly the way I want to:

@reciperouter.get("/me", response_model=t.List[RecipeModelOut])
async def get_users_recipes(conn: HTTPConnection):
    token = conn.cookies.get("session")
    if not token:
        return JSONResponse(
            status_code=401,
            content="You are not logged in."
        )
    s = SessionsBase
    user = (
        s.select(s.user_id)
        .where(s.token == conn.cookies.get("session"))
        .run_sync()
    )
    if not user:
        return JSONResponse(
            status_code=401,
            content="Your token is invalid. Please log in again.",
        )
    user_id = list(map(itemgetter("user_id"), user))
    r = Recipe
    return await r.select().where(r.author == user_id[0]).run()

I'm thinking that this whole part:

token = conn.cookies.get("session")
    if not token:
        return JSONResponse(
            status_code=401,
            content="You are not logged in."
        )
    s = SessionsBase
    user = (
        s.select(s.user_id)
        .where(s.token == conn.cookies.get("session"))
        .run_sync()
    )
    if not user:
        return JSONResponse(
            status_code=401,
            content="Your token is invalid. Please log in again.",
        )
    user_id = list(map(itemgetter("user_id"), user))

could just be separated and then added as a dependency.. But not sure where/how to put it.

[dantownsend]

I still need to wrap my head around the FastAPI dependencies mechanism.

Piccolo's SessionAuth middleware adds the user to the request (this is a fairly recent change - so make sure you have the most recent version of Piccolo API).

Here's an example endpoint from one of my apps:

@map_router.get("/map/node/", tags=["Node"])
async def get_nodes(request: Request) -> NodeListModel:
    nodes = (
        await Node.select().where(Node.creator.id == request.user.user.id).run()
    )

    return NodeListModel(nodes=nodes)

[heliumbrain]

Ok, nice! I'll try that, thanks.

Yeaah, I made some progress on the Dependency part but every time I start to understand it my head starts spinning and I have to take a nap. I'll share it here if I manage to come up with something that actually works.

[dantownsend]

I've improved the session auth docs for Piccolo API - they weren't very good before.

https://piccolo-api.readthedocs.io/en/latest/session_auth/index.html

Will try and improve them some more when I can.

[heliumbrain]

Brilliant! It's much clearer now

[heliumbrain]

This is a good intro in to how Dependencies in FastAPI work: https://www.jeffastor.com/blog/authentication-dependencies-in-fastapi

[sinisaos]

@heliumbrain Take a look here and you might find something interesting. I use Depends(get_current_user) to get topics and replies for authenticated user. This example uses OAuth2, but the same should work for Piccolo SessionAuth and you need to change get_current_user function to get the current user from token = conn.cookies.get ("session"). Sorry if I miss your point.

[heliumbrain]

Thanks @sinisaos! That looks like what I'm after.. I'll check it out a bit more later

[sinisaos]

@heliumbrain I'm happy if any of that can help you. You can try something like this (details is in new Piccolo API docs here)

from fastapi.requests import Request

def get_current_user(request: Request):
    auth_user = request.user.user
    return auth user

and after that you can try to use Depends(get_current_user). I haven’t tried it but I think it should work.
I also don’t work as a programmer but I really like Piccolo because it’s user-friendly and ecosystem is great. Also Daniel is a very friendly guy who is always ready to help and give advice.

[sinisaos]

@heliumbrain I tried this today and it works.

# all imports you need 

async def get_current_user(request: Request):
    token = request.cookies.get("session")
    if not token:
        return JSONResponse(status_code=401, content="You are not logged in.")
    s = SessionsBase
    user = (
        await s.select(s.user_id)
        .where(s.token == request.cookies.get("session"))
        .first()
        .run()
    )
    if not user:
        return JSONResponse(
            status_code=401,
            content="Your token is invalid. Please log in again.",
        )
    user_obj = (
        await BaseUser.select()
        .where(BaseUser.id == user["user_id"])
        .first()
        .run()
    )
    return user_obj


UserModelOut = create_pydantic_model(
    table=BaseUser,
    include_default_columns=True,
    model_name="UserModelOut",
)


@app.get("/profile/me/", response_model=UserModelOut, tags=["User profile"])
async def user_profile(
    current_user: UserModelOut = Depends(get_current_user),
):
    return current_user


@app.get("/profile/products/", tags=["User profile"])
async def user_products(
    current_user: UserModelOut = Depends(get_current_user),
):
    products = (
        await Product.select()
        .where(Product.author == current_user["id"])
        .order_by(Product.id, ascending=False)
        .run()
    )
    return [
        {
            "author": current_user["username"],
            "products": products,
        }
    ]

[heliumbrain]

@sinisaos that works great! good job and thanks a lot 👍