seed: add seed examples to database

Author: pedro-fs-garcia

app/domains/live_chat/routers/chat_router.py

@@ -46,7 +46,7 @@ async def connect_to_conversation(
         )
         return
 
-    await ws.accept()
+    await ws.accept(subprotocol="access_token")
     conn = ChatConnection(ws, response, user)
     joined = False
 

app/seed/run_seed.py

@@ -4,15 +4,37 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
 
+from app.db.mongo.db import mongo_db
 from app.db.postgres.engine import async_session
 from app.seed import seed
+from app.seed.seed_examples import (
+    seed_example_attendances,
+    seed_example_conversations,
+    seed_example_tickets,
+    seed_example_user_roles,
+    seed_example_users,
+)
 
 
 async def run() -> None:
+    # --- Postgres ---
     async with async_session() as db, db.begin():
         await seed.seed_roles(db)
         await seed.seed_permissions(db)
         await seed.seed_role_permissions(db)
+        await seed.seed_users(db)
+        await seed_example_users(db)
+        await seed_example_user_roles(db)
+
+    # --- MongoDB ---
+    await mongo_db.connect()
+    try:
+        mongo = mongo_db.get_db()
+        await seed_example_attendances(mongo)
+        await seed_example_tickets(mongo)
+        await seed_example_conversations(mongo)
+    finally:
+        await mongo_db.disconnect()
 
 
 if __name__ == "__main__":

app/seed/seed.py

@@ -1,8 +1,13 @@
+import secrets
+import string
+from typing import Any
+
 from sqlalchemy import insert, select
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.domains.auth.models import Permission, Role, role_permissions
+from app.core.security import PasswordSecurity
+from app.domains.auth.models import Permission, Role, User, role_permissions, user_roles
 
 
 async def seed_roles(session: AsyncSession) -> None:
@@ -62,7 +67,8 @@ async def seed_permissions(session: AsyncSession) -> None:
         {"name": "ticket:update_status", "description": "Update ticket status"},
     ]
 
-    await session.execute(insert(Permission).values(permissions))
+    insert_stmt = pg_insert(Permission).values(permissions).on_conflict_do_nothing()
+    await session.execute(insert_stmt)
 
 
 async def seed_role_permissions(session: AsyncSession) -> None:
@@ -90,3 +96,57 @@ async def seed_role_permissions(session: AsyncSession) -> None:
         values = [{"role_id": role_id, "permission_id": perm_id} for perm_id in permission_ids]
         insert_stmt = pg_insert(role_permissions).values(values).on_conflict_do_nothing()
         await session.execute(insert_stmt)
+
+
+def generate_random_password(length: int = 16) -> str:
+    alphabet = string.ascii_letters + string.digits + string.punctuation
+
+    while True:
+        password = "".join(secrets.choice(alphabet) for _ in range(length))
+
+        if (
+            any(c.islower() for c in password)
+            and any(c.isupper() for c in password)
+            and any(c.isdigit() for c in password)
+            and any(c in string.punctuation for c in password)
+        ):
+            return password
+
+
+async def seed_users(session: AsyncSession) -> None:
+    password_security = PasswordSecurity()
+
+    admin_names: list[str] = ["angelina", "eduardo", "julia", "mafe", "pedro", "wesley"]
+    default_password = "Admin@123!"
+    users_payload: list[dict[str, Any]] = []
+
+    for name in admin_names:
+        users_payload.append(
+            {
+                "email": f"{name}@syncdesk.pro",
+                "password_hash": password_security.generate_password_hash(default_password),
+                "username": name,
+                "name": name,
+                "must_change_password": False,
+                "must_accept_terms": False
+            }
+        )
+
+    insert_stmt = pg_insert(User).values(users_payload).on_conflict_do_nothing()
+    await session.execute(insert_stmt)
+
+    role_result = await session.execute(select(Role.id).where(Role.name == "admin"))
+    admin_role_id = role_result.scalar_one_or_none()
+    if admin_role_id is None:
+        return
+
+    users_result = await session.execute(
+        select(User.id).where(User.email.in_([user["email"] for user in users_payload]))
+    )
+    user_ids = users_result.scalars().all()
+    if not user_ids:
+        return
+
+    user_role_values = [{"user_id": user_id, "role_id": admin_role_id} for user_id in user_ids]
+    role_insert_stmt = pg_insert(user_roles).values(user_role_values).on_conflict_do_nothing()
+    await session.execute(role_insert_stmt)