feat: add script to migrate users from accounts to events

scripts/migrate_users_from_account_to_events.py

@@ -0,0 +1,56 @@
+import argparse
+import asyncio
+
+from pydantic import BaseModel, TypeAdapter, model_validator
+
+from src.config import settings
+from src.repositories.users.repository import user_repository
+from src.storages.sql import SQLAlchemyStorage
+
+
+class InputUser(BaseModel):
+    # {
+    #     "_id": {
+    #         "$oid": "65f6ef2847289ea08482e3bb"
+    #     },
+    #     "innopolis_sso": {
+    #         "email": "[email protected]"
+    #     }
+    # }
+
+    innohassle_id: str
+    email: str
+
+    @model_validator(mode="before")
+    def extract_fields(cls, values):
+        return {
+            "innohassle_id": values["_id"]["$oid"],
+            "email": values["innopolis_sso"]["email"],
+        }
+
+
+async def _setup_repository():
+    storage = SQLAlchemyStorage.from_url(settings.db_url.get_secret_value())
+    user_repository.update_storage(storage)
+
+
+async def main():
+    parser = argparse.ArgumentParser(description="Migrate users from accounts to events.")
+
+    parser.add_argument("input_file", type=str, help="The filename of the input file")
+    args = parser.parse_args()
+
+    type_adapter = TypeAdapter(list[InputUser])
+
+    with open(args.input_file, "r") as file:
+        data = file.read()
+        users = type_adapter.validate_json(data)
+
+    await _setup_repository()
+    existing_users = await user_repository.read_mapping_by_emails([user.email for user in users])
+    for user in users:
+        await user_repository.update_innohassle_id(existing_users[user.email], user.innohassle_id)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/repositories/users/repository.py

@@ -79,6 +79,15 @@ async def read_all(self) -> list[ViewUser]:
         async with self._create_session() as session:
             return await CRUD.read_all(session)
 
+    async def read_mapping_by_emails(self, emails: list[str]) -> dict[str, int | None]:
+        async with self._create_session() as session:
+            q = select(User.email, User.id).where(User.email.in_(emails))
+            mapping = (await session.execute(q)).all()
+            result = dict.fromkeys(emails)
+            for email, user_id in mapping:
+                result[email] = user_id
+            return result
+
     async def read_id_by_email(self, email: str) -> int:
         async with self._create_session() as session:
             user_id = await session.scalar(select(User.id).where(User.email == email))