FastAPI + SQLAlchemy dependency user_have_permission

Declare user_have_permission dependency for validating permissions

In this topic we will use at first fake database and very modest example of user_have_permission dependency for FastAPI, then write full example, that covers SQLAlchemy integration and more realistic flows.

# main.py

import random

from fastapi import Depends, FastAPI, HTTPException

from respo import RespoClient

from .respo_model import RespoModel

RESPO_MODEL = RespoModel.get_respo_model()

respo_client_admin = RespoClient()
respo_client_admin.add_role(RESPO_MODEL.ROLES.ADMIN, RESPO_MODEL)
respo_client_no_roles = RespoClient()


fake_users = [
    {"name": "Peter", "respo_field": respo_client_admin},
    {"name": "Sara", "respo_field": respo_client_no_roles},
]


app = FastAPI()


def user_have_permission(permission):
    def inner_user_have_permission():
        # normally we would get user by headers or param etc from
        # another dependency, here we use random user from list
        user = random.choice(fake_users)

        respo_client: RespoClient = user["respo_field"]
        if not respo_client.has_permission(permission, RESPO_MODEL):
            raise HTTPException(403)
        return user

    return inner_user_have_permission


@app.get("/")
def get_user(
    user=Depends(user_have_permission(RESPO_MODEL.PERMS.USER__READ_ALL)),
):
    return {"name": user["name"]}

Run FastAPI app

Put above code in main.py and run

uvicorn main:app --reload

At the address localhost:8000/docs you can test the endpoint.

Click on Execute few times in a row to see expected result.

Expected result

Now, in user_have_permission dependency we check using RESPO_MODEL declared in previous chapters if a user have specific permission (or with more precision if a user have at least one role that includes this permission). Since we have only 2 users and we get them by random choice from list, sometimes we get 200 with

{
  "name": "Peter"
}

when Peter is choosed and sometimes 403:

{
  "detail": "Forbidden"
}

when Sara that has no permission.

Declare user_have_permission dependency with SQLAlchemy user table

# main.py

import random

from fastapi import Depends, FastAPI, HTTPException
import asyncio
from respo import RespoClient
from dataclasses import dataclass, field
from .respo_model import RespoModel
from sqlalchemy import Column, Integer, String, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import registry
from sqlalchemy.orm.session import sessionmaker


from respo import RespoClient
from respo.fields.sqlalchemy import SQLAlchemyRespoField
from sqlalchemy.sql import func

RESPO_MODEL = RespoModel.get_respo_model()


async_engine = create_async_engine("sqlite+aiosqlite:///", pool_pre_ping=True)
async_session = sessionmaker(
    async_engine, expire_on_commit=False, class_=AsyncSession  # type: ignore
)
mapper_registry = registry()


@mapper_registry.mapped
@dataclass
class User:
    __tablename__ = "user_model"
    __sa_dataclass_metadata_key__ = "sa"

    id: int = field(
        init=False,
        metadata={"sa": Column(Integer, primary_key=True)},
    )
    name: str = field(
        metadata={"sa": Column(String(254), nullable=False, unique=True, index=True)}
    )
    respo_field: RespoClient = field(
        default_factory=RespoClient,
        metadata={
            "sa": Column(SQLAlchemyRespoField, nullable=False, server_default="")
        },
    )


app = FastAPI()


async def db_setup():
    async with async_engine.begin() as conn:
        await conn.run_sync(mapper_registry.metadata.drop_all)
        await conn.run_sync(mapper_registry.metadata.create_all)

    async with async_session() as session:
        session: AsyncSession
        respo_client_admin = RespoClient()
        respo_client_admin.add_role(RESPO_MODEL.ROLES.ADMIN, RESPO_MODEL)
        respo_client_no_roles = RespoClient()

        user_peter = User("Peter", respo_client_admin)
        user_sara = User("Sara", respo_client_no_roles)

        session.add(user_peter)
        session.add(user_sara)
        await session.commit()


def user_have_permission(permission):
    async def inner_user_have_permission(db_setup=Depends(db_setup)):
        # normally we would get user by access_token from header or
        # another dependency, here we get random user from database
        async with async_session() as session:
            session: AsyncSession
            user: User = (
                (await session.execute(select(User).order_by(func.random()).limit(1)))
                .scalars()
                .one()
            )

        if not user.respo_field.has_permission(permission, RESPO_MODEL):
            raise HTTPException(403)
        return user

    return inner_user_have_permission


@app.get("/")
def get_user(
    user: User = Depends(user_have_permission(RESPO_MODEL.PERMS.USER__READ_ALL)),
):
    return {"name": user.name}

Using SQLAlchemy RespoField

In SQLAlchemy user model we use custom field powered by respo: SQLAlchemyRespoField. It is responsible for serialization and deserialization from respo.RespoClient to string in database back and forth. So we write:

respo_field: RespoClient = field(
    default_factory=RespoClient,
    metadata={
        "sa": Column(SQLAlchemyRespoField, nullable=False, server_default="")
    },
)
# or without extra typing support
respo_field = Column(SQLAlchemyRespoField, nullable=False, server_default="")

Note

Note, this RespoClient-like field is mutable, so any direct changes to attributes won't trigger SQLAlchemy machinery and changes won't be saved to database. The only methods that trigger so called has_changed() event are add_role and remove_role!

Scaling

After above setup scaling is very easy using the same dependency factory for any new endpoint (with different permission). For example we can continue with:

@app.get("/buy-book")
def buy_books(
    user=Depends(user_have_permission(RESPO_MODEL.PERMS.BOOK__BUY)),
):
    return None


@app.get("/sell-book")
def sell_books(
    user=Depends(user_have_permission(RESPO_MODEL.PERMS.BOOK__SELL)),
):
    return None

Recap

In this section, we do basicaly two things:

• create field respo_field based on custom field SQLAlchemyRespoField in our User model in database, so we can store user's roles in database
• create user_have_permission dependency factory that is resposible for additional checking if user has permission to requested resource. Note, it may differ in certain scenarios and use cases, but basically user should have access either when is owner/creator of resource whatever that means, or has role that allow it.