Coverage for opt/mealie/lib/python3.12/site-packages/mealie/alembic/versions/2024-02-23-16.15.07_2298bb460ffd_added_user_to_shopping_list.py: 50%

52 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 14:03 +0000

1"""added user to shopping list 

2 

3Revision ID: 2298bb460ffd 

4Revises: ba1e4a6cfe99 

5Create Date: 2024-02-23 16:15:07.115641 

6 

7""" 

8 

9from uuid import UUID 1a

10 

11import sqlalchemy as sa 1a

12from alembic import op 1a

13from sqlalchemy import orm 1a

14 

15import mealie.db.migration_types 1a

16from mealie.core.root_logger import get_logger 1a

17 

18logger = get_logger() 1a

19 

20# revision identifiers, used by Alembic. 

21revision = "2298bb460ffd" 1a

22down_revision = "ba1e4a6cfe99" 1a

23branch_labels: str | tuple[str, ...] | None = None 1a

24depends_on: str | tuple[str, ...] | None = None 1a

25 

26 

27def is_postgres(): 1a

28 return op.get_context().dialect.name == "postgresql" 

29 

30 

31def find_user_id_for_group(group_id: UUID): 1a

32 bind = op.get_bind() 

33 session = orm.Session(bind=bind) 

34 

35 if is_postgres(): 

36 stmt = "SELECT id FROM users WHERE group_id=:group_id AND admin = TRUE LIMIT 1" 

37 else: 

38 stmt = "SELECT id FROM users WHERE group_id=:group_id AND admin = 1 LIMIT 1" 

39 

40 with session: 

41 try: 

42 # try to find an admin user 

43 return session.execute(sa.text(stmt).bindparams(group_id=group_id)).scalar_one() 

44 except orm.exc.NoResultFound: 

45 pass 

46 

47 try: 

48 # fallback to any user 

49 return session.execute( 

50 sa.text("SELECT id FROM users WHERE group_id=:group_id LIMIT 1").bindparams(group_id=group_id) 

51 ).scalar_one() 

52 except orm.exc.NoResultFound: 

53 pass 

54 

55 # no user could be found 

56 return None 

57 

58 

59def populate_shopping_list_users(): 1a

60 bind = op.get_bind() 1a

61 session = orm.Session(bind=bind) 1a

62 

63 with session: 1a

64 list_ids_and_group_ids = session.execute(sa.text("SELECT id, group_id FROM shopping_lists")).all() 1a

65 for list_id, group_id in list_ids_and_group_ids: 65 ↛ 66line 65 didn't jump to line 66 because the loop on line 65 never started1a

66 user_id = find_user_id_for_group(group_id) 

67 if user_id: 

68 session.execute( 

69 sa.text("UPDATE shopping_lists SET user_id=:user_id WHERE id=:id").bindparams( 

70 user_id=user_id, id=list_id 

71 ) 

72 ) 

73 else: 

74 logger.warning( 

75 f"No user found for shopping list {list_id} with group {group_id}; deleting shopping list" 

76 ) 

77 session.execute(sa.text("DELETE FROM shopping_lists WHERE id=:id").bindparams(id=list_id)) 

78 

79 

80def upgrade(): 1a

81 # ### commands auto generated by Alembic - please adjust! ### 

82 with op.batch_alter_table("shopping_lists") as batch_op: 1a

83 # allow nulls during migration 

84 batch_op.add_column(sa.Column("user_id", mealie.db.migration_types.GUID(), nullable=True)) 1a

85 batch_op.create_index(op.f("ix_shopping_lists_user_id"), ["user_id"], unique=False) 1a

86 batch_op.create_foreign_key("fk_user_shopping_lists", "users", ["user_id"], ["id"]) 1a

87 # ### end Alembic commands ### 

88 

89 populate_shopping_list_users() 1a

90 

91 # forbid nulls after migration 

92 with op.batch_alter_table("shopping_lists") as batch_op: 1a

93 batch_op.alter_column("user_id", nullable=False) 1a

94 

95 

96def downgrade(): 1a

97 # ### commands auto generated by Alembic - please adjust! ### 

98 op.drop_constraint(None, "shopping_lists", type_="foreignkey") 

99 op.drop_index(op.f("ix_shopping_lists_user_id"), table_name="shopping_lists") 

100 op.drop_column("shopping_lists", "user_id") 

101 # ### end Alembic commands ###