56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
|
import sqlite3
|
||
|
|
import aiosqlite
|
||
|
|
|
||
|
|
from model import *
|
||
|
|
from typing import List
|
||
|
|
|
||
|
|
async def create():
|
||
|
|
conn = await connect()
|
||
|
|
await conn.execute('''
|
||
|
|
CREATE TABLE IF NOT EXISTS Product (
|
||
|
|
id INTEGER PRIMARY KEY,
|
||
|
|
name TEXT,
|
||
|
|
link TEXT UNIQUE,
|
||
|
|
img_small TEXT,
|
||
|
|
img_large TEXT,
|
||
|
|
raw_data TEXT
|
||
|
|
);''')
|
||
|
|
|
||
|
|
await conn.execute('''
|
||
|
|
CREATE TABLE IF NOT EXISTS ProductTag (
|
||
|
|
food_item_id INTEGER,
|
||
|
|
tag TEXT,
|
||
|
|
PRIMARY KEY (food_item_id, tag),
|
||
|
|
FOREIGN KEY (food_item_id) REFERENCES Product(id)
|
||
|
|
);''')
|
||
|
|
|
||
|
|
# Commit changes and close the connection
|
||
|
|
await conn.commit()
|
||
|
|
await conn.close()
|
||
|
|
|
||
|
|
async def find_product_by_tag(conn, tag: str) -> List[Product]:
|
||
|
|
async with conn.execute('''
|
||
|
|
SELECT * FROM Product
|
||
|
|
WHERE id IN (
|
||
|
|
SELECT food_item_id FROM ProductTag
|
||
|
|
WHERE tag = ?
|
||
|
|
)
|
||
|
|
''', (tag,)) as cursor:
|
||
|
|
async for row in cursor:
|
||
|
|
yield Product(*row)
|
||
|
|
|
||
|
|
async def insert_product(conn, product: Product):
|
||
|
|
async with conn.execute('''
|
||
|
|
INSERT INTO Product (name, link, img_small, img_large, raw_data)
|
||
|
|
VALUES (?, ?, ?, ?, ?)
|
||
|
|
''', (product.name, product.link, product.img_small, product.img_large, product.raw_data)) as cursor:
|
||
|
|
product.id = cursor.lastrowid
|
||
|
|
|
||
|
|
await conn.commit()
|
||
|
|
|
||
|
|
async def connect():
|
||
|
|
return await aiosqlite.connect('your_database.db')
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
import asyncio
|
||
|
|
asyncio.run(create())
|