Python+SQLite搭建仓库管理数据库

Word count: 2.4k   |   Reading time≈ 11 min

一、启动方法

1, 在python环境中安装sqlite3

1
pip install sqlite3

2, 运行代码

二、步骤1：创建SQLite数据库表结构

SQLite是轻量级数据库，不用装服务，直接用Python的sqlite3库操作。先写表结构（注意SQLite的类型适配）：

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
99
100
101
102
103
104
105
106
107
108
109
110
111
import sqlite3

conn = sqlite3.connect('warehouse.db')
conn.execute("PRAGMA foreign_keys = ON;")
cursor = conn.cursor()

# 1. 仓库表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 仓库 (
    仓库号 CHAR(6) PRIMARY KEY,
    城市 CHAR(10),
    面积 INTEGER
)
''')

# 2. 职工表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 职工 (
    职工号 CHAR(8) PRIMARY KEY,
    仓库号 CHAR(6),
    姓名 CHAR(10),
    工资 REAL,
    班组长 CHAR(8),
    FOREIGN KEY (仓库号) REFERENCES 仓库(仓库号),
    FOREIGN KEY (班组长) REFERENCES 职工(职工号)
)
''')

# 3. 器件表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 器件 (
    器件号 CHAR(8) PRIMARY KEY,
    器件名称 CHAR(20),
    规格 CHAR(16),
    单价 REAL
)
''')

# 4. 库存表（关键：外键关联仓库和器件）
cursor.execute('''
CREATE TABLE IF NOT EXISTS 库存 (
    器件号 CHAR(8),
    仓库号 CHAR(6),
    数量 INTEGER,
    PRIMARY KEY (器件号, 仓库号),
    FOREIGN KEY (器件号) REFERENCES 器件(器件号),
    FOREIGN KEY (仓库号) REFERENCES 仓库(仓库号)
)
''')

# 5. 供应商表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 供应商 (
    供应商号 CHAR(8) PRIMARY KEY,
    供应商名 CHAR(30),
    地址 CHAR(30)
)
''')

# 6. 供应表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 供应 (
    供应商号 CHAR(8),
    器件号 CHAR(8),
    PRIMARY KEY (供应商号, 器件号),
    FOREIGN KEY (供应商号) REFERENCES 供应商(供应商号),
    FOREIGN KEY (器件号) REFERENCES 器件(器件号)
)
''')

# 7. 订购单表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 订购单 (
    订购单号 CHAR(8) PRIMARY KEY,
    经手人 CHAR(8),
    供货方 CHAR(8),
    订货日期 DATETIME,
    金额 REAL,
    FOREIGN KEY (经手人) REFERENCES 职工(职工号),
    FOREIGN KEY (供货方) REFERENCES 供应商(供应商号)
)
''')

# 8. 订购明细表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 订购明细 (
    订购单号 CHAR(8),
    序号 INTEGER,
    器件号 CHAR(8),
    数量 INTEGER,
    单价 REAL,
    PRIMARY KEY (订购单号, 序号),
    FOREIGN KEY (订购单号) REFERENCES 订购单(订购单号),
    FOREIGN KEY (器件号) REFERENCES 器件(器件号)
)
''')

# 9. 保管表
cursor.execute('''
CREATE TABLE IF NOT EXISTS 保管 (
    职工号 CHAR(8),
    器件号 CHAR(8),
    PRIMARY KEY (职工号, 器件号),
    FOREIGN KEY (职工号) REFERENCES 职工(职工号),
    FOREIGN KEY (器件号) REFERENCES 器件(器件号)
)
''')

conn.commit()
conn.close()
print("表结构创建成功！")

三、步骤2：批量插入测试数据

每个表至少插10条数据，用随机数模拟真实业务场景：

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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
# 插入数据代码（保存为 add_data.py 并运行）
import sqlite3
import random
from datetime import datetime, timedelta

conn = sqlite3.connect('warehouse.db')
conn.execute("PRAGMA foreign_keys = ON;") #sqlite通过这种方式打开外键约束
cursor = conn.cursor()

# ---------------------- 1. 插入仓库表（10条，确保主键唯一） ----------------------
warehouse_data = [
    ('CK0001', '北京', 1000), ('CK0002', '上海', 1200), ('CK0003', '广州', 800),
    ('CK0004', '深圳', 1500), ('CK0005', '杭州', 900), ('CK0006', '成都', 1100),
    ('CK0007', '武汉', 700), ('CK0008', '南京', 850), ('CK0009', '西安', 950),
    ('CK0010', '重庆', 1300)
]
cursor.executemany("INSERT INTO 仓库 VALUES (?, ?, ?)", warehouse_data)
valid_warehouse_ids = [wh[0] for wh in warehouse_data]  # 直接从插入的数据中获取仓库号


# ---------------------- 2. 插入职工表（12条） ----------------------
staff_data = [
    ('ZG0001', 'CK0001', '张三', 8000.00, None), ('ZG0002', 'CK0001', '李四', 6500.00, 'ZG0001'),
    ('ZG0003', 'CK0002', '王五', 9000.00, None), ('ZG0004', 'CK0002', '赵六', 7200.00, 'ZG0003'),
    ('ZG0005', 'CK0003', '孙七', 7500.00, None), ('ZG0006', 'CK0003', '周八', 6800.00, 'ZG0005'),
    ('ZG0007', 'CK0004', '吴九', 8200.00, None), ('ZG0008', 'CK0004', '郑十', 6900.00, 'ZG0007'),
    ('ZG0009', 'CK0005', '钱一', 7800.00, None), ('ZG0010', 'CK0005', '冯二', 6300.00, 'ZG0009'),
    ('ZG0011', 'CK0006', '陈三', 8500.00, None), ('ZG0012', 'CK0006', '褚四', 7100.00, 'ZG0011')
]
cursor.executemany("INSERT INTO 职工 VALUES (?, ?, ?, ?, ?)", staff_data)
valid_staff_ids = [st[0] for st in staff_data]  # 从插入的数据中获取职工号


# ---------------------- 3. 插入器件表（10条） ----------------------
device_data = [
    ('QJ0001', '电阻', '10Ω', 2.5), ('QJ0002', '电容', '100μF', 3.2),
    ('QJ0003', '电感', '1mH', 5.8), ('QJ0004', '芯片', 'STM32', 28.5),
    ('QJ0005', '二极管', 'IN4007', 1.8), ('QJ0006', '三极管', '9013', 2.1),
    ('QJ0007', '继电器', '5V', 12.3), ('QJ0008', '晶振', '16MHz', 8.6),
    ('QJ0009', '保险丝', '2A', 4.5), ('QJ0010', '连接器', 'USB', 6.7)
]
cursor.executemany("INSERT INTO 器件 VALUES (?, ?, ?, ?)", device_data)
valid_device_ids = [dv[0] for dv in device_data]  # 从插入的数据中获取器件号


# ---------------------- 4. 插入库存表（15条，确保外键有效且无重复） ----------------------
inventory_data = []
used_inventory = set()
while len(inventory_data) < 15:
    device_id = random.choice(valid_device_ids)
    warehouse_id = random.choice(valid_warehouse_ids)
    if (device_id, warehouse_id) not in used_inventory:
        used_inventory.add((device_id, warehouse_id))
        inventory_data.append((device_id, warehouse_id, random.randint(100, 1000)))
cursor.executemany("INSERT INTO 库存 VALUES (?, ?, ?)", inventory_data)


# ---------------------- 5. 插入供应商表（10条） ----------------------
supplier_data = [
    ('GY0001', '华信电子', '北京海淀'), ('GY0002', '联创科技', '上海浦东'),
    ('GY0003', '南方电子', '广州天河'), ('GY0004', '深科技', '深圳南山'),
    ('GY0005', '浙电元件', '杭州西湖'), ('GY0006', '川渝电子', '成都武侯'),
    ('GY0007', '华中器件', '武汉光谷'), ('GY0008', '苏南科技', '南京江宁'),
    ('GY0009', '西北电子', '西安雁塔'), ('GY0010', '山城元件', '重庆渝中')
]
cursor.executemany("INSERT INTO 供应商 VALUES (?, ?, ?)", supplier_data)
valid_supplier_ids = [sp[0] for sp in supplier_data]  # 从插入的数据中获取供应商号


# ---------------------- 6. 插入供应表（12条） ----------------------
supply_data = []
used_supply = set()
while len(supply_data) < 12:
    supplier_id = random.choice(valid_supplier_ids)
    device_id = random.choice(valid_device_ids)
    if (supplier_id, device_id) not in used_supply:
        used_supply.add((supplier_id, device_id))
        supply_data.append((supplier_id, device_id))
cursor.executemany("INSERT INTO 供应 VALUES (?, ?)", supply_data)


# ---------------------- 7. 插入订购单表（10条） ----------------------
order_data = []
order_ids = [f'DG000{i}' for i in range(1, 11)]
for idx in range(10):
    order_date = (datetime.now() - timedelta(days=random.randint(1, 30))).strftime('%Y-%m-%d %H:%M:%S')
    order_data.append((
        order_ids[idx],
        random.choice(valid_staff_ids),
        random.choice(valid_supplier_ids),
        order_date,
        round(random.uniform(500, 5000), 2)
    ))
cursor.executemany("INSERT INTO 订购单 VALUES (?, ?, ?, ?, ?)", order_data)


# ---------------------- 8. 插入订购明细表（20条） ----------------------
order_detail_data = []
used_order_detail = set()
for order_id in order_ids:
    for seq in range(1, random.randint(2, 4)):  # 每个订单2-3条明细
        if (order_id, seq) not in used_order_detail:
            used_order_detail.add((order_id, seq))
            order_detail_data.append((
                order_id,
                seq,
                random.choice(valid_device_ids),
                random.randint(10, 50),
                round(random.uniform(2, 30), 2)
            ))
cursor.executemany("INSERT INTO 订购明细 VALUES (?, ?, ?, ?, ?)", order_detail_data)


# ---------------------- 9. 插入保管表（15条） ----------------------
keep_data = []
used_keep = set()
while len(keep_data) < 15:
    staff_id = random.choice(valid_staff_ids)
    device_id = random.choice(valid_device_ids)
    if (staff_id, device_id) not in used_keep:
        used_keep.add((staff_id, device_id))
        keep_data.append((staff_id, device_id))
cursor.executemany("INSERT INTO 保管 VALUES (?, ?)", keep_data)


conn.commit()
conn.close()
print("所有数据插入成功！")

四、步骤3：常用查询示例

搭好库后，就能做各种业务查询了，比如：

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
import sqlite3

conn = sqlite3.connect('warehouse.db')
conn.execute("PRAGMA foreign_keys = ON;")
cursor = conn.cursor()


# 1. 查询“北京仓库”的所有职工信息
print("=== 北京仓库的职工 ===")
cursor.execute('''
SELECT 职工.职工号, 职工.姓名, 职工.工资, 仓库.城市
FROM 职工 JOIN 仓库 ON 职工.仓库号 = 仓库.仓库号
WHERE 仓库.城市 = '北京'
''')
for row in cursor.fetchall():
    print(row)


# 2. 查询“器件QJ0004（STM32）”的库存分布（哪个仓库有多少）
print("\n=== 器件QJ0004的库存分布 ===")
cursor.execute('''
SELECT 仓库.仓库号, 仓库.城市, 库存.数量
FROM 库存 JOIN 仓库 ON 库存.仓库号 = 仓库.仓库号
WHERE 库存.器件号 = 'QJ0004'
''')
for row in cursor.fetchall():
    print(row)


# 3. 查询“订购单DG0001”的所有明细（包含器件名称、数量、单价）
print("\n=== 订购单DG0001的明细 ===")
cursor.execute('''
SELECT 订购明细.序号, 器件.器件名称, 订购明细.数量, 订购明细.单价
FROM 订购明细 JOIN 器件 ON 订购明细.器件号 = 器件.器件号
WHERE 订购明细.订购单号 = 'DG0001'
''')
for row in cursor.fetchall():
    print(row)


# 4. 查询“工资≥8000”的职工及其所属仓库
print("\n=== 工资≥8000的职工 ===")
cursor.execute('''
SELECT 职工.姓名, 职工.工资, 仓库.仓库号, 仓库.城市
FROM 职工 JOIN 仓库 ON 职工.仓库号 = 仓库.仓库号
WHERE 职工.工资 >= 8000
''')
for row in cursor.fetchall():
    print(row)


# 5. 查询“供应商华信电子（GY0001）”供应的所有器件
print("\n=== 华信电子供应的器件 ===")
cursor.execute('''
SELECT 器件.器件号, 器件.器件名称, 器件.规格
FROM 供应 JOIN 器件 ON 供应.器件号 = 器件.器件号
WHERE 供应.供应商号 = 'GY0001'
''')
for row in cursor.fetchall():
    print(row)


# 6. 统计每个城市的仓库总面积
print("\n=== 各城市仓库总面积 ===")
cursor.execute('''
SELECT 城市, SUM(面积) AS 总面积
FROM 仓库
GROUP BY 城市
''')
for row in cursor.fetchall():
    print(row)


conn.close()

五、总结

用Python+SQLite搭小型数据库真的很方便：不用装服务，代码写完直接跑，适合做Demo或小型项目。如果是生产环境，可能需要换成MySQL/PostgreSQL，但核心逻辑是一样的~

要不要我帮你补充这篇博客的Markdown格式优化版（比如加代码高亮、配图说明）？

Donate
  • Copyright： Copyright is owned by the author. For commercial reprints, please contact the author for authorization. For non-commercial reprints, please indicate the source.
  • Copyrights © 2023-2025 John Doe
  • Visitors: | Views:

请我喝杯茶吧~

支付宝
微信