feat: implement user_favori table and drop favori

migrations/versions/16c8bc08a922_create_user_favori_table.py

@@ -0,0 +1,100 @@
+"""Delete and re-create favori table
+
+Revision ID: 16c8bc08a922
+Revises: a3fe3092f50c
+Create Date: 2023-10-29 18:56:52.112998
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision = "16c8bc08a922"
+down_revision = "a3fe3092f50c"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "user_favori",
+        sa.Column("formation_id", sa.String(length=36), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["formation_id"],
+            ["formation.id"],
+            ondelete="CASCADE",
+            name="favori_formation_id_fk",
+        ),
+        sa.ForeignKeyConstraint(
+            ["user_id"], ["user.id"], ondelete="CASCADE", name="favori_user_id_fk"
+        ),
+        sa.PrimaryKeyConstraint("formation_id", "user_id", name="favori_pk"),
+    )
+    op.create_index(
+        op.f("ix_user_favori_formation_id"),
+        "user_favori",
+        ["formation_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_user_favori_user_id"), "user_favori", ["user_id"], unique=False
+    )
+
+    op.execute(
+        """
+        INSERT INTO user_favori (formation_id, user_id, created_at, updated_at)
+        SELECT fr.id, f.request_user_id, f.created_at, f.updated_at
+        FROM favori f
+        JOIN formation fr ON f.code_nsf = fr.code_nsf
+                        AND f.sigle_type_formation = fr.type
+                        AND f.libelle_formation_principal = fr.libelle
+                        AND f.tutelle = fr.tutelle
+                        AND f.url_et_id_onisep = fr.url
+                        AND f.niveau_de_sortie_indicatif = fr.niveau_de_sortie
+                        AND f.duree = fr.duree
+                        AND f.created_at = fr.created_at
+                        AND f.updated_at = fr.updated_at;
+        """
+    )
+
+    op.drop_table("favori")
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "favori",
+        sa.Column("id", mysql.INTEGER(), autoincrement=True, nullable=False),
+        sa.Column("code_nsf", mysql.TEXT(), nullable=True),
+        sa.Column("sigle_type_formation", mysql.TEXT(), nullable=True),
+        sa.Column("libelle_type_formation", mysql.TEXT(), nullable=True),
+        sa.Column("libelle_formation_principal", mysql.TEXT(), nullable=True),
+        sa.Column("sigle_formation", mysql.TEXT(), nullable=True),
+        sa.Column("duree", mysql.TEXT(), nullable=True),
+        sa.Column("niveau_de_sortie_indicatif", mysql.TEXT(), nullable=True),
+        sa.Column("code_rncp", mysql.TEXT(), nullable=True),
+        sa.Column("niveau_de_certification", mysql.TEXT(), nullable=True),
+        sa.Column("libelle_niveau_de_certification", mysql.TEXT(), nullable=True),
+        sa.Column("tutelle", mysql.TEXT(), nullable=True),
+        sa.Column("url_et_id_onisep", mysql.TEXT(), nullable=False),
+        sa.Column(
+            "request_user_id", mysql.INTEGER(), autoincrement=False, nullable=True
+        ),
+        sa.Column("created_at", mysql.DATETIME(), nullable=False),
+        sa.Column("updated_at", mysql.DATETIME(), nullable=False),
+        sa.ForeignKeyConstraint(["request_user_id"], ["user.id"], name="favori_ibfk_1"),
+        sa.PrimaryKeyConstraint("id"),
+        mysql_collate="utf8mb4_0900_ai_ci",
+        mysql_default_charset="utf8mb4",
+        mysql_engine="InnoDB",
+    )
+    op.drop_index(op.f("ix_user_favori_user_id"), table_name="user_favori")
+    op.drop_index(op.f("ix_user_favori_formation_id"), table_name="user_favori")
+    op.drop_table("user_favori")
+    # ### end Alembic commands ###

migrations/versions/a3fe3092f50c_create_formation_table.py

@@ -34,6 +34,24 @@ def upgrade():
         sa.PrimaryKeyConstraint("id"),
         sa.UniqueConstraint("url"),
     )
+    op.execute(
+        """
+        INSERT INTO formation (id, code_nsf, type, libelle, tutelle, url, domain, niveau_de_sortie, duree, created_at, updated_at)
+        SELECT
+            UUID(),
+            f.code_nsf,
+            f.sigle_type_formation,
+            f.libelle_formation_principal,
+            f.tutelle,
+            f.url_et_id_onisep,
+            "NULL",
+            f.niveau_de_sortie_indicatif,
+            f.duree,
+            f.created_at,
+            f.updated_at
+        FROM favori f;
+        """
+    )
     # ### end Alembic commands ###
 
 

src/blueprints/auth.py

@@ -19,7 +19,7 @@
     HTTP_409_CONFLICT,
 )
 from src import db
-from src.models import User, Favori
+from src.models import User, UserFavori
 
 import re
 from typing import Tuple
@@ -212,7 +212,7 @@ def edit_user() -> Tuple[Response, int] | HTTPException:
 
 
 def remove_favoris(user_id: int) -> None:
-    favoris = Favori.query.filter_by(request_user_id=user_id).all()
+    favoris = UserFavori.query.filter_by(request_user_id=user_id).all()
     list(map(lambda f: db.session.delete(f), favoris))
     db.session.commit()
 

src/blueprints/favoris.py

@@ -12,7 +12,7 @@
     HTTP_409_CONFLICT,
 )
 from src import db
-from src.models import Favori
+from src.models import UserFavori
 
 from typing import Tuple
 
@@ -30,12 +30,12 @@ def post_favori_by_user_id() -> Tuple[Response, int] | HTTPException:
     if not validators.url(favori_data.get("url_et_id_onisep", "")):
         abort(HTTP_400_BAD_REQUEST, "Enter valid url")
 
-    if Favori.query.filter_by(
+    if UserFavori.query.filter_by(
         request_user_id=current_user,
         url_et_id_onisep=favori_data.get("url_et_id_onisep", ""),
     ).first():
         abort(HTTP_409_CONFLICT, "URL already exists")
-    favori = Favori(**favori_data, request_user_id=current_user)
+    favori = UserFavori(**favori_data, request_user_id=current_user)
     db.session.add(favori)
     db.session.commit()
     return (
@@ -50,8 +50,8 @@ def post_favori_by_user_id() -> Tuple[Response, int] | HTTPException:
 def get_favoris_by_user_id() -> Tuple[Response, int]:
     current_user = get_jwt_identity()
     favoris = (
-        Favori.query.filter(Favori.request_user_id == current_user)
-        .order_by(Favori.created_at.asc())
+        UserFavori.query.filter(UserFavori.request_user_id == current_user)
+        .order_by(UserFavori.created_at.asc())
         .all()
     )
     return jsonify({"size": len(favoris), "results": favoris}), HTTP_200_OK
@@ -65,8 +65,8 @@ def get_favoris_by_user_id() -> Tuple[Response, int]:
 def get_favoris_ids() -> Tuple[Response, int]:
     current_user = get_jwt_identity()
     result = (
-        Favori.query.with_entities(Favori.url_et_id_onisep, Favori.id)
-        .filter(Favori.request_user_id == current_user)
+        UserFavori.query.with_entities(UserFavori.url_et_id_onisep, UserFavori.id)
+        .filter(UserFavori.request_user_id == current_user)
         .all()
     )
     favori_data = [{"id": row.id, "url": row.url_et_id_onisep} for row in result]
@@ -79,7 +79,7 @@ def get_favoris_ids() -> Tuple[Response, int]:
 def remove_favori(id: int) -> Tuple[Response, int] | HTTPException:
     current_user = get_jwt_identity()
 
-    favori = Favori.query.filter_by(request_user_id=current_user, id=id).first()
+    favori = UserFavori.query.filter_by(request_user_id=current_user, id=id).first()
 
     if not favori:
         abort(HTTP_404_NOT_FOUND, "Favoris not found")

src/blueprints/formations.py

@@ -12,7 +12,7 @@
 from src.constants.http_status_codes import (
     HTTP_200_OK,
 )
-from src.models.favori import Favori
+from src.models.user_favori import UserFavori
 from src.models.user import User
 
 from src import db
@@ -42,8 +42,8 @@ def get_formation_by_id(id: str) -> Tuple[Response, int] | HTTPException:
     db.session.add(new_user)
     db.session.flush()
 
-    # Create a Favori object
-    new_favori = Favori(
+    # Create a UserFavori object
+    new_favori = UserFavori(
         code_nsf="123",
         sigle_type_formation="ABC",
         libelle_type_formation="Type ABC",
@@ -56,7 +56,7 @@ def get_formation_by_id(id: str) -> Tuple[Response, int] | HTTPException:
         libelle_niveau_de_certification="Certification ABC",
         tutelle="Example Tutelle",
         url_et_id_onisep="https://example.com/2",
-        request_user_id=new_user.id,  # Assign the user object to the Favori
+        request_user_id=new_user.id,  # Assign the user object to the UserFavori
     )
     db.session.add(new_favori)
     db.session.commit()

src/models/__init__.py

@@ -1,5 +1,5 @@
 from src.models.formation import Formation
 from .user import User
-from .favori import Favori
+from .user_favori import UserFavori
 
-models = [User, Favori, Formation]
+models = [User, UserFavori, Formation]

src/models/user.py

@@ -20,4 +20,4 @@ class User(BaseModel):
     email = db.Column(db.String(200), unique=True, nullable=False)
     password = db.Column(db.Text(), nullable=False)
     profile_pic_url = db.Column(db.Text)
-    favoris = db.relationship("Favori", backref="user", lazy="dynamic")
+    favoris = db.relationship("UserFavori", backref="user", lazy="dynamic")

src/models/favori.py → src/models/user_favori.py

@@ -1,24 +1,28 @@
 from dataclasses import dataclass
+from typing import Callable
+
+from cuid2 import cuid_wrapper
 from src import db
 from src.models.base_model import BaseModel
 
 
 @dataclass
-class Favori(BaseModel):
-    __tablename__ = "favori"
-
-    formation_id: str
-    user_id: int
+class UserFavori(BaseModel):
+    __tablename__ = "user_favori"
 
     formation_id = db.Column(
         db.String(36),
         db.ForeignKey("formation.id", ondelete="CASCADE"),
-        index=True,
         primary_key=True,
+        index=True,
     )
+
     user_id = db.Column(
         db.Integer,
         db.ForeignKey("user.id", ondelete="CASCADE"),
-        index=True,
         primary_key=True,
+        index=True,
     )
+
+    formation = db.relationship("Formation", back_populates="favoris")
+    user = db.relationship("User", back_populates="users")