Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/groups/controller_migrations.py: 52%

25 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 17:29 +0000

1import shutil 1a

2 

3from fastapi import File, Form 1a

4from fastapi.datastructures import UploadFile 1a

5 

6from mealie.core.dependencies import get_temporary_zip_path 1a

7from mealie.routes._base import BaseUserController, controller 1a

8from mealie.routes._base.routers import UserAPIRouter 1a

9from mealie.schema.group.group_migration import SupportedMigrations 1a

10from mealie.schema.reports.reports import ReportSummary 1a

11from mealie.services.migrations import ( 1a

12 BaseMigrator, 

13 ChowdownMigrator, 

14 CooknMigrator, 

15 CopyMeThatMigrator, 

16 MealieAlphaMigrator, 

17 MyRecipeBoxMigrator, 

18 NextcloudMigrator, 

19 PaprikaMigrator, 

20 PlanToEatMigrator, 

21 RecipeKeeperMigrator, 

22 TandoorMigrator, 

23) 

24 

25router = UserAPIRouter(prefix="/groups/migrations", tags=["Groups: Migrations"]) 1a

26 

27 

28@controller(router) 1a

29class GroupMigrationController(BaseUserController): 1a

30 @router.post("", response_model=ReportSummary) 1a

31 def start_data_migration( 1a

32 self, 

33 add_migration_tag: bool = Form(False), 

34 migration_type: SupportedMigrations = Form(...), 

35 archive: UploadFile = File(...), 

36 ): 

37 with get_temporary_zip_path() as temp_path: 

38 # Save archive to temp_path 

39 with temp_path.open("wb") as buffer: 

40 shutil.copyfileobj(archive.file, buffer) 

41 

42 args = { 

43 "archive": temp_path, 

44 "db": self.repos, 

45 "session": self.session, 

46 "user_id": self.user.id, 

47 "household_id": self.household_id, 

48 "group_id": self.group_id, 

49 "add_migration_tag": add_migration_tag, 

50 "translator": self.translator, 

51 } 

52 

53 table: dict[SupportedMigrations, type[BaseMigrator]] = { 

54 SupportedMigrations.chowdown: ChowdownMigrator, 

55 SupportedMigrations.copymethat: CopyMeThatMigrator, 

56 SupportedMigrations.mealie_alpha: MealieAlphaMigrator, 

57 SupportedMigrations.nextcloud: NextcloudMigrator, 

58 SupportedMigrations.paprika: PaprikaMigrator, 

59 SupportedMigrations.tandoor: TandoorMigrator, 

60 SupportedMigrations.plantoeat: PlanToEatMigrator, 

61 SupportedMigrations.myrecipebox: MyRecipeBoxMigrator, 

62 SupportedMigrations.recipekeeper: RecipeKeeperMigrator, 

63 SupportedMigrations.cookn: CooknMigrator, 

64 } 

65 

66 constructor = table.get(migration_type, None) 

67 

68 if constructor is None: 

69 raise ValueError(f"Unsupported migration type: {migration_type}") 

70 

71 migrator = constructor(**args) 

72 

73 migration_result = migrator.migrate(f"{migration_type.value.title()} Migration") 

74 return migration_result