History Tracking

[AsfanUlla]

History tracking for changes made in table similar to django-simple-history

[dantownsend]

There isn't builtin support for this currently.

You could build something similar though.

Either using Piccolo Admin hooks:

https://piccolo-admin.readthedocs.io/en/latest/table_config/index.html?highlight=hooks#hooks

Or a Postgres row level trigger:

https://www.postgresql.org/docs/current/sql-createtrigger.html

With either of those approaches, you can detect when a row is updated, and insert the values into a history table.

[sinisaos]

@AsfanUlla It was a PR for audit logs, but it's closed. Here is a modified version of that. You can also see the changes in Piccolo Admin. I used FastAPI, but the same can be used with Litestar. A complete script that you can customize.

import asyncio
import typing as t
import uuid
from enum import Enum

import uvicorn
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from piccolo.apps.user.tables import BaseUser
from piccolo.columns import JSON, Boolean, Text, Timestamp, Varchar
from piccolo.columns.readable import Readable
from piccolo.table import Table
from piccolo.utils.pydantic import create_pydantic_model
from piccolo_admin.endpoints import create_admin
from piccolo_api.session_auth.tables import SessionsBase
from starlette.routing import Mount

from piccolo_conf import DB


# history log table
class HistoryLog(Table, db=DB):
    class ActionType(str, Enum):
        """An enumeration of HistoryLog table actions type."""

        creating = "creating"
        updating = "updating"
        deleting = "deleting"

    action_time = Timestamp()
    action_type = Varchar(choices=ActionType)
    table_name = Varchar()
    change_message = Text()
    changes_in_row = JSON()

    @classmethod
    async def record_save_action(
        cls,
        table: t.Type[Table],
        new_row_id=t.Union[str, uuid.UUID, int],
    ):
        """
        A method for tracking creating record actions.

        :param table:
            A table for which we monitor activities.
        :param new_row_id:
            The ``primary key`` of the newly created record.
        """
        result = cls(
            action_type=cls.ActionType.creating,
            table_name=table._meta.tablename.title(),
            change_message=f"Create row {new_row_id} in "
            f"{table._meta.tablename.title()} table",
        )
        await result.save().run()

    @classmethod
    async def record_patch_action(
        cls,
        table: t.Type[Table],
        row_id: t.Union[str, uuid.UUID, int],
        changes_in_row: t.Dict[str, t.Any],
    ):
        """
        A method for tracking updating record actions.

        :param table:
            A table for which we monitor activities.
        :param row_id:
            The ``primary key`` of the table for which we
            monitor activities.
        :param changes_in_row:
            JSON with all changed columns in the existing row.
        """
        result = cls(
            action_type=cls.ActionType.updating,
            table_name=table._meta.tablename.title(),
            change_message=f"Update row "
            f"{row_id} in {table._meta.tablename.title()} table",
            changes_in_row=changes_in_row,
        )
        await result.save().run()

    @classmethod
    async def record_delete_action(
        cls,
        table: t.Type[Table],
        row_id: t.Union[str, uuid.UUID, int],
    ):
        """
        A method for tracking deletion record actions.

        :param table:
            A table for which we monitor activities.
        :param row_id:
            The ``primary key`` of the table for which we
            monitor activities.
        """
        result = cls(
            action_type=cls.ActionType.deleting,
            table_name=table._meta.tablename.title(),
            change_message=f"Delete row "
            f"{row_id} in {table._meta.tablename.title()} table",
        )
        await result.save().run()


# return single row for PATCH request
def get_single_row(
    table: t.Type[Table],
    row_id: t.Union[str, uuid.UUID, int],
) -> t.Dict[str, t.Any]:
    """
    Return a single row.
    """
    row = table.select().where(table._meta.primary_key == row_id).first().run_sync()
    return row


# Tables example
class Task(Table, db=DB):
    name = Varchar()
    completed = Boolean(default=False)


TaskModelIn: t.Any = create_pydantic_model(
    table=Task,
    model_name="TaskModelIn",
)
TaskModelOut: t.Any = create_pydantic_model(
    table=Task,
    include_default_columns=True,
    model_name="TaskModelOut",
)
TaskModelPartial: t.Any = create_pydantic_model(
    table=Task,
    model_name="TaskModelPartial",
    all_optional=True,
)


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


@app.get("/tasks/", response_model=t.List[TaskModelOut])
async def tasks():
    return await Task.select().order_by(Task.id, ascending=False)


@app.post("/tasks/", response_model=TaskModelOut)
async def create_task(task_model: TaskModelIn):
    task = Task(**task_model.dict())
    await task.save()
    task = task.to_dict()
    # add created row in HistoryLog table
    await HistoryLog.record_save_action(
        table=Task,
        new_row_id=task["id"],
    )
    return task


@app.patch("/tasks/{task_id}/", response_model=TaskModelOut)
async def update_task(task_id: int, task_model: TaskModelPartial):
    changes_in_row: t.Dict[str : t.Any] = {}
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        return JSONResponse({}, status_code=404)

    for key, value in task_model.model_dump().items():
        if value is not None:
            changes_in_row[key] = value
            setattr(task, key, value)

    await task.save()
    task = task.to_dict()
    # add updated row in HistoryLog table
    new_row = get_single_row(Task, task["id"])
    await HistoryLog.record_patch_action(
        table=Task,
        row_id=task["id"],
        changes_in_row=changes_in_row,
    )

    return task


@app.delete("/tasks/{task_id}/")
async def delete_task(task_id: int):
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        return JSONResponse({}, status_code=404)

    # add deleted row in HistoryLog table
    await HistoryLog.record_delete_action(
        table=Task,
        row_id=task["id"],
    )

    await task.remove()

    return JSONResponse({})


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

    # Creating admin user
    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 __name__ == "__main__":
    asyncio.run(main())

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

Hope this helps.

[AsfanUlla]

Thanks, I tried with PostgreSQL triggers but then i even had to log the user_id of the logged in user so i think this might work.

[sinisaos]

Originally there was an action_user column (for an authenticated user in your app), but I removed that to make the example simpler.

[AsfanUlla]

cool, thanks this works