db.py (5420B)
1 """CRUD functions for recipes, ingredients, staples, and grocery list items.""" 2 from __future__ import annotations 3 4 import sqlite3 5 6 7 # --- recipes --- 8 9 def add_recipe(conn, name, servings, instructions, source, is_keto) -> int: 10 cur = conn.execute( 11 "INSERT INTO recipes (name, servings, instructions, source, is_keto) " 12 "VALUES (?, ?, ?, ?, ?)", 13 (name, servings, instructions, source, int(is_keto)), 14 ) 15 conn.commit() 16 return cur.lastrowid 17 18 19 def get_recipe(conn, recipe_id: int) -> sqlite3.Row | None: 20 return conn.execute("SELECT * FROM recipes WHERE id = ?", (recipe_id,)).fetchone() 21 22 23 def get_recipe_by_name(conn, name: str) -> sqlite3.Row | None: 24 return conn.execute( 25 "SELECT * FROM recipes WHERE name = ? COLLATE NOCASE", (name,) 26 ).fetchone() 27 28 29 def update_recipe_instructions(conn, recipe_id: int, instructions: str) -> None: 30 conn.execute( 31 "UPDATE recipes SET instructions = ? WHERE id = ?", (instructions, recipe_id) 32 ) 33 conn.commit() 34 35 36 def list_recipes(conn, tag: str | None = None) -> list[sqlite3.Row]: 37 if tag == "keto": 38 return conn.execute("SELECT * FROM recipes WHERE is_keto = 1").fetchall() 39 return conn.execute("SELECT * FROM recipes").fetchall() 40 41 42 def search_recipes_by_name(conn, term: str) -> list[sqlite3.Row]: 43 return conn.execute( 44 "SELECT * FROM recipes WHERE name LIKE ? COLLATE NOCASE", (f"%{term}%",) 45 ).fetchall() 46 47 48 def delete_recipe(conn, recipe_id: int) -> None: 49 # Ingredients cascade-delete automatically (see models.py schema / PRAGMA). 50 conn.execute("DELETE FROM recipes WHERE id = ?", (recipe_id,)) 51 conn.commit() 52 53 54 # --- tags / groups --- 55 56 def get_tag_by_name(conn, name: str) -> sqlite3.Row | None: 57 return conn.execute( 58 "SELECT * FROM tags WHERE name = ? COLLATE NOCASE", (name,) 59 ).fetchone() 60 61 62 def get_or_create_tag(conn, name: str) -> sqlite3.Row: 63 conn.execute( 64 "INSERT INTO tags (name) VALUES (?) ON CONFLICT(name) DO NOTHING", (name,) 65 ) 66 conn.commit() 67 return get_tag_by_name(conn, name) 68 69 70 def delete_tag(conn, tag_id: int) -> None: 71 # recipe_tags rows cascade-delete automatically (see models.py schema). 72 conn.execute("DELETE FROM tags WHERE id = ?", (tag_id,)) 73 conn.commit() 74 75 76 def count_recipes_for_tag(conn, tag_id: int) -> int: 77 row = conn.execute( 78 "SELECT COUNT(*) AS n FROM recipe_tags WHERE tag_id = ?", (tag_id,) 79 ).fetchone() 80 return row["n"] 81 82 83 def assign_tag_to_recipe(conn, recipe_id: int, tag_id: int) -> None: 84 conn.execute( 85 "INSERT OR IGNORE INTO recipe_tags (recipe_id, tag_id) VALUES (?, ?)", 86 (recipe_id, tag_id), 87 ) 88 conn.commit() 89 90 91 def remove_tag_from_recipe(conn, recipe_id: int, tag_id: int) -> None: 92 conn.execute( 93 "DELETE FROM recipe_tags WHERE recipe_id = ? AND tag_id = ?", 94 (recipe_id, tag_id), 95 ) 96 conn.commit() 97 98 99 def list_recipes_by_tag(conn, tag_name: str) -> list[sqlite3.Row]: 100 return conn.execute( 101 "SELECT recipes.* FROM recipes " 102 "JOIN recipe_tags ON recipe_tags.recipe_id = recipes.id " 103 "JOIN tags ON tags.id = recipe_tags.tag_id " 104 "WHERE tags.name = ? COLLATE NOCASE", 105 (tag_name,), 106 ).fetchall() 107 108 109 def list_tags(conn) -> list[sqlite3.Row]: 110 return conn.execute( 111 "SELECT tags.*, COUNT(recipe_tags.recipe_id) AS recipe_count " 112 "FROM tags LEFT JOIN recipe_tags ON recipe_tags.tag_id = tags.id " 113 "GROUP BY tags.id ORDER BY tags.name COLLATE NOCASE" 114 ).fetchall() 115 116 117 # --- ingredients --- 118 119 def add_ingredient(conn, recipe_id, name, quantity, unit, category) -> int: 120 cur = conn.execute( 121 "INSERT INTO ingredients (recipe_id, name, quantity, unit, category) " 122 "VALUES (?, ?, ?, ?, ?)", 123 (recipe_id, name, quantity, unit, category), 124 ) 125 conn.commit() 126 return cur.lastrowid 127 128 129 def list_ingredients_for_recipe(conn, recipe_id: int) -> list[sqlite3.Row]: 130 return conn.execute( 131 "SELECT * FROM ingredients WHERE recipe_id = ?", (recipe_id,) 132 ).fetchall() 133 134 135 # --- staples --- 136 137 def add_staple(conn, name, default_quantity, unit, category) -> int: 138 cur = conn.execute( 139 "INSERT INTO staples (name, default_quantity, unit, category) " 140 "VALUES (?, ?, ?, ?) " 141 "ON CONFLICT(name) DO UPDATE SET " 142 "default_quantity=excluded.default_quantity, " 143 "unit=excluded.unit, category=excluded.category", 144 (name, default_quantity, unit, category), 145 ) 146 conn.commit() 147 return cur.lastrowid 148 149 150 def remove_staple(conn, name: str) -> None: 151 conn.execute("DELETE FROM staples WHERE name = ? COLLATE NOCASE", (name,)) 152 conn.commit() 153 154 155 def list_staples(conn) -> list[sqlite3.Row]: 156 return conn.execute("SELECT * FROM staples ORDER BY category, name").fetchall() 157 158 159 # --- grocery list --- 160 161 def add_grocery_item(conn, name, quantity, unit, category, source) -> int: 162 cur = conn.execute( 163 "INSERT INTO grocery_list_items (name, quantity, unit, category, source) " 164 "VALUES (?, ?, ?, ?, ?)", 165 (name, quantity, unit, category, source), 166 ) 167 conn.commit() 168 return cur.lastrowid 169 170 171 def list_grocery_items(conn) -> list[sqlite3.Row]: 172 return conn.execute( 173 "SELECT * FROM grocery_list_items ORDER BY category, name" 174 ).fetchall() 175 176 177 def clear_grocery_list(conn) -> None: 178 conn.execute("DELETE FROM grocery_list_items") 179 conn.commit()