ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python药物管理系统数据库设计:SQLite建表、事务与并发控制实战

Python药物管理系统数据库设计:SQLite建表、事务与并发控制实战 简介一套基于Python的药物管理系统完整项目包面向希望在真实业务场景中掌握Python、数据库与GUI开发的初中级学习者以及需要快速搭建药店或医院药品库存管理原型的技术人员。项目将药品录入、查询、更新、删除等核心流程整合在界面化应用中并提供配套数据库文件可直接运行与二次扩展。压缩包共51个文件约23.73MB涵盖5个py源码、3个sql数据库脚本、24个html页面模板、7个xml配置或元数据文件以及README、poetry配置、license等辅助文档目录按源码、测试、模板和配置分层便于按模块理解。当前已有108人浏览学习。除基础增删改查外项目还涉及SQLite/MySQL等数据库交互、Tkinter或PyQt式GUI布局、ORM映射、正则校验、异常处理、模块化工程组织与单元测试等知识点。通过阅读py源码、sql脚本与html模板的联动关系可以掌握从数据建模到界面渲染的完整设计思路是巩固Python综合开发能力的实用素材。1. 当 python 遇到药物管理系统数据库文件才是骨架拿到基于python的药物管理系统含数据库文件.zip这个压缩包先别急着翻main.py第一件事应该是打开里面的数据库文件看看表结构和数据量。这类系统的业务逻辑并不复杂真正的难度在于把药品、库存流水、供应商、处方和操作员之间的关系理顺并保证出库入库不丢数据。常见的实现是 Python 连接一个 SQLite 数据库文件界面只承担录入和展示。要注意的是压缩包里的数据库文件通常不是 动态生成的而是建库脚本跑过一次后留下的结果。无论你的目标是把这个项目跑起来还是想把它改造成可交付的内部系统都需要从数据库这一层入手。本文不打算复述界面操作而是沿着python 药物管理系统 数据库文件这条线把建表、增删改查、查询预警和并发控制讲清楚。适合刚开始做信息管理系统、或者接手类似课设代码想优化的人阅读——只要本机有 Python 环境就能动手因为sqlite3是标准库不需要额外安装。2. 药物管理系统的数据库文件设计建表、关系与初始化脚本2.1 先拆核心表药品信息与库存流水为什么要分开很多药物管理系统的早期版本会把库存数量直接写死在药品表里然后通过一次UPDATE drugs SET quantity quantity - 1完成出库。这种做法在演示时能跑通但一旦需要追溯「这批药是什么时候进的、谁操作的」就没有任何记录可查。所以第一刀必须把「当前库存」和「库存变动记录」分成两张表。drugs表只保存当前状态比如编号、名称、规格、库存数量、进价、售价、有效期和供应商。stock_log表保存每一次变动入库记正数出库记负数。查询当前库存可以直接读drugs.quantity做统计时再汇总stock_log这样两侧各有分工。除了这两张表一般还需要suppliers供应商表以及users操作员表否则出库流水没办法关联到具体责任人权限控制也无从谈起。下面是这套系统里最常见的基础表设计。字段不追求多但关系必须明确。表名作用与药品表的关系users操作员账号与角色stock_log.operator_id - users.user_idsuppliers供应商联系方式drugs.supplier_id - suppliers.supplier_iddrugs药品当前库存与价格被 stock_log 引用stock_log每次入库/出库/盘点流水drug_id - drugs.drug_id2.2 建库脚本用 SQL 快速生成数据库文件SQLite 的数据库就是一个.db文件所以初始化脚本就是一段可重复执行的 SQL。注意AUTOINCREMENT不是必需的主键加INTEGER PRIMARY KEY本身就能自增。只在需要严格保证 id 不回退时才加AUTOINCREMENT代价是每次插入多一次系统表查询。PRAGMA foreign_keys ON; CREATE TABLE IF NOT EXISTS users ( user_id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password TEXT NOT NULL, role TEXT NOT NULL DEFAULT clerk CHECK (role IN (admin, clerk)), created_at TEXT NOT NULL DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS suppliers ( supplier_id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, phone TEXT, contact TEXT ); CREATE TABLE IF NOT EXISTS drugs ( drug_id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, spec TEXT, unit TEXT NOT NULL DEFAULT 盒, quantity INTEGER NOT NULL DEFAULT 0 CHECK (quantity 0), purchase_price REAL NOT NULL DEFAULT 0 CHECK (purchase_price 0), sale_price REAL NOT NULL DEFAULT 0 CHECK (sale_price 0), expire_date TEXT NOT NULL CHECK (expire_date 2000-01-01), supplier_id INTEGER, created_at TEXT NOT NULL DEFAULT (datetime(now, localtime)), FOREIGN KEY (supplier_id) REFERENCES suppliers(supplier_id) ); CREATE TABLE IF NOT EXISTS stock_log ( log_id INTEGER PRIMARY KEY AUTOINCREMENT, drug_id INTEGER NOT NULL, change_type TEXT NOT NULL CHECK (change_type IN (in, out, adjust)), quantity INTEGER NOT NULL, remark TEXT, operator_id INTEGER, created_at TEXT NOT NULL DEFAULT (datetime(now, localtime)), FOREIGN KEY (drug_id) REFERENCES drugs(drug_id), FOREIGN KEY (operator_id) REFERENCES users(user_id) ); CREATE INDEX IF NOT EXISTS idx_drugs_name ON drugs(name); CREATE INDEX IF NOT EXISTS idx_drugs_expire ON drugs(expire_date); CREATE INDEX IF NOT EXISTS idx_stock_log_drug_time ON stock_log(drug_id, created_at);选择TEXT保存日期而不是时间戳是因为 SQLite 的日期函数能把YYYY-MM-DD HH:MM:SS字符串直接比较和格式化在查询和打印时都不需要额外转换。CHECK (quantity 0)是数据库层防止库存变负数的兜底应用层校验只是一个位置DBA 不能只信接口。价格用REAL在演示系统够用涉及金额精确计算时应该换成整数分但那样会加大前后端转换成本。这段 SQL 值得注意的地方是expire_date 2000-01-01它排除了明显的脏数据但真正的效期校验还应该在应用层做因为药品有效期是业务规则数据库约束管不了「采购时填入已经过期的日期」这种场景。2.3 用 PRAGMA 控制外键生效和数据库结构版本SQLite 默认不开启外键约束连接后需要执行一次PRAGMA foreign_keys ON否则上面的FOREIGN KEY定义全都不会被执行。这个问题在 Python 里尤其隐蔽因为sqlite3模块不会主动打开它。初始化时还需要维护一个结构版本号方便后续加字段或改索引。import sqlite3 from pathlib import Path DB_PATH Path(pharmacy.db) SCHEMA_PATH Path(schema.sql) def init_db(db_path: Path DB_PATH) - None: db_path.parent.mkdir(parentsTrue, exist_okTrue) conn sqlite3.connect(db_path) try: conn.executescript(SCHEMA_PATH.read_text(encodingutf-8)) conn.execute(PRAGMA user_version 1) conn.commit() print(database initialized, version1) finally: conn.close() if __name__ __main__: init_db()executescript会先提交当前事务再逐条执行脚本里的 SQL如果某条语句报错前面的建表结果会保留所以重复执行这个脚本是安全的因为所有语句都加了IF NOT EXISTS。PRAGMA user_version是一个只占一个整数的元数据不参与业务查询但升级表结构时可以用来判断要不要执行ALTER TABLE。相比新建一张schema_version表user_version更轻量缺点是只能存一个整数复杂迁移需要靠代码里划分版本号区间来处理。3. python 连接数据库文件增删改查的落地写法与事务控制3.1 连接参数row_factory、foreign_keys 与 check_same_thread写数据库访问层之前先把连接函数固定住避免每个业务方法都连接一次、忘记设置关键参数。最核心的三个参数是timeout、row_factory和foreign_keys。import sqlite3 def get_connection(db_path: str pharmacy.db) - sqlite3.Connection: conn sqlite3.connect(db_path, timeout5.0) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) return conntimeout5.0表示当数据库文件被其他连接锁住时等待 5 秒再抛OperationalError: database is locked。row_factory sqlite3.Row让查询结果可以按列名访问row[name]比row[1]可读性高得多而且不会因为SELECT *改变字段顺序而出错。PRAGMA foreign_keys ON上面已经提过必须放在连接对象上执行不能写进.db文件它是会话级的设置。参数取值作用timeout5.0等待锁的最长时间单位秒row_factorysqlite3.Row让结果集支持按列名取值check_same_threadFalse多线程共享连接时开启一般不建议滥用check_same_threadFalse常用在多线程 GUI 或 Web 服务里。注意它只是解除 Python 层面的线程检查SQLite 本身仍不允许同一连接跨线程安全并发真正可靠的做法是每个线程单独建立连接或者用连接池。3.2 入库与出库的完整事务一次 UPDATE 和一次 INSERT 必须同时成功入库不只是给drugs表加数量还要在stock_log里留下变动记录。如果两步之间程序崩溃会出现库存变了但流水没记录或者流水有记录而库存没变。解决办法就是用事务把两条 SQL 包在一起。def add_stock(drug_id: int, quantity: int, operator_id: int, remark: str ) - None: if quantity 0: raise ValueError(入库数量必须大于 0) with get_connection() as conn: cur conn.execute( UPDATE drugs SET quantity quantity ? WHERE drug_id ?, (quantity, drug_id), ) if cur.rowcount ! 1: raise KeyError(f药品 {drug_id} 不存在) conn.execute( INSERT INTO stock_log(drug_id, change_type, quantity, operator_id, remark) VALUES (?, in, ?, ?, ?), (drug_id, quantity, operator_id, remark), )Python 的sqlite3模块在进入with conn:块时不会自动开事务但一旦执行第一条 DML 语句就会隐式开启块结束后如果没抛异常就commit如果函数内任何位置抛异常就rollback。所以上面代码里rowcount ! 1触发KeyError后UPDATE和后续的INSERT都会被回滚不会出现只加库存不写流水的情况。出库逻辑比入库多两步先查当前库存够不够再决定是否减库存。这里要小心「先查后写」的并发窗口。在 SQLite 单写者的模型下两个连接同时出库时后者会阻塞等待如果库存只剩 5 件两个请求同时读到 5各自扣 3按上面的普通写法可能出现库存变成 2 和 2 的叠加实际库存变成 -1。稳妥的做法是在UPDATE时加条件判断。def sell_drug(drug_id: int, quantity: int, operator_id: int) - int: if quantity 0: raise ValueError(销售数量必须大于 0) with get_connection() as conn: row conn.execute( SELECT quantity, sale_price FROM drugs WHERE drug_id ?, (drug_id,), ).fetchone() if row is None: raise KeyError(药品不存在) if row[quantity] quantity: raise RuntimeError(库存不足) cur conn.execute( UPDATE drugs SET quantity quantity - ? WHERE drug_id ? AND quantity ?, (quantity, drug_id, quantity), ) if cur.rowcount ! 1: raise RuntimeError(库存不足并发下扣减失败) conn.execute( INSERT INTO stock_log(drug_id, change_type, quantity, operator_id) VALUES (?, out, -?, ?), (drug_id, quantity, operator_id), ) return int(row[sale_price] * quantity)WHERE quantity ?是乐观锁的数据库实现方式即使第一次 SELECT 后库存被其他线程扣减UPDATE的影响行数也是 0这时回滚整个事务比用SELECT FOR UPDATE简单且数据库无关。库存流水里的quantity我习惯存负数这样统计销售时可以直接SUM(-quantity)无需额外区分方向。3.3 参数化查询与业务校验拦截重复编码和非法操作拼接 SQL 字符串是这类管理系统最常见的隐患。药品名称和备注是自由文本一旦有人输入 OR 11拼接出的 SQL 就会改变语义。标准做法是统一使用?占位符让sqlite3模块处理转义。插入新药品时数据库层的唯一约束code TEXT NOT NULL UNIQUE已经能挡住重复编码但应用层可以先查一次给出更友好的提示。def create_drug( code: str, name: str, expire_date: str, purchase_price: float, sale_price: float, supplier_id: int | None None, ) - int: if code.strip() or name.strip() : raise ValueError(药品编码和名称不能为空) if sale_price 0 or purchase_price 0: raise ValueError(价格不能为负数) try: with get_connection() as conn: cur conn.execute( INSERT INTO drugs(code, name, expire_date, purchase_price, sale_price, supplier_id) VALUES (?, ?, ?, ?, ?, ?), (code, name, expire_date, purchase_price, sale_price, supplier_id), ) return int(cur.lastrowid) except sqlite3.IntegrityError as exc: raise ValueError(药品编码已存在或供应商不存在) from excSQLite 的IntegrityError既包含主键冲突也包含外键冲突和CHECK约束失败。把所有情况统一报成「编码或供应商有问题」对界面调用方来说可以接受但日志里要保留原始异常exc否则排错时很难定位具体是哪一种约束被触发。分页列表查询同样用参数占位注意LIKE的百分号要拼进参数里而不是拼在 SQL 中def search_drugs(keyword: str, offset: int 0, limit: int 20) - list[sqlite3.Row]: pattern f%{keyword}% with get_connection() as conn: return conn.execute( SELECT d.*, s.name AS supplier_name FROM drugs d LEFT JOIN suppliers s ON s.supplier_id d.supplier_id WHERE d.name LIKE ? OR d.code LIKE ? ORDER BY d.drug_id DESC LIMIT ? OFFSET ?, (pattern, pattern, limit, offset), ).fetchall()这里的LIMIT和OFFSET也可以作为参数传入SQLite 支持这种写法。分页接口最容易被忽视的是排序稳定性如果表里存在大量created_at相同的数据只按时间排序会导致翻页时出现重复或遗漏所以ORDER BY d.drug_id DESC才是唯一稳定条件。4. 让数据库文件回答业务问题过期预警、统计与索引提速4.1 过期预警用日期函数直接筛出 90 天内到期药品药物管理系统的一个核心价值不是录药品而是提前发现将要过期的库存。SQLite 的date(now, localtime)取当天日期配合N days修饰符就能算出预警窗口。下面这段 SQL 直接查 90 天内到期且还有库存的药品SELECT drug_id, code, name, expire_date, quantity FROM drugs WHERE expire_date BETWEEN date(now, localtime) AND date(now, localtime, 90 days) AND quantity 0 ORDER BY expire_date;对应的 Python 调用需要小心90 days这个修饰符它不能直接作为绑定参数放在date()函数里因为 SQLite 会把参数当字符串值而不是修饰符。正确做法是把整个修饰符字符串作为参数传入def get_expiring(days: int 90) - list[sqlite3.Row]: modifier f{days} days with get_connection() as conn: return conn.execute( SELECT drug_id, code, name, expire_date, quantity FROM drugs WHERE expire_date BETWEEN date(now, localtime) AND date(now, localtime, ?) AND quantity 0 ORDER BY expire_date, (modifier,), ).fetchall()expire_date列存的是YYYY-MM-DD文本字符串比较就是日期比较所以BETWEEN能直接工作。如果你之前用时间戳存这个字段这里会比较麻烦需要先datetime(expire_date, unixepoch, localtime)转换索引也失效。这解释了为什么建表时选 TEXT 而不是整数。4.2 销售统计与库存排行GROUP BY 与条件聚合组合查询统计「最近 30 天哪些药卖得多」要从stock_log里取change_type out的记录再按drug_id分组。需要注意LEFT JOIN和WHERE顺序的问题如果时间过滤写在WHERE没有流水的药品会被过滤掉LEFT JOIN等于失效。正确写法是把时间条件放进ON子句SELECT d.drug_id, d.name, COALESCE(SUM(CASE WHEN sl.change_type out THEN -sl.quantity ELSE 0 END), 0) AS sold_qty FROM drugs d LEFT JOIN stock_log sl ON sl.drug_id d.drug_id AND sl.created_at datetime(now, localtime, -30 days) GROUP BY d.drug_id, d.name ORDER BY sold_qty DESC LIMIT 10;CASE WHEN change_type out THEN -sl.quantity是因为stock_log.quantity在出库记录中是负数-sl.quantity转成正数后求和得到销售数量。COALESCE包一层是为了让没有流水的药品显示 0 而不是NULL。这个查询的数据量不大时性能没问题但stock_log增长到几十万行后created_at上的索引会变得关键。如果需要「每个药品最近一笔出库时间」窗口函数比GROUP BY更简洁SQLite 3.25 开始支持SELECT drug_id, name, last_out_time, rank FROM ( SELECT d.drug_id, d.name, sl.created_at AS last_out_time, RANK() OVER (PARTITION BY d.drug_id ORDER BY sl.created_at DESC) AS rank FROM stock_log sl JOIN drugs d ON d.drug_id sl.drug_id WHERE sl.change_type out ) WHERE rank 1 ORDER BY last_out_time DESC;窗口函数RANK() OVER (PARTITION BY drug_id ORDER BY created_at DESC)按药品分组、按时间倒序编号外层再取rank 1就得到每个药品最近一次出库。这种写法避免了对同一个表做两次自连接对熟悉传统GROUP BY MAX的人来说是值得换用的方案。4.3 查询慢的排查EXPLAIN QUERY PLAN 与索引调整如果感觉自己写出的查询越来越慢第一步不是加索引而是先看 SQLite 实际走了哪条路线。SQLite 提供了EXPLAIN QUERY PLAN它不需要真的执行查询只返回查询计划EXPLAIN QUERY PLAN SELECT * FROM drugs WHERE name LIKE 阿莫% AND expire_date 2026-01-01;一种可能的输出是SCAN drugs表示全表扫描。这时针对报警查询经常用到的两个字段建索引是最直接的加速方式索引名索引字段适配场景idx_drugs_namename按药品名称前缀匹配idx_drugs_expireexpire_date效期排序与范围过滤idx_stock_log_drug_timedrug_id, created_at按药品统计流水索引并非越多越好。LIKE %关键词%这类模糊查询不会走普通 B-Tree 索引SQLite 会退化成全表扫描此时建索引也没意义。索引的收益集中在等值查询和前缀匹配、以及ORDER BY排序上。stock_log表的idx_stock_log_drug_time是组合索引查询WHERE drug_id ? ORDER BY created_at DESC时可以一次命中不需要回表排序。这类系统最实用的调优顺序是先开EXPLAIN看扫描方式然后只给「查询频率高、区分度好」的列建索引最后用真实数据量测试一遍而不是把每个 column 都加上索引。5. 数据库文件被锁用 WAL、busy_timeout 和连接检查解决并发出库5.1 先复现「database is locked」的三种来源SQLite 的锁机制在默认 rollback journal 模式下写事务会独占整个数据库文件读事务和写事务互斥。一个后台线程在做月度统计时长时间占用读锁另一个线程这时出库就可能抛OperationalError: database is locked。这类问题在「启动界面 定时任务」并存的管理系统里很常见先别骂代码要分清三种来源写写冲突、读写冲突、长事务。写写冲突比较直观两个连接同时UPDATE后者默认等 5 秒超时后放弃。读写冲突更隐蔽默认模式下写事务开始时会获得 RESERVED 锁如果此时有读事务正在执行写事务要等读事务结束。长事务则往往由「业务处理中不小心执行了耗时的文件操作」引发锁一直不释放其他连接全部卡住。5.2 打开 WAL 模式并调整锁等待参数把 journal 模式切换成 WAL是解决读写阻塞的第一选择。WAL 模式下读操作不阻塞写操作写操作也不阻塞读操作只是所有写仍然串行。在连接上执行下面的 PRAGMA 即可PRAGMA journal_modeWAL; PRAGMA synchronousNORMAL; PRAGMA busy_timeout5000;journal_modeWAL会让数据库文件旁边多出-wal和-shm两个临时文件这是正常现象不是损坏备份时不能只拷pharmacy.db要同时处理 WAL 文件或者先跑一次checkpoint。synchronousNORMAL在 WAL 模式下性能与安全平衡较好应用崩溃最多丢最近一次提交但不会损坏数据库synchronousFULL更稳但写延迟更高。参数设置值效果journal_modeWAL读写并发不互相阻塞synchronousNORMAL崩溃后最多丢最后提交不会损坏文件busy_timeout5000锁等待 5 秒后再报错验证 PRAGMA 是否生效可以直接用命令行工具读回当前值sqlite3 pharmacy.db PRAGMA journal_mode; PRAGMA busy_timeout; PRAGMA synchronous;正常输出依次是wal、5000、1。如果输出是delete或0说明设置没有被正确保存——因为这些 PRAGMA 大多是连接级的必须在每一次连接建立时都执行而不是只在初始化时执行一次。5.3 连接策略上避免锁扩散即使开了 WAL多个写连接并发提交时仍可能出现SQLITE_BUSY。常见做法是在应用层收敛写连接如果是单进程多线程用一个模块级唯一连接配合threading.Lock把所有写操作串行化如果是 Flask 这类多线程 Web 服务最省事的方案是每个请求新建一个连接配合上面 5 秒的busy_timeout让写请求等待而不是立刻报错。还有一种错误做法是把连接设为全局单例且开启check_same_threadFalse后又多线程并发写这会让database is locked变成偶发随机 bug排错成本远高于单连接加锁。最后提供一个排错顺序先看是不是journal_mode回退成了delete再看锁等待时间是否被其他代码重新设置过最后用sqlite3的PRAGMA wal_checkpoint(PASSIVE)手动合并 WAL 文件确认没有后台长事务卡住 checkpoint。把这四步走完并发出库的稳定性基本就有保障了。本文还有配套的精品资源点击获取
返回列表