how to have sqlalchemy mixins that need user id?

[shizidushu]

First check

• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.

Description

I see Flask-appbuilder's Mixins and want to implement the AuditMixin. Here is a part code of AuditMixin from Flask-appbuilder:

from flask import g
from sqlalchemy import Column, ForeignKey, Integer
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import relationship

class AuditMixin(object):
    @declared_attr
    def created_by_fk(cls):
        return Column(
            Integer, ForeignKey("fa_user.id"), default=cls.get_user_id, nullable=False
        )

    @declared_attr
    def created_by(cls):
        return relationship(
            "User",
            primaryjoin="%s.created_by_fk == User.id" % cls.__name__,
            enable_typechecks=False,
        )

    @classmethod
    def get_user_id(cls):
        try:
            return g.user.id
        except Exception:
            return None

It seems there is no object like g in FastAPI.

Is there a way to get this AuditMixin to work in FastAPI?

[maverick42]

Hi,

I'm looking to do exactly the same thing...
Another exemple (with flask) : https://gist.github.com/techniq/5174410

I can't get the current_user.id in my mixin ...

base_class.py

from sqlalchemy.ext.declarative import declarative_base, declared_attr

# for mixin
from datetime import datetime
from sqlalchemy import Column, Integer, DateTime, ForeignKey

class CustomBase(object):
    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()


Base = declarative_base(cls=CustomBase)

class AuditMixin(object):
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, onupdate=datetime.now)

    @declared_attr
    def created_by_id(cls):
        return Column(Integer,
            ForeignKey('mp.usr_user.id', name='fk_%s_created_by_id' % cls.__name__, use_alter=True),
            # nullable=False,
            default=_current_user_id_or_none
        )

    @declared_attr
    def updated_by_id(cls):
        return Column(Integer,
            ForeignKey('mp.usr_user.id', name='fk_%s_updated_by_id' % cls.__name__, use_alter=True),
            # nullable=False,
            onupdate=_current_user_id_or_none
        )        

def _current_user_id_or_none():
    #current_user: usrUser = Security(get_current_user)

    try:
        return 2 #current_user.id
    except:
        return None

db_model exemple :

from sqlalchemy import Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.orm import relationship

from app.db.base_class import Base, AuditMixin

DATABASE_SCHEMA = 'mp'
MODULE_NAME = 'cpg_'

class cpgCampaignCategorie(AuditMixin,Base):
    __tablename__  = MODULE_NAME + 'campaign_categorie'
    __table_args__ = ({ "schema": DATABASE_SCHEMA})

    id = Column(Integer, primary_key=True, index=True)
    description = Column(String, index=True)

Thanks

[shizidushu]

Maybe we could use monkey patching. This is to say: Leave the function get_user_id not implemented, and implement it in crud part and patch it to the model class.
I'll try to implement it after a few days and put code here if I'm lucky.

[tiangolo]

I have never used Flask-appbuilder, but from your code, I see it uses g, the global thread local proxy object from Flask.

I imagine Flask-appbuilder sets the "current user" in it, but I don't know how. It probably reads authentication details from a cookie, a JWT token, or some other specific method, then it probably verifies it, extracts some ID, and gets the user record from the DB with that ID. Or something similar.

FastAPI doesn't make any compromise on which DB, ORM, nor specific authentication method you use, but you can integrate anything you want.

In FastAPI there are no global proxy objects, you access the data directly. To get the current user you would probably read the cookie, JWT, or method you chose in a dependency, then verify it, and then get the user from the DB using those credentials. You could probably do all that in dependencies. And that way the data that you require (e.g. JWT tokens using OAuth2) will be integrated into OpenAPI and the docs UI.

To read more about a similar example, check the docs here for SQL DBs with SQLAlchemy: https://fastapi.tiangolo.com/tutorial/sql-databases/

And the docs for dependencies and auth stuff here: https://fastapi.tiangolo.com/tutorial/security/get-current-user/

You would probably take that out of your custom class and put it in a simple function to be used as a dependency. And then you would get the current user with it.

[maverick42]

Hi,

I had this problem with my Audit Mixin (get current user) and the TranslationHybrid plugin from SqlAlchemy-utils (get current locale). As I wrote here I solved both problems with the help of a middleware. It works perfectly in both cases. For the current user, in the middleware, I decode my JWT to extract the user_id.
I added in my Postgresql base model (on which my tables depend) a user_id property which is set by the middleware. So I have now the current user in my audit mixin.

Regards

[shizidushu]

@tiangolo Thanks for your reply. I didn't realize I can access the parameter of request with the method in a class. So I need to have a parameter with the same name like current_user in every request when I need to create a new instance of the audited class. Or set a variable like request.state.current_user inside a middleware as @maverick42 suggests.

[l2m2]

I am looking for the same thing too.
Could you give some examples for "I added in my Postgresql base model (on which my tables depend) a user_id property which is set by the middleware. ". How to set it by the middleware?

[wangxin688]

you can use contextvars to get user_id.
like the following:

class AuditUserMixin:
    """
    AuditUserMixin
    mixin for models, add create_user, user_id
    """

    @declared_attr
    def created_by_fk(cls) -> Mapped[uuid.UUID]:
        return mapped_column(GUID, ForeignKey("auth_user.id"), default=auth_user_ctx.get, nullable=False)

    @declared_attr
    def created_by(cls) -> Mapped["User"]:
        return relationship(
            "User",
            foreign_keys=[cls.created_by_fk],
            primaryjoin="%s.created_by_fk==User.id" % cls.__name__,
            enable_typechecks=False,
        )

    @declared_attr
    def updated_by_fk(cls) -> Mapped[Optional[uuid.UUID]]:
        return mapped_column(GUID, ForeignKey("auth_user.id"), default=auth_user_ctx.get, nullable=True)

    @declared_attr
    def updated_by(cls) -> Mapped["User"]:
        return relationship(
            "User",
            foreign_keys=[cls.updated_by_fk],
            primaryjoin="%s.updated_by_fk==User.id" % cls.__name__,
            enable_typechecks=False,
        )

and deploy the contextvar in get_current_user
auth_user_ctx.set(user.id)