Coverage for opt/mealie/lib/python3.12/site-packages/mealie/alembic/versions/2024-10-08-21.17.31_86054b40fd06_added_query_filter_string_to_cookbook_.py: 61%

94 statements  

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

1"""added query_filter_string to cookbook and mealplan 

2 

3Revision ID: 86054b40fd06 

4Revises: 602927e1013e 

5Create Date: 2024-10-08 21:17:31.601903 

6 

7""" 

8 

9import sqlalchemy as sa 1a

10from alembic import op 1a

11from sqlalchemy import orm 1a

12 

13from mealie.db.models._model_utils import guid 1a

14 

15# revision identifiers, used by Alembic. 

16revision = "86054b40fd06" 1a

17down_revision: str | None = "602927e1013e" 1a

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

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

20 

21 

22# Intermediate table definitions 

23class SqlAlchemyBase(orm.DeclarativeBase): 1a

24 pass 1a

25 

26 

27class Category(SqlAlchemyBase): 1a

28 __tablename__ = "categories" 1a

29 id: orm.Mapped[guid.GUID] = orm.mapped_column(guid.GUID, primary_key=True, default=guid.GUID.generate) 1a

30 

31 

32class Tag(SqlAlchemyBase): 1a

33 __tablename__ = "tags" 1a

34 id: orm.Mapped[guid.GUID] = orm.mapped_column(guid.GUID, primary_key=True, default=guid.GUID.generate) 1a

35 

36 

37class Tool(SqlAlchemyBase): 1a

38 __tablename__ = "tools" 1a

39 id: orm.Mapped[guid.GUID] = orm.mapped_column(guid.GUID, primary_key=True, default=guid.GUID.generate) 1a

40 

41 

42class Household(SqlAlchemyBase): 1a

43 __tablename__ = "households" 1a

44 id: orm.Mapped[guid.GUID] = orm.mapped_column(guid.GUID, primary_key=True, default=guid.GUID.generate) 1a

45 

46 

47cookbooks_to_categories = sa.Table( 1a

48 "cookbooks_to_categories", 

49 SqlAlchemyBase.metadata, 

50 sa.Column("cookbook_id", guid.GUID, sa.ForeignKey("cookbooks.id"), index=True), 

51 sa.Column("category_id", guid.GUID, sa.ForeignKey("categories.id"), index=True), 

52) 

53 

54cookbooks_to_tags = sa.Table( 1a

55 "cookbooks_to_tags", 

56 SqlAlchemyBase.metadata, 

57 sa.Column("cookbook_id", guid.GUID, sa.ForeignKey("cookbooks.id"), index=True), 

58 sa.Column("tag_id", guid.GUID, sa.ForeignKey("tags.id"), index=True), 

59) 

60 

61cookbooks_to_tools = sa.Table( 1a

62 "cookbooks_to_tools", 

63 SqlAlchemyBase.metadata, 

64 sa.Column("cookbook_id", guid.GUID, sa.ForeignKey("cookbooks.id"), index=True), 

65 sa.Column("tool_id", guid.GUID, sa.ForeignKey("tools.id"), index=True), 

66) 

67 

68plan_rules_to_categories = sa.Table( 1a

69 "plan_rules_to_categories", 

70 SqlAlchemyBase.metadata, 

71 sa.Column("group_plan_rule_id", guid.GUID, sa.ForeignKey("group_meal_plan_rules.id"), index=True), 

72 sa.Column("category_id", guid.GUID, sa.ForeignKey("categories.id"), index=True), 

73) 

74 

75plan_rules_to_tags = sa.Table( 1a

76 "plan_rules_to_tags", 

77 SqlAlchemyBase.metadata, 

78 sa.Column("plan_rule_id", guid.GUID, sa.ForeignKey("group_meal_plan_rules.id"), index=True), 

79 sa.Column("tag_id", guid.GUID, sa.ForeignKey("tags.id"), index=True), 

80) 

81 

82plan_rules_to_households = sa.Table( 1a

83 "plan_rules_to_households", 

84 SqlAlchemyBase.metadata, 

85 sa.Column("group_plan_rule_id", guid.GUID, sa.ForeignKey("group_meal_plan_rules.id"), index=True), 

86 sa.Column("household_id", guid.GUID, sa.ForeignKey("households.id"), index=True), 

87) 

88 

89 

90class CookBook(SqlAlchemyBase): 1a

91 __tablename__ = "cookbooks" 1a

92 

93 id: orm.Mapped[guid.GUID] = orm.mapped_column(guid.GUID, primary_key=True, default=guid.GUID.generate) 1a

94 query_filter_string: orm.Mapped[str] = orm.mapped_column(sa.String, nullable=False, default="") 1a

95 

96 categories: orm.Mapped[list[Category]] = orm.relationship( 1a

97 Category, secondary=cookbooks_to_categories, single_parent=True 

98 ) 

99 require_all_categories: orm.Mapped[bool | None] = orm.mapped_column(sa.Boolean, default=True) 1a

100 

101 tags: orm.Mapped[list[Tag]] = orm.relationship(Tag, secondary=cookbooks_to_tags, single_parent=True) 1a

102 require_all_tags: orm.Mapped[bool | None] = orm.mapped_column(sa.Boolean, default=True) 1a

103 

104 tools: orm.Mapped[list[Tool]] = orm.relationship(Tool, secondary=cookbooks_to_tools, single_parent=True) 1a

105 require_all_tools: orm.Mapped[bool | None] = orm.mapped_column(sa.Boolean, default=True) 1a

106 

107 

108class GroupMealPlanRules(SqlAlchemyBase): 1a

109 __tablename__ = "group_meal_plan_rules" 1a

110 

111 id: orm.Mapped[guid.GUID] = orm.mapped_column(guid.GUID, primary_key=True, default=guid.GUID.generate) 1a

112 query_filter_string: orm.Mapped[str] = orm.mapped_column(sa.String, nullable=False, default="") 1a

113 

114 categories: orm.Mapped[list[Category]] = orm.relationship(Category, secondary=plan_rules_to_categories) 1a

115 tags: orm.Mapped[list[Tag]] = orm.relationship(Tag, secondary=plan_rules_to_tags) 1a

116 households: orm.Mapped[list["Household"]] = orm.relationship("Household", secondary=plan_rules_to_households) 1a

117 

118 

119def migrate_cookbooks(): 1a

120 bind = op.get_bind() 1a

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

122 

123 cookbooks = session.query(CookBook).all() 1a

124 for cookbook in cookbooks: 124 ↛ 125line 124 didn't jump to line 125 because the loop on line 124 never started1a

125 parts = [] 

126 if cookbook.categories: 

127 relop = "CONTAINS ALL" if cookbook.require_all_categories else "IN" 

128 vals = ",".join([f'"{cat.id}"' for cat in cookbook.categories]) 

129 parts.append(f"recipe_category.id {relop} [{vals}]") 

130 if cookbook.tags: 

131 relop = "CONTAINS ALL" if cookbook.require_all_tags else "IN" 

132 vals = ",".join([f'"{tag.id}"' for tag in cookbook.tags]) 

133 parts.append(f"tags.id {relop} [{vals}]") 

134 if cookbook.tools: 

135 relop = "CONTAINS ALL" if cookbook.require_all_tools else "IN" 

136 vals = ",".join([f'"{tool.id}"' for tool in cookbook.tools]) 

137 parts.append(f"tools.id {relop} [{vals}]") 

138 

139 cookbook.query_filter_string = " AND ".join(parts) 

140 

141 session.commit() 1a

142 

143 

144def migrate_mealplan_rules(): 1a

145 bind = op.get_bind() 1a

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

147 

148 rules = session.query(GroupMealPlanRules).all() 1a

149 for rule in rules: 149 ↛ 150line 149 didn't jump to line 150 because the loop on line 149 never started1a

150 parts = [] 

151 if rule.categories: 

152 vals = ",".join([f'"{cat.id}"' for cat in rule.categories]) 

153 parts.append(f"recipe_category.id CONTAINS ALL [{vals}]") 

154 if rule.tags: 

155 vals = ",".join([f'"{tag.id}"' for tag in rule.tags]) 

156 parts.append(f"tags.id CONTAINS ALL [{vals}]") 

157 if rule.households: 

158 vals = ",".join([f'"{household.id}"' for household in rule.households]) 

159 parts.append(f"household_id IN [{vals}]") 

160 

161 rule.query_filter_string = " AND ".join(parts) 

162 

163 session.commit() 1a

164 

165 

166def upgrade(): 1a

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

168 with op.batch_alter_table("cookbooks", schema=None) as batch_op: 1a

169 batch_op.add_column(sa.Column("query_filter_string", sa.String(), nullable=False, server_default="")) 1a

170 

171 with op.batch_alter_table("group_meal_plan_rules", schema=None) as batch_op: 1a

172 batch_op.add_column(sa.Column("query_filter_string", sa.String(), nullable=False, server_default="")) 1a

173 

174 # ### end Alembic commands ### 

175 

176 migrate_cookbooks() 1a

177 migrate_mealplan_rules() 1a

178 

179 

180def downgrade(): 1a

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

182 with op.batch_alter_table("group_meal_plan_rules", schema=None) as batch_op: 

183 batch_op.drop_column("query_filter_string") 

184 

185 with op.batch_alter_table("cookbooks", schema=None) as batch_op: 

186 batch_op.drop_column("query_filter_string") 

187 

188 # ### end Alembic commands ###