Working 'found'
This commit is contained in:
parent
be1d2b75ad
commit
6b684fadd8
4 changed files with 119 additions and 44 deletions
24
main.py
24
main.py
|
|
@ -236,9 +236,29 @@ async def get_shopping_list(list_id: Union[int, str], conn: sqlite3.Connection =
|
|||
|
||||
return await shopping.load_shopping_list(conn, list_id)
|
||||
|
||||
class FoundResult(BaseModel):
|
||||
created: List[shopping.ShoppingListResult] = []
|
||||
removed: List[shopping.ShoppingListResult] = []
|
||||
|
||||
@app.post("/shopping/current/found")
|
||||
async def mark_shopping_list(ingredient: ingredients.Ingredient, conn: sqlite3.Connection = Depends(get_db)) -> shopping.ShoppingListResult:
|
||||
return await shopping.mark_found(conn, ingredient.product, ingredient.quantity, ingredient.unit)
|
||||
async def mark_shopping_list(ingredients: List[ingredients.Ingredient], conn: sqlite3.Connection = Depends(get_db)) -> FoundResult:
|
||||
now = datetime.datetime.now()
|
||||
result = FoundResult()
|
||||
for ingredient in ingredients:
|
||||
existing, created = await shopping.mark_found(conn, ingredient, now)
|
||||
result.created.append(created)
|
||||
|
||||
if existing:
|
||||
result.removed.append(existing)
|
||||
|
||||
await conn.commit()
|
||||
return result
|
||||
|
||||
@app.delete("/shopping/current/found/{product_id}")
|
||||
async def unmark_shopping_list(product_id: int, conn: sqlite3.Connection = Depends(get_db)) -> List[shopping.ShoppingListResult]:
|
||||
response = await shopping.unmark_found(conn, product_id)
|
||||
await conn.commit()
|
||||
return response
|
||||
|
||||
@app.get("/shopping/current/me/ingredients")
|
||||
async def get_my_shopping_list(conn: sqlite3.Connection = Depends(get_db), person: persons.Person = Depends(cookie_person)) -> List[shopping.ShoppingListRequest]:
|
||||
|
|
|
|||
|
|
@ -1,2 +1,2 @@
|
|||
from shopping.db import ShoppingList, ShoppingListRequest, ShoppingListResult, current_shopping_list, mark_found, sync_persons_requested_ingredients, load_shopping_list, get_persons_requests, request_meal, delete_requests
|
||||
from shopping.db import ShoppingList, ShoppingListRequest, ShoppingListResult, current_shopping_list, mark_found, unmark_found, sync_persons_requested_ingredients, load_shopping_list, get_persons_requests, request_meal, delete_requests
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ from persons import Person
|
|||
from products import Product
|
||||
|
||||
from pydantic import BaseModel
|
||||
from typing import AsyncIterator, List, ClassVar, Optional
|
||||
from typing import AsyncIterator, List, ClassVar, Optional, Tuple
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
|
@ -27,8 +27,9 @@ class ShoppingListRequest(BaseModel):
|
|||
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
|
||||
product_id: int
|
||||
product: Optional[Product] = None
|
||||
|
||||
quantity: float
|
||||
unit: str
|
||||
|
|
@ -43,7 +44,7 @@ class ShoppingList(BaseModel):
|
|||
purchased_date: Optional[datetime] = None
|
||||
|
||||
requests: List[ShoppingListRequest] = []
|
||||
items: List[ShoppingListResult] = []
|
||||
results: List[ShoppingListResult] = []
|
||||
|
||||
async def create(conn):
|
||||
await conn.execute('''
|
||||
|
|
@ -97,7 +98,7 @@ async def insert_shopping_list(conn, shopping_list: ShoppingList):
|
|||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
''', (request.id, request.ingredient_id, shopping_list.id, request.person_id, request.meal_id, request.created_date))
|
||||
|
||||
for item in shopping_list.items:
|
||||
for item in shopping_list.results:
|
||||
item.product_id = item.product.id
|
||||
await conn.execute('''
|
||||
INSERT INTO ShoppingListResult (id, product_id, list_id, quantity, unit, created_date, found_date)
|
||||
|
|
@ -106,13 +107,13 @@ async def insert_shopping_list(conn, shopping_list: ShoppingList):
|
|||
|
||||
async def find_request(conn, id: int) -> Optional[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]
|
||||
ingredient_keys = [f'ingredient.{key}' for key in Ingredient.KEYS]
|
||||
person_keys = [f'person.{key}' for key in Person.KEYS]
|
||||
request_keys = [f'shoppinglistrequest.{key}' for key in ShoppingListRequest.KEYS]
|
||||
|
||||
async with conn.execute(f'''
|
||||
SELECT {','.join(ingredient_keys + product_keys + request_keys + person_keys)}
|
||||
SELECT {','.join(product_keys + ingredient_keys + person_keys + request_keys)}
|
||||
FROM ShoppingListRequest
|
||||
LEFT JOIN Ingredient ON ShoppingListRequest.ingredient_id = Ingredient.id
|
||||
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
||||
|
|
@ -120,17 +121,21 @@ async def find_request(conn, id: int) -> Optional[ShoppingListRequest]:
|
|||
WHERE ShoppingListRequest.id = ?
|
||||
''', (id,)) as cursor:
|
||||
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_keys = {k:v for k,v in zip(Product.KEYS, row[: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)
|
||||
ingredient_keys = {k:v for k,v in zip(Ingredient.KEYS, row[len(Product.KEYS):len(Product.KEYS) + len(Ingredient.KEYS)])}
|
||||
ingredient = Ingredient(**ingredient_keys, product=product) if ingredient_keys['id'] else None
|
||||
|
||||
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_keys = {k:v for k,v in zip(ShoppingListRequest.KEYS, row[len(Product.KEYS) + len(Ingredient.KEYS):-len(Person.KEYS)])}
|
||||
request = ShoppingListRequest(**request_keys, ingredient=ingredient, person=person)
|
||||
|
||||
if request.meal_id:
|
||||
request.meal = await find_meal_by_id(conn, request.meal_id)
|
||||
|
||||
return request
|
||||
|
||||
async def find_requests_by_list_id(conn, list_id: int) -> AsyncIterator[ShoppingListRequest]:
|
||||
|
|
@ -190,7 +195,7 @@ async def fill_related(conn, shopping_list: ShoppingList) -> ShoppingList:
|
|||
shopping_list.requests.append(request)
|
||||
|
||||
async for item in find_items_by_list_id(conn, shopping_list.id):
|
||||
shopping_list.items.append(item)
|
||||
shopping_list.results.append(item)
|
||||
|
||||
async def load_shopping_list(conn, id: int) -> ShoppingList:
|
||||
shopping_list = None
|
||||
|
|
@ -296,26 +301,59 @@ async def sync_persons_requested_ingredients(conn, shopping_list: ShoppingList,
|
|||
ingredient.id = 0
|
||||
yield await request_ingredient(conn, shopping_list, person, ingredient)
|
||||
|
||||
async def mark_found(conn, product: Product, quantity: float, unit: str) -> ShoppingListResult:
|
||||
async def find_existing_result(conn, shopping_list: ShoppingList, product: Product, unit: str) -> Optional[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 ShoppingListResult.list_id = ? AND ShoppingListResult.product_id = ? AND ShoppingListResult.unit = ?
|
||||
LIMIT 1
|
||||
''', (shopping_list.id, product.id, unit)) 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):])}
|
||||
return ShoppingListResult(**result_keys, product=product)
|
||||
|
||||
return None
|
||||
|
||||
async def mark_found(conn, ingredient: Ingredient, date_found: datetime) -> Tuple[ShoppingListResult, ShoppingListResult]:
|
||||
product, quantity, unit = ingredient.product, ingredient.quantity, ingredient.unit
|
||||
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()
|
||||
created = ShoppingListResult(product=product, product_id=product.id, list_id=shopping_list.id, quantity=quantity, unit=unit, found_date=date_found)
|
||||
removed = await find_existing_result(conn, shopping_list, product, unit)
|
||||
|
||||
async with conn.execute('''
|
||||
INSERT INTO ShoppingListResult (product_id, list_id, quantity, unit, created_date, found_date)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
''', (created.product_id, created.list_id, quantity, unit, created.created_date, created.found_date)) as cursor:
|
||||
created.id = cursor.lastrowid
|
||||
|
||||
if removed:
|
||||
shopping_list.results = [r for r in shopping_list.results if r.id != removed.id]
|
||||
await conn.execute('''
|
||||
UPDATE ShoppingListResult
|
||||
SET quantity = ?, unit = ?, found_date = ?
|
||||
DELETE FROM ShoppingListResult
|
||||
WHERE id = ?
|
||||
''', (existing.quantity, existing.unit, existing.found_date, existing.id))
|
||||
''', (removed.id,))
|
||||
|
||||
return removed, created
|
||||
|
||||
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
|
||||
async def unmark_found(conn, product_id: int) -> List[ShoppingListRequest]:
|
||||
shopping_list = await current_shopping_list(conn)
|
||||
|
||||
requests = []
|
||||
async with conn.execute('''
|
||||
DELETE FROM ShoppingListResult
|
||||
WHERE list_id = ? AND product_id = ?
|
||||
''', (shopping_list.id, product_id)) as cursor:
|
||||
async for row in cursor:
|
||||
requests.append(row)
|
||||
|
||||
deleted = [r for r in shopping_list.results if r.product_id == product_id]
|
||||
shopping_list.results = [r for r in shopping_list.results if r.product_id != product_id]
|
||||
return deleted
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
from datetime import datetime
|
||||
import importlib
|
||||
import unittest
|
||||
import tests.test_data as test_data
|
||||
|
|
@ -31,7 +32,7 @@ class TestShopping(unittest.IsolatedAsyncioTestCase):
|
|||
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)
|
||||
self.assertEqual(len(shopping_list.results), 0)
|
||||
|
||||
async def test_sync_persons_requests(self):
|
||||
ingredient = test_data.Ingredients.one_apple
|
||||
|
|
@ -44,7 +45,7 @@ class TestShopping(unittest.IsolatedAsyncioTestCase):
|
|||
|
||||
shopping_list = await shopping.current_shopping_list(self.conn)
|
||||
self.assertEqual(len(shopping_list.requests), 1)
|
||||
self.assertEqual(len(shopping_list.items), 0)
|
||||
self.assertEqual(len(shopping_list.results), 0)
|
||||
|
||||
request = shopping_list.requests[0]
|
||||
self.assertEqual(request.person_id, person.id)
|
||||
|
|
@ -65,7 +66,7 @@ class TestShopping(unittest.IsolatedAsyncioTestCase):
|
|||
|
||||
shopping_list = await shopping.current_shopping_list(self.conn)
|
||||
self.assertEqual(len(shopping_list.requests), 2)
|
||||
self.assertEqual(len(shopping_list.items), 0)
|
||||
self.assertEqual(len(shopping_list.results), 0)
|
||||
|
||||
request_by_line = {r.ingredient.line: r for r in shopping_list.requests}
|
||||
self.assertEqual(len(request_by_line), 2)
|
||||
|
|
@ -77,19 +78,35 @@ class TestShopping(unittest.IsolatedAsyncioTestCase):
|
|||
|
||||
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_requested_ingredients(self.conn, shopping_list, person, [ingredient])
|
||||
await shopping.mark_found(self.conn, ingredient.product, 2, 'items')
|
||||
await shopping.mark_found(self.conn, ingredient, date_found=datetime.now())
|
||||
|
||||
shopping_list = await shopping.current_shopping_list(self.conn)
|
||||
self.assertEqual(len(shopping_list.requests), 1)
|
||||
self.assertEqual(len(shopping_list.results), 1)
|
||||
self.assertEqual(shopping_list.results[0].product_id, ingredient.product.id)
|
||||
self.assertEqual(shopping_list.results[0].quantity, 1)
|
||||
self.assertEqual(shopping_list.results[0].unit, ingredient.unit)
|
||||
|
||||
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')
|
||||
|
||||
ingredient.quantity = 2
|
||||
await shopping.mark_found(self.conn, ingredient, date_found=datetime.now())
|
||||
|
||||
shopping_list = await shopping.current_shopping_list(self.conn)
|
||||
self.assertEqual(len(shopping_list.results), 1)
|
||||
self.assertEqual(shopping_list.results[0].product_id, ingredient.product.id)
|
||||
self.assertEqual(shopping_list.results[0].quantity, 2)
|
||||
self.assertEqual(shopping_list.results[0].unit, ingredient.unit)
|
||||
|
||||
ingredient.unit = 'kg'
|
||||
await shopping.mark_found(self.conn, ingredient, date_found=datetime.now())
|
||||
|
||||
shopping_list = await shopping.current_shopping_list(self.conn)
|
||||
self.assertEqual(len(shopping_list.results), 2)
|
||||
|
||||
results_by_unit = {r.unit: r for r in shopping_list.results}
|
||||
self.assertEqual(len(results_by_unit), 2)
|
||||
self.assertIn('kg', results_by_unit)
|
||||
self.assertIn('Items', results_by_unit)
|
||||
self.assertEqual(results_by_unit['kg'].product_id, results_by_unit['Items'].product_id)
|
||||
Loading…
Reference in a new issue