How can I add added_by and updated_by field based on the session running in Piccolo Table?

[sumitsharansatsangi]

I have a table as follows.

class Band(Table):
    id = Varchar(length=8, index=True, primary_key=True, default=get_id, index_method=IndexMethod.btree)
    name = Varchar(unique=True)
    added_by = ForeignKey(references=User)
    updated_by = ForeignKey(references=User)
    added_at = Timestamptz(default=TimestamptzNow(), null=True, primary_key=False, unique=False, index=False,
                           index_method=IndexMethod.btree, db_column_name=None, secret=False)
    updated_at = Timestamptz(auto_update=datetime.now, null=True, primary_key=False, unique=False, index=False,
                             index_method=IndexMethod.btree, db_column_name=None, secret=False)

How can the added_by and updated_by field is to be filled according to the user updating the field?
These are the fields which cannot be updated by the user (manually) updating the record, but to be filled based on the session running.

[sinisaos]

@sumitsharansatsangi You can do something like this to get the current session admin user from SessionBase and then set this value as the auto_update for the FK column like this. The user must be a superuser!

from piccolo_api.session_auth.tables import SessionsBase

def current_session_admin_user():
    data = (
        SessionsBase.select(SessionsBase.user_id)
        .first()
        .order_by(SessionsBase.user_id, ascending=False)
        .run_sync()
    )
    return data["user_id"]

class Band(Table):
    name = Varchar(unique=True)
    added_by = ForeignKey(
        auto_update=current_session_admin_user,
        references=BaseUser,
    )
    updated_by = ForeignKey(
        auto_update=current_session_admin_user,
        references=BaseUser,
    )

Additionally, whenever a record is updated, the current admin user will be set to the added_byand updated_by columns.
I'm sorry if I didn't correctly understand what you mean.

[sumitsharansatsangi]

Let me try this, @sinisaos .

[sumitsharansatsangi]

@sinisaos , added_by column, won't be updated when the record is going to be updated, only updated_by will be updated.

[sinisaos]

@sumitsharansatsangi Strange. In my case, both columns are updated with the session admin user of as in the video.

admin.mp4

[sumitsharansatsangi]

@sinisaos, I know that in your case it will be updating both column, but we don't have to update both column.
So, for my case this is probably the right answer.

from piccolo_api.session_auth.tables import SessionsBase

def current_session_admin_user():
    data = (
        SessionsBase.select(SessionsBase.user_id)
        .first()
        .order_by(SessionsBase.user_id, ascending=False)
        .run_sync()
    )
    return data["user_id"]

class Band(Table):
    name = Varchar(unique=True)
    added_by = ForeignKey(
        default=current_session_admin_user,  # notice the change here
        references=BaseUser,
    )
    updated_by = ForeignKey(
        auto_update=current_session_admin_user,
        references=BaseUser,
    )

But the change I have done i.e. default=current_session_admin_user, is not working properly. I just tried to mimic added_at column, Can you please suggest a answer for added_by(It will be added once in the lifetime of the record and won't be updated on the updation of the record)

[sinisaos]

@sumitsharansatsangi Sorry, but your initial question was this

How can the added_by and updated_by field is to be filled according to the user updating the field?

and the suggested solution was based on that. You cannot set the default as you have tried. You can use Piccolo pre_save hooks and this will set the default session user before saving the records to the database. Here is the entire script that you can adapt to your needs.

import asyncio

import uvicorn
from fastapi import FastAPI
from piccolo.apps.user.tables import BaseUser
from piccolo.columns import ForeignKey, Varchar
from piccolo.engine.sqlite import SQLiteEngine
from piccolo.table import Table
from piccolo_admin.endpoints import TableConfig, create_admin
from piccolo_api.crud.hooks import Hook, HookType
from piccolo_api.session_auth.tables import SessionsBase
from starlette.routing import Mount

DB = SQLiteEngine()


# used for setting admin user for auto_update
def current_session_admin_user():
    data = (
        SessionsBase.select(SessionsBase.user_id)
        .first()
        .order_by(SessionsBase.user_id, ascending=False)
        .run_sync()
    )
    return data["user_id"]


class Band(Table, db=DB):
    name = Varchar(unique=True)
    added_by = ForeignKey(
        references=BaseUser,
    )
    updated_by = ForeignKey(
        auto_update=current_session_admin_user,
        references=BaseUser,
    )


# used to set the default admin user through hooks
def default_session_user(row: Band):
    row.added_by = current_session_admin_user()
    return row


band_config = TableConfig(
    Band,
    hooks=[Hook(hook_type=HookType.pre_save, callable=default_session_user)],
)


# FastAPI app instantiation and mounting admin
app = FastAPI(
    routes=[
        Mount(
            "/admin/",
            create_admin(
                tables=[band_config, BaseUser],
            ),
        )
    ],
)


async def main():
    # Tables creating
    await BaseUser.create_table(if_not_exists=True)
    await SessionsBase.create_table(if_not_exists=True)
    await Band.create_table(if_not_exists=True)

    # Creating admin users
    if not await BaseUser.exists().where(BaseUser.email == "admin@test.com"):
        user = BaseUser(
            username="piccolo",
            password="piccolo123",
            email="admin@test.com",
            admin=True,
            active=True,
            superuser=True,
        )
        await user.save()

    if not await BaseUser.exists().where(BaseUser.email == "john@test.com"):
        user = BaseUser(
            username="john",
            password="john123",
            email="john@test.com",
            admin=True,
            active=True,
            superuser=True,
        )
        await user.save()


if __name__ == "__main__":
    asyncio.run(main())

    uvicorn.run(app, host="127.0.0.1", port=8000)

and here is a video that shows that everything works as it should.

default_user.mp4