Add database auditability

src/api/qualicharge/api/v1/routers/static.py

@@ -210,7 +210,7 @@ async def update(
 
     transaction = session.begin_nested()
     try:
-        update = update_statique(session, id_pdc_itinerance, statique)
+        update = update_statique(session, id_pdc_itinerance, statique, author=user)
     except QCIntegrityError as err:
         transaction.rollback()
         raise HTTPException(
@@ -243,7 +243,7 @@ async def create(
 
     transaction = session.begin_nested()
     try:
-        db_statique = save_statique(session, statique)
+        db_statique = save_statique(session, statique, author=user)
     except ObjectDoesNotExist as err:
         transaction.rollback()
         raise HTTPException(
@@ -277,7 +277,7 @@ async def bulk(
     )
 
     transaction = session.begin_nested()
-    importer = StatiqueImporter(df, transaction.session.connection())
+    importer = StatiqueImporter(df, transaction.session.connection(), author=user)
     try:
         importer.save()
     except (

src/api/qualicharge/auth/schemas.py

@@ -9,7 +9,7 @@
 from sqlmodel import Field, Relationship, SQLModel
 
 from qualicharge.conf import settings
-from qualicharge.schemas import BaseTimestampedSQLModel
+from qualicharge.schemas.audit import BaseAuditableSQLModel
 from qualicharge.schemas.core import OperationalUnit
 
 
@@ -53,7 +53,7 @@ class ScopesEnum(StrEnum):
 
 
 # -- Core schemas
-class User(BaseTimestampedSQLModel, table=True):
+class User(BaseAuditableSQLModel, table=True):
     """QualiCharge User."""
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)
@@ -103,7 +103,7 @@ def check_password(self, password: str) -> bool:
         return settings.PASSWORD_CONTEXT.verify(password, self.password)
 
 
-class Group(BaseTimestampedSQLModel, table=True):
+class Group(BaseAuditableSQLModel, table=True):
     """QualiCharge Group."""
 
     id: UUID = Field(default_factory=uuid4, primary_key=True)

src/api/qualicharge/factories/__init__.py

@@ -32,5 +32,7 @@ class TimestampedSQLModelFactory(Generic[T], SQLAlchemyFactory[T]):
     __is_base_factory__ = True
 
     id = Use(uuid4)
+    created_by_id = None
+    updated_by_id = None
     created_at = Use(lambda: datetime.now(timezone.utc) - timedelta(hours=1))
     updated_at = Use(datetime.now, timezone.utc)

src/api/qualicharge/fixtures/operational_units.py

@@ -15,12 +15,13 @@
 """
 
 from collections import namedtuple
+from typing import List
 
 from qualicharge.schemas.core import OperationalUnit, OperationalUnitTypeEnum
 
 # Operational units
 Item = namedtuple("Item", ["code", "name"])
-data = [
+data: List[Item] = [
     Item(
         "FR073",
         "ACELEC CHARGE",

src/api/qualicharge/migrations/env.py

@@ -15,6 +15,7 @@
     Group,
     GroupOperationalUnit,
 )
+from qualicharge.schemas.audit import Audit
 from qualicharge.schemas.core import (  # noqa: F401
     Amenageur,
     Enseigne,

src/api/qualicharge/migrations/versions/86b08ec6e6d1_add_auditability.py

@@ -0,0 +1,268 @@
+"""add auditability
+
+Revision ID: 86b08ec6e6d1
+Revises: c09664a85912
+Create Date: 2024-10-16 17:03:58.431420
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+# revision identifiers, used by Alembic.
+revision: str = "86b08ec6e6d1"
+down_revision: Union[str, None] = "c09664a85912"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "audit",
+        sa.Column("id", sa.Uuid(), nullable=False),
+        sa.Column("table", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("author_id", sa.Uuid(), nullable=False),
+        sa.Column("target_id", sa.Uuid(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("changes", sa.JSON(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["author_id"],
+            ["user.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.add_column("amenageur", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("amenageur", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "amenageur_created_by_id_fkey", "amenageur", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "amenageur_updated_by_id_fkey", "amenageur", "user", ["updated_by_id"], ["id"]
+    )
+    op.add_column("city", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("city", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "city_created_by_id_fkey", "city", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "city_updated_by_id_fkey", "city", "user", ["updated_by_id"], ["id"]
+    )
+    op.add_column("department", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("department", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "department_created_by_id_fkey", "department", "user", ["updated_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "department_updated_by_id_fkey", "department", "user", ["created_by_id"], ["id"]
+    )
+    op.add_column("enseigne", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("enseigne", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "enseigne_created_by_id_fkey", "enseigne", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "enseigne_updated_by_id_fkey", "enseigne", "user", ["updated_by_id"], ["id"]
+    )
+    op.add_column("epci", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("epci", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "epci_created_by_id_fkey", "epci", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "epci_updated_by_id_fkey", "epci", "user", ["updated_by_id"], ["id"]
+    )
+    op.add_column("group", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("group", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "group_created_by_id_fkey", "group", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "group_updated_by_id_fkey", "group", "user", ["updated_by_id"], ["id"]
+    )
+    op.add_column("localisation", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("localisation", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "localisation_created_by_id_fkey",
+        "localisation",
+        "user",
+        ["updated_by_id"],
+        ["id"],
+    )
+    op.create_foreign_key(
+        "localisation_updated_by_id_fkey",
+        "localisation",
+        "user",
+        ["created_by_id"],
+        ["id"],
+    )
+    op.add_column("operateur", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("operateur", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "operateur_created_by_id_fkey", "operateur", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "operateur_updated_by_id_fkey", "operateur", "user", ["updated_by_id"], ["id"]
+    )
+    op.add_column(
+        "operationalunit", sa.Column("created_by_id", sa.Uuid(), nullable=True)
+    )
+    op.add_column(
+        "operationalunit", sa.Column("updated_by_id", sa.Uuid(), nullable=True)
+    )
+    op.create_foreign_key(
+        "operationalunit_created_by_id_fkey",
+        "operationalunit",
+        "user",
+        ["updated_by_id"],
+        ["id"],
+    )
+    op.create_foreign_key(
+        "operationalunit_updated_by_id_fkey",
+        "operationalunit",
+        "user",
+        ["created_by_id"],
+        ["id"],
+    )
+    op.add_column("pointdecharge", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("pointdecharge", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "pointdecharge_created_by_id_fkey",
+        "pointdecharge",
+        "user",
+        ["created_by_id"],
+        ["id"],
+    )
+    op.create_foreign_key(
+        "pointdecharge_updated_by_id_fkey",
+        "pointdecharge",
+        "user",
+        ["updated_by_id"],
+        ["id"],
+    )
+    op.add_column("region", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("region", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "region_created_by_id_fkey", "region", "user", ["updated_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "region_updated_by_id_fkey", "region", "user", ["created_by_id"], ["id"]
+    )
+    op.add_column("session", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("session", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "session_created_by_id_fkey", "session", "user", ["updated_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "session_updated_by_id_fkey", "session", "user", ["created_by_id"], ["id"]
+    )
+    op.add_column("station", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("station", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "station_created_by_id_fkey", "station", "user", ["updated_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "station_updated_by_id_fkey", "station", "user", ["created_by_id"], ["id"]
+    )
+    op.add_column("status", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("status", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "status_created_by_id_fkey", "status", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "status_updated_by_id_fkey", "status", "user", ["updated_by_id"], ["id"]
+    )
+    op.add_column("user", sa.Column("created_by_id", sa.Uuid(), nullable=True))
+    op.add_column("user", sa.Column("updated_by_id", sa.Uuid(), nullable=True))
+    op.create_foreign_key(
+        "user_created_by_id_fkey", "user", "user", ["created_by_id"], ["id"]
+    )
+    op.create_foreign_key(
+        "user_updated_by_id_fkey", "user", "user", ["updated_by_id"], ["id"]
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint("user_created_by_id_fkey", "user", type_="foreignkey")
+    op.drop_constraint("user_updated_by_id_fkey", "user", type_="foreignkey")
+    op.drop_column("user", "updated_by_id")
+    op.drop_column("user", "created_by_id")
+    op.drop_constraint("status_created_by_id_fkey", "status", type_="foreignkey")
+    op.drop_constraint("status_updated_by_id_fkey", "status", type_="foreignkey")
+    op.drop_column("status", "updated_by_id")
+    op.drop_column("status", "created_by_id")
+    op.drop_constraint("station_created_by_id_fkey", "station", type_="foreignkey")
+    op.drop_constraint("station_updated_by_id_fkey", "station", type_="foreignkey")
+    op.drop_column("station", "updated_by_id")
+    op.drop_column("station", "created_by_id")
+    op.drop_constraint("session_created_by_id_fkey", "session", type_="foreignkey")
+    op.drop_constraint("session_updated_by_id_fkey", "session", type_="foreignkey")
+    op.drop_column("session", "updated_by_id")
+    op.drop_column("session", "created_by_id")
+    op.drop_constraint("region_created_by_id_fkey", "region", type_="foreignkey")
+    op.drop_constraint("region_updated_by_id_fkey", "region", type_="foreignkey")
+    op.drop_column("region", "updated_by_id")
+    op.drop_column("region", "created_by_id")
+    op.drop_constraint(
+        "pointdecharge_created_by_id_fkey", "pointdecharge", type_="foreignkey"
+    )
+    op.drop_constraint(
+        "pointdecharge_updated_by_id_fkey", "pointdecharge", type_="foreignkey"
+    )
+    op.drop_column("pointdecharge", "updated_by_id")
+    op.drop_column("pointdecharge", "created_by_id")
+    op.drop_constraint(
+        "operationalunit_created_by_id_fkey", "operationalunit", type_="foreignkey"
+    )
+    op.drop_constraint(
+        "operationalunit_updated_by_id_fkey", "operationalunit", type_="foreignkey"
+    )
+    op.drop_column("operationalunit", "updated_by_id")
+    op.drop_column("operationalunit", "created_by_id")
+    op.drop_constraint("operateur_created_by_id_fkey", "operateur", type_="foreignkey")
+    op.drop_constraint("operateur_updated_by_id_fkey", "operateur", type_="foreignkey")
+    op.drop_column("operateur", "updated_by_id")
+    op.drop_column("operateur", "created_by_id")
+    op.drop_constraint(
+        "localisation_created_by_id_fkey", "localisation", type_="foreignkey"
+    )
+    op.drop_constraint(
+        "localisation_updated_by_id_fkey", "localisation", type_="foreignkey"
+    )
+    op.drop_column("localisation", "updated_by_id")
+    op.drop_column("localisation", "created_by_id")
+    op.drop_constraint("group_created_by_id_fkey", "group", type_="foreignkey")
+    op.drop_constraint("group_updated_by_id_fkey", "group", type_="foreignkey")
+    op.drop_column("group", "updated_by_id")
+    op.drop_column("group", "created_by_id")
+    op.drop_constraint("epci_created_by_id_fkey", "epci", type_="foreignkey")
+    op.drop_constraint("epci_updated_by_id_fkey", "epci", type_="foreignkey")
+    op.drop_column("epci", "updated_by_id")
+    op.drop_column("epci", "created_by_id")
+    op.drop_constraint("enseigne_created_by_id_fkey", "enseigne", type_="foreignkey")
+    op.drop_constraint("enseigne_updated_by_id_fkey", "enseigne", type_="foreignkey")
+    op.drop_column("enseigne", "updated_by_id")
+    op.drop_column("enseigne", "created_by_id")
+    op.drop_constraint(
+        "department_created_by_id_fkey", "department", type_="foreignkey"
+    )
+    op.drop_constraint(
+        "department_updated_by_id_fkey", "department", type_="foreignkey"
+    )
+    op.drop_column("department", "updated_by_id")
+    op.drop_column("department", "created_by_id")
+    op.drop_constraint("city_created_by_id_fkey", "city", type_="foreignkey")
+    op.drop_constraint("city_updated_by_id_fkey", "city", type_="foreignkey")
+    op.drop_column("city", "updated_by_id")
+    op.drop_column("city", "created_by_id")
+    op.drop_constraint("amenageur_created_by_id_fkey", "amenageur", type_="foreignkey")
+    op.drop_constraint("amenageur_updated_by_id_fkey", "amenageur", type_="foreignkey")
+    op.drop_column("amenageur", "updated_by_id")
+    op.drop_column("amenageur", "created_by_id")
+    op.drop_table("audit")
+    # ### end Alembic commands ###