Shopping list db functionality

db.py

@@ -18,3 +18,6 @@ async def create(conn: aiosqlite.Connection):
 
     import meals.db as meals_db
     await meals_db.create(conn)
+    
+    import shopping.db as shopping_db
+    await shopping_db.create(conn)

ingredients/db.py

@@ -33,7 +33,6 @@ async def create(conn):
             FOREIGN KEY (meal_id) REFERENCES Meal(id)
         );''')
 
-
 async def insert_ingredient(conn, ingredient: Ingredient):
     if not ingredient.product_id and ingredient.product:
         ingredient.product_id = ingredient.product.id

persons/db.py

@@ -1,7 +1,9 @@
 from pydantic import BaseModel
-from typing import AsyncIterator
+from typing import AsyncIterator, ClassVar, List
 
 class Person(BaseModel):
+    KEYS: ClassVar[List[str]] = ['id', 'name']
+
     id: int
     name: str
 

shopping/__init__.py

@@ -0,0 +1,2 @@
+from shopping.db import ShoppingList, ShoppingListRequest, ShoppingListResult, current_shopping_list, mark_found, sync_persons_requests
+

shopping/db.py

@@ -0,0 +1,261 @@
+from meals import Meal
+from ingredients import Ingredient, insert_ingredient
+from persons import Person
+from products import Product
+
+from pydantic import BaseModel
+from typing import AsyncIterator, List, ClassVar, Optional
+
+from datetime import datetime
+
+class ShoppingListRequest(BaseModel):
+    KEYS: ClassVar[List[str]] = ['id', 'ingredient_id', 'list_id', 'person_id', 'meal_id', 'created_date']
+    id: int = 0
+    list_id: int
+    
+    ingredient_id: int
+    ingredient: Optional[Ingredient] = None
+
+    person_id: Optional[int] = None
+    person: Optional[Person] = None
+
+    meal_id: Optional[int] = None
+    meal: Optional[Meal] = None
+
+    created_date: datetime = datetime.now()
+
+class ShoppingListResult(BaseModel):
+    KEYS: ClassVar[List[str]] = ['id', 'product_id', 'list_id', 'quantity', 'unit', 'created_date', 'found_date']
+    id: int = 0
+    product_id: int
+    list_id: int
+
+    quantity: float
+    unit: str
+
+    created_date: datetime = datetime.now()
+    found_date: Optional[datetime] = None
+
+class ShoppingList(BaseModel):
+    KEYS: ClassVar[List[str]] = ['id', 'created_date', 'purchased_date']
+    id: int = 0
+    created_date: datetime = datetime.now()
+    purchased_date: Optional[datetime] = None
+
+    requests: List[ShoppingListRequest] = []
+    items: List[ShoppingListResult] = []
+
+async def create(conn):
+    await conn.execute('''
+        CREATE TABLE IF NOT EXISTS ShoppingList (
+            id INTEGER PRIMARY KEY,
+            created_date TEXT,
+            purchased_date TEXT
+        );''')
+
+    await conn.execute('''
+        CREATE TABLE IF NOT EXISTS ShoppingListRequest (
+            id INTEGER PRIMARY KEY,
+            ingredient_id INTEGER,
+            list_id INTEGER,
+            person_id INTEGER,
+            meal_id INTEGER,
+            created_date TEXT,
+            FOREIGN KEY(ingredient_id) REFERENCES Ingredient(id),
+            FOREIGN KEY(person_id) REFERENCES Person(id),
+            FOREIGN KEY(meal_id) REFERENCES Meal(id),
+            FOREIGN KEY(list_id) REFERENCES ShoppingList(id)
+        );''')
+
+    await conn.execute('''
+        CREATE TABLE IF NOT EXISTS ShoppingListResult (
+            id INTEGER PRIMARY KEY,
+            product_id INTEGER,
+            list_id INTEGER,
+            quantity REAL,
+            unit TEXT,
+            created_date TEXT,
+            found_date TEXT,
+            FOREIGN KEY(product_id) REFERENCES Product(id),
+            FOREIGN KEY(list_id) REFERENCES ShoppingList(id)
+        );''')
+    
+async def insert_shopping_list(conn, shopping_list: ShoppingList):
+    async with conn.execute('''
+        INSERT INTO ShoppingList (id, created_date, purchased_date)
+        VALUES (?, ?, ?)
+    ''', (shopping_list.id, shopping_list.created_date, shopping_list.purchased_date)) as cursor:
+        shopping_list.id = cursor.lastrowid
+
+    for request in shopping_list.requests:
+        if not request.ingredient_id:
+            await insert_ingredient(conn, request.ingredient)
+
+        request.ingredient_id = request.ingredient.id
+        await conn.execute('''
+            INSERT INTO ShoppingListRequest (id, ingredient_id, list_id, person_id, meal_id, created_date)
+            VALUES (?, ?, ?, ?, ?, ?)
+        ''', (request.id, request.ingredient_id, shopping_list.id, request.person_id, request.meal_id, request.created_date))
+
+    for item in shopping_list.items:
+        item.product_id = item.product.id
+        await conn.execute('''
+            INSERT INTO ShoppingListResult (id, product_id, list_id, quantity, unit, created_date, found_date)
+            VALUES (?, ?, ?, ?, ?, ?, ?)
+        ''', (item.id, item.product_id, shopping_list.id, item.quantity, item.unit, item.created_date, item.found_date))
+
+async def find_requests_by_list_id(conn, list_id: int) -> AsyncIterator[ShoppingListRequest]:
+    # Join Ingredient and Product to also load ingredient and product
+    ingredient_keys = [f'ingredient.{key}' for key in Ingredient.KEYS]
+    product_keys = [f'product.{key}' for key in Product.KEYS]
+    request_keys = [f'shoppinglistrequest.{key}' for key in ShoppingListRequest.KEYS]
+    person_keys = [f'person.{key}' for key in Person.KEYS]
+
+    cursor = await conn.execute(f'''
+        SELECT {','.join(ingredient_keys + product_keys + request_keys + person_keys)}
+        FROM ShoppingListRequest
+        LEFT JOIN Ingredient ON ShoppingListRequest.ingredient_id = Ingredient.id
+        LEFT JOIN Product ON Ingredient.product_id = Product.id
+        LEFT JOIN Person ON ShoppingListRequest.person_id = Person.id
+        WHERE list_id = ?
+    ''', (list_id,))
+    
+    async for row in cursor:
+        product_keys = {k:v for k,v in zip(Product.KEYS, row[len(Ingredient.KEYS):len(Ingredient.KEYS) + len(Product.KEYS)])}
+        product = Product(**product_keys) if product_keys['id'] else None
+
+        ingredient_keys = {k:v for k,v in zip(Ingredient.KEYS, row[:len(Ingredient.KEYS)])}
+        ingredient = Ingredient(**ingredient_keys, product=product)
+
+        person_keys = {k:v for k,v in zip(Person.KEYS, row[-len(Person.KEYS):])}
+        person = Person(**person_keys) if person_keys['id'] else None
+
+        request_keys = {k:v for k,v in zip(ShoppingListRequest.KEYS, row[len(Ingredient.KEYS) + len(Product.KEYS):-len(Person.KEYS)])}
+        request = ShoppingListRequest(**request_keys, ingredient=ingredient, person=person)
+        yield request
+
+async def find_items_by_list_id(conn, list_id: int) -> AsyncIterator[ShoppingListResult]:
+    product_keys = [f'product.{key}' for key in Product.KEYS]
+    result_keys = [f'shoppinglistresult.{key}' for key in ShoppingListResult.KEYS]
+
+    async with conn.execute(f'''
+        SELECT {','.join(product_keys + result_keys)}
+        FROM ShoppingListResult
+        LEFT JOIN Product ON ShoppingListResult.product_id = Product.id
+        WHERE list_id = ?
+    ''', (list_id,)) as cursor:
+        async for row in cursor:
+            product_keys = {k:v for k,v in zip(Product.KEYS, row[:len(Product.KEYS)])}
+            product = Product(**product_keys) if product_keys['id'] else None
+
+            result_keys = {k:v for k,v in zip(ShoppingListResult.KEYS, row[len(Product.KEYS):])}
+            result = ShoppingListResult(**result_keys, product=product)
+            yield result
+
+async def fill_related(conn, shopping_list: ShoppingList) -> ShoppingList:
+    async for request in find_requests_by_list_id(conn, shopping_list.id):
+        shopping_list.requests.append(request)
+
+    async for item in find_items_by_list_id(conn, shopping_list.id):
+        shopping_list.items.append(item)
+        
+async def load_shopping_list(conn, id: int) -> ShoppingList:
+    shopping_list = None
+    async with conn.execute(f'''
+        SELECT {','.join(ShoppingList.KEYS)} FROM ShoppingList
+        WHERE id = ?
+        LIMIT 1
+    ''', (id,)) as cursor:
+        async for row in cursor:
+            shopping_list = ShoppingList(**{k:v for k,v in zip(ShoppingList.KEYS, row)})
+            break
+
+    if shopping_list:
+        await fill_related(conn, shopping_list)
+
+    return shopping_list
+
+async def current_shopping_list(conn) -> ShoppingList:
+    shopping_list = None
+    async with conn.execute(f'''
+        SELECT {','.join(ShoppingList.KEYS)} FROM ShoppingList
+        WHERE purchased_date IS NULL
+        LIMIT 1
+    ''') as cursor:
+        async for row in cursor:
+            shopping_list = ShoppingList(**{k:v for k,v in zip(ShoppingList.KEYS, row)})
+            await fill_related(conn, shopping_list)
+            break
+
+    if not shopping_list:
+        shopping_list = ShoppingList()
+        await insert_shopping_list(conn, shopping_list)
+
+    return shopping_list
+
+async def find_existing_result(conn, product: Product, shopping_list: ShoppingList) -> ShoppingListResult:
+    async with conn.execute(f'''
+        SELECT {','.join(ShoppingListResult.KEYS)} FROM ShoppingListResult
+        WHERE product_id = ? AND list_id = ?
+        LIMIT 1
+    ''', (product.id, shopping_list.id)) as cursor:
+        async for row in cursor:
+            return ShoppingListResult(**{k:v for k,v in zip(ShoppingListResult.KEYS, row)}, product=product)
+
+    return None
+
+async def get_persons_requests(conn, shopping_list: ShoppingList, person: Person) -> AsyncIterator[ShoppingListRequest]:
+    async for request in find_requests_by_list_id(conn, shopping_list.id):
+        if request.person_id == person.id:
+            yield request
+
+async def request_ingredient(conn, shopping_list: ShoppingList, person: Person, ingredient: Ingredient):
+    if ingredient.id:
+        raise ValueError('How did you get an existing ingredient?')
+    
+    await insert_ingredient(conn, ingredient)
+
+    request = ShoppingListRequest(ingredient_id=ingredient.id, list_id=shopping_list.id, person_id=person.id, created_date=datetime.now())
+    async with conn.execute('''
+        INSERT INTO ShoppingListRequest (ingredient_id, list_id, person_id, created_date)
+        VALUES (?, ?, ?, ?)
+    ''', (request.ingredient_id, request.list_id, request.person_id, request.created_date)) as cursor:
+        request.id = cursor.lastrowid
+
+    shopping_list.requests.append(request)
+
+async def sync_persons_requests(conn, shopping_list: ShoppingList, person: Person, requests: List[Ingredient]):
+    # Delete existing and insert all as new
+    async for request in get_persons_requests(conn, shopping_list, person):
+        await conn.execute('''
+            DELETE FROM ShoppingListRequest
+            WHERE id = ?
+        ''', (request.id,))
+
+    for ingredient in requests:
+        ingredient.id = 0
+        await request_ingredient(conn, shopping_list, person, ingredient)
+    
+async def mark_found(conn, product: Product, quantity: float, unit: str) -> ShoppingListResult:
+    shopping_list = await current_shopping_list(conn)
+    existing = await find_existing_result(conn, product, shopping_list)
+
+    if existing:
+        existing.quantity, existing.unit, existing.found_date = quantity, unit, datetime.now()
+
+        await conn.execute('''
+            UPDATE ShoppingListResult
+            SET quantity = ?, unit = ?, found_date = ?
+            WHERE id = ?
+        ''', (existing.quantity, existing.unit, existing.found_date, existing.id))
+
+        return existing
+    else:
+        result = ShoppingListResult(product=product, product_id=product.id, list_id=shopping_list.id, quantity=quantity, unit=unit, found_date=datetime.now())
+        async with conn.execute('''
+            INSERT INTO ShoppingListResult (product_id, list_id, quantity, unit, created_date, found_date)
+            VALUES (?, ?, ?, ?, ?, ?)
+        ''', (result.product_id, result.list_id, quantity, unit, result.created_date, result.found_date)) as cursor:
+            result.id = cursor.lastrowid
+        shopping_list.items.append(result)
+        return result

tests/test_shopping.py

@@ -0,0 +1,95 @@
+import importlib
+import unittest
+import tests.test_data as test_data
+import shopping
+
+import ingredients, products
+
+from db import connect, create
+
+def reload_test_data():
+    global test_data
+    test_data = importlib.reload(test_data)
+
+def first(iterable: list, predicate: callable):
+    for item in iterable:
+        if predicate(item):
+            return item
+    return None
+
+class TestShopping(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.conn = await connect(':memory:')
+        await create(self.conn)
+        reload_test_data()
+        return await super().asyncSetUp()
+
+    async def asyncTearDown(self) -> None:
+        await self.conn.close()
+        return await super().asyncTearDown()
+
+    async def test_current_shopping_list(self):
+        shopping_list = await shopping.current_shopping_list(self.conn)
+        self.assertEqual(len(shopping_list.requests), 0)
+        self.assertEqual(len(shopping_list.items), 0)
+
+    async def test_sync_persons_requests(self):
+        ingredient = test_data.Ingredients.one_apple
+        person = test_data.Persons.jacob
+        
+        await products.insert_product(self.conn, ingredient.product, {})
+        
+        shopping_list = await shopping.current_shopping_list(self.conn)
+        await shopping.sync_persons_requests(self.conn, shopping_list, person, [ingredient])
+
+        shopping_list = await shopping.current_shopping_list(self.conn)
+        self.assertEqual(len(shopping_list.requests), 1)
+        self.assertEqual(len(shopping_list.items), 0)
+
+        request = shopping_list.requests[0]
+        self.assertEqual(request.person_id, person.id)
+        self.assertEqual(request.ingredient.line, ingredient.line)
+
+    async def test_sync_persons_requests_multiple_add_item(self):
+        first = test_data.Ingredients.one_apple
+        second = test_data.Ingredients.salt
+
+        person = test_data.Persons.jacob
+        
+        await products.insert_product(self.conn, first.product, {})
+        await products.insert_product(self.conn, second.product, {})
+        
+        shopping_list = await shopping.current_shopping_list(self.conn)
+        await shopping.sync_persons_requests(self.conn, shopping_list, person, [first])
+        await shopping.sync_persons_requests(self.conn, shopping_list, person, [first, second])
+
+        shopping_list = await shopping.current_shopping_list(self.conn)
+        self.assertEqual(len(shopping_list.requests), 2)
+        self.assertEqual(len(shopping_list.items), 0)
+
+        request_by_line = {r.ingredient.line: r for r in shopping_list.requests}
+        self.assertEqual(len(request_by_line), 2)
+        
+        for requested_ingredient in [first, second]:
+            request = request_by_line[requested_ingredient.line]
+            self.assertEqual(request.person_id, person.id)
+            self.assertEqual(request.ingredient.line, requested_ingredient.line)
+
+    async def test_mark_found(self):
+        ingredient = test_data.Ingredients.one_apple
+        person = test_data.Persons.jacob
+        
+        await products.insert_product(self.conn, ingredient.product, {})
+        
+        shopping_list = await shopping.current_shopping_list(self.conn)
+        await shopping.sync_persons_requests(self.conn, shopping_list, person, [ingredient])
+        await shopping.mark_found(self.conn, ingredient.product, 2, 'items')
+
+        shopping_list = await shopping.current_shopping_list(self.conn)
+        self.assertEqual(len(shopping_list.requests), 1)
+
+        self.assertEqual(len(shopping_list.items), 1)
+        self.assertEqual(shopping_list.items[0].product_id, ingredient.product.id)
+        self.assertEqual(shopping_list.items[0].quantity, 2)
+        self.assertEqual(shopping_list.items[0].unit, 'items')
+