forked from write-magic-code/python_base_final
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsqlite.py
More file actions
98 lines (87 loc) · 4.23 KB
/
sqlite.py
File metadata and controls
98 lines (87 loc) · 4.23 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
import sqlite3
import json
import os
from .tables import WarehouseData, OrderData
from typing import List
class DB:
def __init__(self):
# 如果数据库不存在，则创建数据库
if not os.path.exists('warehouse.db'):
with open('warehouse.db', 'w'):
pass
self.conn = sqlite3.connect('warehouse.db', check_same_thread=False)
self.cursor = self.conn.cursor()
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS warehouse_data (
wid INTEGER PRIMARY KEY,
name TEXT NOT NULL,
tel TEXT NOT NULL,
address TEXT NOT NULL,
selling_price TEXT NOT NULL,
transport_price TEXT NOT NULL
);
''')
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS order_data (
order_id INTEGER PRIMARY KEY,
wid INTEGER NOT NULL,
date TEXT NOT NULL,
transport_volume TEXT NOT NULL,
selling_volume TEXT NOT NULL,
FOREIGN KEY (wid) REFERENCES warehouse_data(wid)
);
''')
self.conn.commit()
def add_warehouse(self, warehouse: WarehouseData) -> int:
self.cursor.execute('INSERT INTO warehouse_data (wid, name, tel, address, selling_price, transport_price) VALUES (?, ?, ?, ?, ?, ?)',
(warehouse.wid, warehouse.name, warehouse.tel, warehouse.address, json.dumps(warehouse.selling_price), json.dumps(warehouse.transport_price)))
self.conn.commit()
return self.cursor.lastrowid
def get_warehouse(self, condition: str) -> List[WarehouseData]:
self.cursor.execute(f'SELECT * FROM warehouse_data WHERE {condition}')
rows = self.cursor.fetchall()
result: List[WarehouseData] = []
for row in rows:
result.append(WarehouseData(
wid = row[0],
name = row[1],
tel = row[2],
address = row[3],
selling_price = json.loads(row[4]),
transport_price = json.loads(row[5]),
))
return result
def update_warehouse(self, warehouse: WarehouseData) -> int:
self.cursor.execute('UPDATE warehouse_data SET name = ?, tel = ?, address = ?, selling_price = ?, transport_price = ? WHERE wid = ?', (warehouse.name, warehouse.tel, warehouse.address, json.dumps(warehouse.selling_price), json.dumps(warehouse.transport_price), warehouse.wid))
self.conn.commit()
return self.cursor.rowcount
def delete_warehouse(self, condition: str) -> int:
self.cursor.execute(f'DELETE FROM warehouse_data WHERE {condition}')
self.conn.commit()
return self.cursor.rowcount
def add_order(self, order: OrderData) -> int:
self.cursor.execute('INSERT INTO order_data (order_id, wid, date, transport_volume, selling_volume) VALUES (?, ?, ?, ?, ?)', (order.order_id, order.wid, order.date, json.dumps(order.transport_volume), json.dumps(order.selling_volume)))
self.conn.commit()
return self.cursor.lastrowid
def get_order(self, condition: str) -> List[OrderData]:
self.cursor.execute(f'SELECT * FROM order_data WHERE {condition}')
rows = self.cursor.fetchall()
result: List[OrderData] = []
for row in rows:
result.append(OrderData(
order_id = row[0],
wid = row[1],
date = row[2],
transport_volume = json.loads(row[3]),
selling_volume = json.loads(row[4])
))
return result
def update_order(self, order: OrderData) -> int:
self.cursor.execute('UPDATE order_data SET wid = ?, date = ?, transport_volume = ?, selling_volume = ? WHERE order_id = ?', (order.wid, order.date, json.dumps(order.transport_volume), json.dumps(order.selling_volume), order.order_id))
self.conn.commit()
return self.cursor.rowcount
def delete_order(self, condition: str) -> int:
self.cursor.execute(f'DELETE FROM order_data WHERE {condition}')
self.conn.commit()
return self.cursor.rowcount
db = DB()