Unit of work 3: Transactional database: auto-commit or rollback per endpoint (#729)

Open a transaction per request, to ensure that everything will be either
committed or rollback.

## Migration strategy

Existing `db.commit()` were replaced by `db.flush()` even when it may
not have been necessary to prevent creating unexpected regressions.


## Advanced usage

The database transaction is automatically commited at the end.

- If an HTTPException is raised during the request, we consider that the
error was expected and managed by the endpoint. We commit the session.
 - If an other exception is raised, we rollback the session to avoid.

> Cruds and endpoints should never call `db.commit()` or `db.rollback()`
directly.
>After adding an object to the session, calling `await db.flush()` will
integrate the changes in the transaction without committing them.

> If an endpoint needs to add objects to the sessions that should be
committed even in case of an unexpected error, it should start a
SAVEPOINT after adding the object.
> ```python
> # Add here the object that should always be committed, even in case of
an unexpected error
> await db.add(object)
> await db.flush()
> # Start a SAVEPOINT. If the code in the following context manager
raises an exception, the changes will be rolled back to this point.
> async with db.begin_nested():
>     # Add objects that may be rolled back in case of an error here
> ```
---


Supersede #498

Author: armanddidierjean

app/core/auth/cruds_auth.py

@@ -3,7 +3,6 @@
 from datetime import UTC, datetime
 
 from sqlalchemy import delete, select, update
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.auth import models_auth
@@ -29,13 +28,8 @@ async def create_authorization_token(
     """Create a new group in database and return it"""
 
     db.add(db_authorization_code)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
-    else:
-        return db_authorization_code
+    await db.flush()
+    return db_authorization_code
 
 
 async def delete_authorization_token_by_token(
@@ -49,7 +43,7 @@ async def delete_authorization_token_by_token(
             models_auth.AuthorizationCode.code == code,
         ),
     )
-    await db.commit()
+    await db.flush()
     return None
 
 
@@ -71,13 +65,8 @@ async def create_refresh_token(
     """Create a new refresh token in database and return it"""
 
     db.add(db_refresh_token)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
-    else:
-        return db_refresh_token
+    await db.flush()
+    return db_refresh_token
 
 
 async def revoke_refresh_token_by_token(
@@ -94,7 +83,7 @@ async def revoke_refresh_token_by_token(
         )
         .values(revoked_on=datetime.now(UTC)),
     )
-    await db.commit()
+    await db.flush()
     return None
 
 
@@ -114,7 +103,7 @@ async def revoke_refresh_token_by_client_and_user_id(
         )
         .values(revoked_on=datetime.now(UTC)),
     )
-    await db.commit()
+    await db.flush()
 
 
 async def revoke_refresh_token_by_user_id(
@@ -131,4 +120,4 @@ async def revoke_refresh_token_by_user_id(
         )
         .values(revoked_on=datetime.now(UTC)),
     )
-    await db.commit()
+    await db.flush()

app/core/core_endpoints/cruds_core.py

@@ -1,7 +1,6 @@
 from collections.abc import Sequence
 
 from sqlalchemy import delete, select
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.core_endpoints import models_core
@@ -87,11 +86,7 @@ async def create_module_group_visibility(
     """Create a new module visibility in database and return it"""
 
     db.add(module_visibility)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
+    await db.flush()
 
 
 async def create_module_account_type_visibility(
@@ -101,11 +96,7 @@ async def create_module_account_type_visibility(
     """Create a new module visibility in database and return it"""
 
     db.add(module_visibility)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
+    await db.flush()
 
 
 async def delete_module_group_visibility(
@@ -119,7 +110,7 @@ async def delete_module_group_visibility(
             models_core.ModuleGroupVisibility.allowed_group_id == allowed_group_id,
         ),
     )
-    await db.commit()
+    await db.flush()
 
 
 async def delete_module_account_type_visibility(
@@ -134,7 +125,7 @@ async def delete_module_account_type_visibility(
             == allowed_account_type,
         ),
     )
-    await db.commit()
+    await db.flush()
 
 
 async def get_core_data_crud(
@@ -164,13 +155,8 @@ async def add_core_data_crud(
     To manipulate core data, prefer using the `get_core_data` and `set_core_data` utils.
     """
     db.add(core_data)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
-    else:
-        return core_data
+    await db.flush()
+    return core_data
 
 
 async def delete_core_data_crud(
@@ -182,4 +168,4 @@ async def delete_core_data_crud(
             models_core.CoreData.schema == schema,
         ),
     )
-    await db.commit()
+    await db.flush()

app/core/google_api/cruds_google_api.py

@@ -1,5 +1,4 @@
 from sqlalchemy import delete, select
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.google_api import models_google_api
@@ -10,11 +9,7 @@ async def create_oauth_flow_state(
     db: AsyncSession,
 ) -> None:
     db.add(oauth_flow_state)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
+    await db.flush()
 
 
 async def get_oauth_flow_state_by_state(

app/core/groups/cruds_groups.py

@@ -3,7 +3,6 @@
 from collections.abc import Sequence
 
 from sqlalchemy import delete, select, update
-from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -54,13 +53,8 @@ async def create_group(
     """Create a new group in database and return it"""
 
     db.add(group)
-    try:
-        await db.commit()
-    except IntegrityError:
-        await db.rollback()
-        raise
-    else:
-        return group
+    await db.flush()
+    return group
 
 
 async def delete_group(db: AsyncSession, group_id: str):
@@ -69,7 +63,7 @@ async def delete_group(db: AsyncSession, group_id: str):
     await db.execute(
         delete(models_groups.CoreGroup).where(models_groups.CoreGroup.id == group_id),
     )
-    await db.commit()
+    await db.flush()
 
 
 async def create_membership(
@@ -79,12 +73,8 @@ async def create_membership(
     """Add a user to a group using a membership"""
 
     db.add(membership)
-    try:
-        await db.commit()
-        return await get_group_by_id(db, membership.group_id)
-    except IntegrityError:
-        await db.rollback()
-        raise
+    await db.flush()
+    return await get_group_by_id(db, membership.group_id)
 
 
 async def delete_membership_by_group_id(
@@ -96,7 +86,7 @@ async def delete_membership_by_group_id(
             models_groups.CoreMembership.group_id == group_id,
         ),
     )
-    await db.commit()
+    await db.flush()
 
 
 async def delete_membership_by_group_and_user_id(
@@ -110,7 +100,7 @@ async def delete_membership_by_group_and_user_id(
             models_groups.CoreMembership.user_id == user_id,
         ),
     )
-    await db.commit()
+    await db.flush()
 
 
 async def update_group(
@@ -123,4 +113,4 @@ async def update_group(
         .where(models_groups.CoreGroup.id == group_id)
         .values(**group_update.model_dump(exclude_none=True)),
     )
-    await db.commit()
+    await db.flush()

app/core/memberships/endpoints_memberships.py

@@ -132,11 +132,7 @@ async def create_association_membership(
         db=db,
         membership=db_association_membership,
     )
-    try:
-        await db.commit()
-    except Exception:
-        await db.rollback()
-        raise
+    await db.flush()
     return db_association_membership
 
 
@@ -170,13 +166,7 @@ async def update_association_membership(
         membership=membership,
     )
 
-    try:
-        await db.commit()
-    except Exception:
-        raise HTTPException(
-            status_code=500,
-            detail="Failed to update membership",
-        )
+    await db.flush()
 
 
 @router.delete(
@@ -219,14 +209,6 @@ async def delete_association_membership(
         membership_id=association_membership_id,
     )
 
-    try:
-        await db.commit()
-    except Exception:
-        raise HTTPException(
-            status_code=500,
-            detail="Failed to delete membership",
-        )
-
 
 @router.get(
     "/memberships/users/{user_id}",
@@ -324,13 +306,9 @@ async def create_user_membership(
     await validate_user_new_membership(db_user_membership, db)
 
     cruds_memberships.create_user_membership(db=db, user_membership=db_user_membership)
-    try:
-        await db.commit()
-    except Exception:
-        raise HTTPException(
-            status_code=500,
-            detail="Failed to create user membership",
-        )
+
+    await db.flush()
+
     return schemas_memberships.UserMembershipComplete(
         **db_user_membership.__dict__,
         user=schemas_users.CoreUserSimple(
@@ -401,11 +379,7 @@ async def add_batch_membership(
                     end_date=detail.end_date,
                 ),
             )
-    try:
-        await db.commit()
-    except Exception:
-        await db.rollback()
-        raise
+    await db.flush()
     return unknown_users
 
 
@@ -448,11 +422,7 @@ async def update_user_membership(
         user_membership_edit=user_membership,
     )
 
-    try:
-        await db.commit()
-    except Exception:
-        await db.rollback()
-        raise
+    await db.flush()
 
 
 @router.delete(
@@ -481,11 +451,3 @@ async def delete_user_membership(
         db=db,
         user_membership_id=membership_id,
     )
-
-    try:
-        await db.commit()
-    except Exception:
-        raise HTTPException(
-            status_code=500,
-            detail="Failed to delete user membership",
-        )

app/core/myeclpay/cruds_myeclpay.py

@@ -68,7 +68,6 @@ async def init_structure_manager_transfer(
             confirmation_token=confirmation_token,
         ),
     )
-    await db.commit()
 
 
 async def get_structure_manager_transfer_by_secret(
@@ -196,7 +195,6 @@ async def update_store(
         .where(models_myeclpay.Store.id == store_id)
         .values(**store_update.model_dump(exclude_none=True)),
     )
-    await db.commit()
 
 
 async def delete_store(