initial commit for audit logs

docs/source/audit_logs/commands.rst

@@ -0,0 +1,16 @@
+Commands
+========
+
+If you've registered the ``audit_logs`` app in your ``piccolo_conf.py`` file
+(see the :ref:`migrations docs <AuditLogMigrations>`), it gives you access to a
+custom command.
+
+clean
+-----
+
+If you run the following on the command line, it will delete all logs
+from the database.
+
+.. code-block:: bash
+
+    piccolo audit_logs clean

docs/source/audit_logs/tables.rst

@@ -0,0 +1,63 @@
+Tables
+======
+
+``audit_logs`` is a ``Piccolo`` app that records changes made by users to database tables.
+We store the audit logs in :class:`AuditLog <piccolo_api.audit_logs.tables.AuditLog>`.
+
+-------------------------------------------------------------------------------
+
+.. _AuditLogMigrations:
+
+Migrations
+----------
+
+We recommend creating the ``audit_logs`` tables using migrations.
+
+You can add ``piccolo_api.audit_logs.piccolo_app`` to the ``apps`` arguments
+of the :class:`AppRegistry <piccolo.conf.apps.AppRegistry>` in ``piccolo_conf.py``.
+
+.. code-block:: bash
+
+    APP_REGISTRY = AppRegistry(
+        apps=[
+            ...
+            "piccolo_api.audit_logs.piccolo_app",
+            ...
+        ]
+    )
+
+To learn more about Piccolo apps, see the `Piccolo docs <https://piccolo-orm.readthedocs.io/en/latest/piccolo/projects_and_apps/index.html>`_.
+
+To run the migrations and create the table, run:
+
+.. code-block:: bash
+
+    piccolo migrations forwards audit_logs
+
+-------------------------------------------------------------------------------
+
+Creating them manually
+----------------------
+
+If you prefer not to use migrations, and want to create them manually, you can
+do this instead:
+
+.. code-block:: python
+
+    from piccolo_api.audit_logs.tables import AuditLog
+    from piccolo.tables import create_db_tables_sync
+
+    create_db_tables_sync(AuditLog, if_not_exists=True)
+
+-------------------------------------------------------------------------------
+
+Source
+------
+
+AuditLog
+~~~~~~~~
+
+.. currentmodule:: piccolo_api.audit_logs.tables
+
+.. autoclass:: AuditLog
+    :members:

docs/source/index.rst

@@ -40,6 +40,13 @@ ASGI app, covering authentication, security, and more.
    ./change_password/index
    ./advanced_auth/index
 
+.. toctree::
+   :caption: Audit logs
+   :maxdepth: 1
+
+   ./audit_logs/tables.rst
+   ./audit_logs/commands.rst
+
 .. toctree::
    :caption: Contributing
    :maxdepth: 1

piccolo_api/audit_logs/commands.py

@@ -0,0 +1,10 @@
+from .tables import AuditLog
+
+
+async def clean():
+    """
+    Removes all audit logs.
+    """
+    print("Removing audit logs ...")
+    await AuditLog.delete(force=True).run()
+    print("Successfully removed audit logs")

piccolo_api/audit_logs/piccolo_app.py

@@ -0,0 +1,24 @@
+"""
+Import all of the Tables subclasses in your app here, and register them with
+the APP_CONFIG.
+"""
+
+import os
+
+from piccolo.conf.apps import AppConfig
+
+from .commands import clean
+from .tables import AuditLog
+
+CURRENT_DIRECTORY = os.path.dirname(os.path.abspath(__file__))
+
+
+APP_CONFIG = AppConfig(
+    app_name="audit_logs",
+    migrations_folder_path=os.path.join(
+        CURRENT_DIRECTORY, "piccolo_migrations"
+    ),
+    table_classes=[AuditLog],
+    migration_dependencies=[],
+    commands=[clean],
+)

piccolo_api/audit_logs/piccolo_migrations/2022-06-25T17-11-22-238052.py

@@ -0,0 +1,129 @@
+from enum import Enum
+
+from piccolo.apps.migrations.auto.migration_manager import MigrationManager
+from piccolo.columns.column_types import JSON, Text, Timestamp, Varchar
+from piccolo.columns.defaults.timestamp import TimestampNow
+from piccolo.columns.indexes import IndexMethod
+
+ID = "2022-06-26T22:48:09:433352"
+VERSION = "0.80.0"
+DESCRIPTION = ""
+
+
+async def forwards():
+    manager = MigrationManager(
+        migration_id=ID, app_name="audit_logs", description=DESCRIPTION
+    )
+
+    manager.add_table("AuditLog", tablename="audit_log")
+
+    manager.add_column(
+        table_class_name="AuditLog",
+        tablename="audit_log",
+        column_name="action_time",
+        db_column_name="action_time",
+        column_class_name="Timestamp",
+        column_class=Timestamp,
+        params={
+            "default": TimestampNow(),
+            "null": False,
+            "primary_key": False,
+            "unique": False,
+            "index": False,
+            "index_method": IndexMethod.btree,
+            "choices": None,
+            "db_column_name": None,
+            "secret": False,
+        },
+    )
+
+    manager.add_column(
+        table_class_name="AuditLog",
+        tablename="audit_log",
+        column_name="action_type",
+        db_column_name="action_type",
+        column_class_name="Varchar",
+        column_class=Varchar,
+        params={
+            "length": 255,
+            "default": "",
+            "null": False,
+            "primary_key": False,
+            "unique": False,
+            "index": False,
+            "index_method": IndexMethod.btree,
+            "choices": Enum(
+                "ActionType",
+                {
+                    "CREATING": "CREATING",
+                    "UPDATING": "UPDATING",
+                    "DELETING": "DELETING",
+                },
+            ),
+            "db_column_name": None,
+            "secret": False,
+        },
+    )
+
+    manager.add_column(
+        table_class_name="AuditLog",
+        tablename="audit_log",
+        column_name="action_user",
+        db_column_name="action_user",
+        column_class_name="Varchar",
+        column_class=Varchar,
+        params={
+            "length": 255,
+            "default": "",
+            "null": False,
+            "primary_key": False,
+            "unique": False,
+            "index": False,
+            "index_method": IndexMethod.btree,
+            "choices": None,
+            "db_column_name": None,
+            "secret": False,
+        },
+    )
+
+    manager.add_column(
+        table_class_name="AuditLog",
+        tablename="audit_log",
+        column_name="change_message",
+        db_column_name="change_message",
+        column_class_name="Text",
+        column_class=Text,
+        params={
+            "default": "",
+            "null": False,
+            "primary_key": False,
+            "unique": False,
+            "index": False,
+            "index_method": IndexMethod.btree,
+            "choices": None,
+            "db_column_name": None,
+            "secret": False,
+        },
+    )
+
+    manager.add_column(
+        table_class_name="AuditLog",
+        tablename="audit_log",
+        column_name="changes_in_row",
+        db_column_name="changes_in_row",
+        column_class_name="JSON",
+        column_class=JSON,
+        params={
+            "default": "{}",
+            "null": False,
+            "primary_key": False,
+            "unique": False,
+            "index": False,
+            "index_method": IndexMethod.btree,
+            "choices": None,
+            "db_column_name": None,
+            "secret": False,
+        },
+    )
+
+    return manager

piccolo_api/audit_logs/tables.py

@@ -0,0 +1,118 @@
+import typing as t
+import uuid
+from enum import Enum
+
+from piccolo.apps.user.tables import BaseUser
+from piccolo.columns import JSON, Text, Timestamp, Varchar
+from piccolo.table import Table
+
+
+class AuditLog(Table):
+    class ActionType(str, Enum):
+        """An enumeration of AuditLog table actions type."""
+
+        CREATING = "CREATING"
+        UPDATING = "UPDATING"
+        DELETING = "DELETING"
+
+    action_time = Timestamp()
+    action_type = Varchar(choices=ActionType)
+    action_user = Varchar()
+    change_message = Text()
+    changes_in_row = JSON()
+
+    @classmethod
+    async def record_save_action(
+        cls,
+        table: t.Type[Table],
+        user_id: int,
+        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 user_id:
+            The ``primary key`` of authenticated user.
+        """
+        result = cls(
+            action_type=cls.ActionType.CREATING,
+            action_user=cls.get_user_username(user_id),
+            change_message=f"User {cls.get_user_username(user_id)} "
+            f"create row {new_row_id} in {table._meta.tablename.title()} "
+            f"table",
+        )
+        await result.save().run()
+
+    @classmethod
+    async def record_patch_action(
+        cls,
+        table: t.Type[Table],
+        row_id: t.Union[str, uuid.UUID, int],
+        user_id: 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 user_id:
+            The ``primary key`` of authenticated user.
+        :param changes_in_row:
+            JSON with all changed columns in the existing row.
+        """
+        result = cls(
+            action_type=cls.ActionType.UPDATING,
+            action_user=cls.get_user_username(user_id),
+            change_message=f"User {cls.get_user_username(user_id)} 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],
+        user_id: 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.
+        :param user_id:
+            The ``primary key`` of authenticated user.
+        """
+        result = cls(
+            action_type=cls.ActionType.DELETING,
+            action_user=cls.get_user_username(user_id),
+            change_message=f"User {cls.get_user_username(user_id)} delete row "
+            f"{row_id} in {table._meta.tablename.title()} table",
+        )
+        await result.save().run()
+
+    @classmethod
+    def get_user_username(cls, user_id: int) -> str:
+        """
+        Returns the username of authenticated user.
+
+        :param user_id:
+            The ``primary key`` of authenticated user.
+        """
+        user = (
+            BaseUser.select(BaseUser.username)
+            .where(BaseUser._meta.primary_key == user_id)
+            .first()
+            .run_sync()
+        )
+        return user["username"]