资讯详情

Flask药物管理系统:数据库建模与库存预警实战拆解

📅 2026/9/17 16:55:33 | 华诺云谱 👁 阅读
Flask药物管理系统:数据库建模与库存预警实战拆解
简介基于Python的药物管理系统面向药剂科人员、药店管理者及Python学习者用于药品信息录入、查询、更新与删除通过图形界面和数据库交互提升库存管理效率。资源包共51个文件约23.73MB包含5个Python源码、3个SQL数据库脚本、24个HTML页面模板、7个XML配置以及说明文档与测试文件涵盖Tkinter/PyQt界面、SQLite/MySQL数据操作、ORM映射和单元测试等典型技术栈。目前已有108人学习下载适合正在练习桌面应用开发或课程设计的开发者。通过源码可直观理解模块化分层、错误处理与配置文件管理SQL脚本便于快速还原数据库结构HTML模板展示前端交互方式是一份从数据库设计到界面实现较完整的实训参考。1. 药物管理系统从解压 ZIP 到初始化数据库的第一轮拆解拿到一个Medicine-System-main压缩包第一反应是别急着跑python app.py。这个项目结构里同时出现了app.py、route.py、models.py、config.py、templates和static说明它不是单文件教学脚本而是一个按 Flask 分层组织的小型 Web 应用目录里还有poetry.lock依赖管理走的是 Poetry 而非裸requirements.txt。仓库里带着sql/目录意味着数据库不是启动时用 ORM 自动建的而是先提供 SQL 脚本再让应用层连接使用——这种设计在需要预置基础数据或由 DBA 控制表结构的场景里很常见。很多药房、诊所的库存不准根源往往不是录入错误而是把批号、有效期当成普通字符串存着查询时全靠LIKE碰运气。这个项目既然单独建了models.py放 ORM 模型再去读核心业务代码就能学到一条完整链路SQL 初始化、模型映射、路由层参数解析、库存预警、批量数据交换。下面按这条链路把项目从解压到能跑通的关键点全部过一遍。2. 数据库初始化与 SQLAlchemy 模型设计批次与库存怎么建模2.1 解压后的目录结构与入口定位解压 ZIP 后实际参与运行的目录是medicine_system它下面才是核心代码根目录的.idea、.iml是 PyCharm 工程文件可以直接无视。sql/里是建表与预置数据的 SQL 脚本config.py负责配置数据库连接和 Flask 运行参数route.py承载视图函数models.py定义 ORM 模型app.py是应用工厂和启动入口。templates/是 Jinja2 模板static/放 CSS、JS 等静态资源。这种结构下业务代码和数据定义是分离的。我一般拿到手第一步是看config.py里SQLALCHEMY_DATABASE_URI指向哪里它决定你要往哪个数据库里灌数据# 查看默认配置确认是 SQLite 还是 MySQL grep -n SQLALCHEMY_DATABASE_URI config.py # 如果目录里有 .db 或者 .sqlite 文件先确认它是否是最新的库 ls -lh *.db *.sqlite 2/dev/null如果配置指向sqlite:///medicine.db数据库文件会在应用启动时自动创建在项目根目录如果指向 MySQL 或 PostgreSQL需要先手动建库再把sql/里的脚本导入进去。SQLite 管自己MySQL 管连接串这是两套完全不同的初始化路径先分清楚再动手。2.2 从 SQL 脚本导入初始数据两种可复现的方式项目带了 SQL 文件最常见的应用方式是把它交给sqlite3命令行工具一次性执行。以 SQLite 为例进入sql/目录后执行sqlite3 ../medicine.db init.sql这条命令将init.sql里的全部CREATE TABLE、INSERT语句按顺序执行到medicine.db。如果 SQL 脚本里已经带了DROP TABLE IF EXISTS重复执行会安全地重建表如果没有幂等保护重复执行会报 table already exists这时候要么删库重来要么只执行增量部分。不想装 SQLite 命令行的话用 Python 内置的sqlite3模块也能完成同样的导入适合在 Windows 上懒得配 PATH 的场景import sqlite3 import sys db_path medicine.db sql_path sql/init.sql conn sqlite3.connect(db_path) with open(sql_path, r, encodingutf-8) as f: sql_text f.read() # 注意 executescript 会自动提交事务不要把多语句脚本传给 execute() conn.executescript(sql_text) conn.commit() conn.close() print(f已从 {sql_path} 导入数据到 {db_path})executescript与execute的区别在于executescript会先隐式提交当前事务再逐条执行脚本中的 SQL 语句因此适合导入整个文件而execute一次只允许一条语句直接传入带多个分号的脚本会报语法错误。实际做数据迁移时我更倾向于把executescript封装成一个init_db()函数放在app.py被if __name__ __main__调用之前按需触发。2.3 models.py 中的表结构设计与建模理由models.py定义了核心实体。一个药物管理系统里药品信息、入库记录、出库记录是三个最基本的表。药品表负责描述「现在有什么药」入库表记录「什么时候进了什么批次」出库表记录「消耗到了哪里」。三者之间用外键关联而不是把库存数量直接冗余在药品表里——冗余看似简单但每次出入库都要改同一行并发高了容易丢更新。以下是按常见业务补全的模型骨架字段命名与项目风格保持一致from datetime import datetime from sqlalchemy import (Column, Integer, String, Date, DateTime, ForeignKey, Numeric, Text) from sqlalchemy.orm import relationship from flask_sqlalchemy import SQLAlchemy db SQLAlchemy() class Drug(db.Model): 药品主表 __tablename__ drugs id Column(Integer, primary_keyTrue) drug_name Column(String(128), nullableFalse, indexTrue) # 通用名 specification Column(String(64)) # 规格如 0.25g*24片 dosage_form Column(String(32)) # 剂型片剂/胶囊/注射液 manufacturer Column(String(128)) # 生产厂家 batch_number Column(String(64), uniqueTrue, indexTrue) # 批号全局唯一 expiry_date Column(Date, nullableFalse) # 有效期至 quantity Column(Integer, default0) # 当前库存 warning_threshold Column(Integer, default20) # 低库存预警线 created_at Column(DateTime, defaultdatetime.now) updated_at Column(DateTime, defaultdatetime.now, onupdatedatetime.now) stock_records relationship(StockRecord, backrefdrug) class StockRecord(db.Model): 出入库流水 __tablename__ stock_records id Column(Integer, primary_keyTrue) drug_id Column(Integer, ForeignKey(drugs.id), nullableFalse) change_type Column(String(8), nullableFalse) # in / out / adjust change_qty Column(Integer, nullableFalse) operator Column(String(64)) remark Column(Text) created_at Column(DateTime, defaultdatetime.now)这里有三个设计要点值得留意。第一batch_number加了唯一约束同一批药只占一行库存出入库都在这一行上累加或扣减第二expiry_date用Date类型而不是DateTime有效期精确到天就够了带时分秒只会让比较逻辑更啰嗦第三quantity是冗余字段它等于所有入库流水减去出库流水的累计值。理论上可以实时聚合算出库存但查询频繁时冗余字段换性能是值得的。维护一致性靠事务每次写入流水的同时更新drugs.quantity两个操作包在同一个事务里。数据库字段的含义和约束表格如下字段类型约束作用drug_nameString(128)索引药品名称模糊搜索的主要对象batch_numberString(64)唯一 索引批号唯一防止同一批药重复建档expiry_dateDate非空有效期临期预警依赖它quantityInteger默认 0当前库存冗余值warning_thresholdInteger默认 20低于该值视为缺货change_typeString(8)非空流水类型控制库存加减方向实际项目中这个表还会扩展unit最小包装单位和storage_condition阴凉/冷藏/常温两个字段前者影响出库计量后者决定存放位置。但核心的库存事务逻辑不变任何对quantity的修改都必须产生一条stock_records流水否则追溯时就是一笔糊涂账。3. 路由层查询与库存预警从参数解析到聚合过滤3.1 Flask 路由如何承接查询参数route.py里的视图函数是系统对外的门面。一个合格的列表接口至少要支持按名称搜索、按库存状态过滤、分页三个能力。用 Flask 的request.args接收查询字符串配合 SQLAlchemy 的查询对象可以组装出灵活且安全的动态查询避免手工拼接 SQL 字符串from flask import Blueprint, request, jsonify from sqlalchemy import or_ from .models import Drug, db drug_bp Blueprint(drug, __name__) drug_bp.route(/api/drugs, methods[GET]) def list_drugs(): # 1. 读取查询参数并做类型转换 page request.args.get(page, 1, typeint) per_page request.args.get(per_page, 20, typeint) name request.args.get(name, ).strip() low_stock_only request.args.get(low_stock, 0) 1 # 2. 动态组装过滤条件 filters [] if name: # 名称或厂家模糊匹配 filters.append(or_( Drug.drug_name.like(f%{name}%), Drug.manufacturer.like(f%{name}%) )) if low_stock_only: filters.append(Drug.quantity Drug.warning_threshold) # 3. 执行查询并分页 query Drug.query.filter(*filters).order_by(Drug.updated_at.desc()) pagination query.paginate(pagepage, per_pageper_page, error_outFalse) return jsonify({ total: pagination.total, items: [{ id: d.id, drug_name: d.drug_name, batch_number: d.batch_number, quantity: d.quantity, expiry_date: d.expiry_date.isoformat() } for d in pagination.items] })这段代码的关键在request.args.get带typeint的写法。它让 Flask 把查询字符串里的page和per_page自动转成整型用户传了pageabc时不会抛 500 错误而是回退到默认值 1。注意error_outFalse保证了页码超出范围时返回空列表而不是 404前端拿到total自己计算总页数这是列表接口最常用的约定。3.2 低库存与临期预警的动态过滤预警是药物管理系统的刚需。低库存要看quantity是否低于阈值临期要看expiry_date距离今天是否在某个天数以内。用 SQLAlchemy 写日期比较时最干净的方式是利用datetime计算目标日期再与字段做比较避免在 SQL 层写时区敏感的NOW()函数from datetime import datetime, timedelta drug_bp.route(/api/drugs/warnings, methods[GET]) def warning_drugs(): # 预警条件低库存 或 90 天内过期 expiring_soon datetime.now().date() timedelta(days90) warnings Drug.query.filter( db.or_( Drug.quantity Drug.warning_threshold, Drug.expiry_date expiring_soon ) ).all() result [] for d in warnings: if d.quantity d.warning_threshold: result.append({drug: d.drug_name, type: low_stock, level: danger if d.quantity 0 else warning}) elif d.expiry_date expiring_soon: result.append({drug: d.drug_name, type: expiring, days_left: (d.expiry_date - datetime.now().date()).days}) return jsonify(result)timedelta(days90)这个 90 天阈值是业务参数实际药房会拆成两个档180 天以上为正常30 到 90 天要重点提醒30 天以内直接下架。这个逻辑放在 Python 层还是 SQL 层都行但放在models.py里做成property会更利于复用from sqlalchemy.ext.hybrid import hybrid_property class Drug(db.Model): # ... 省略已有字段 ... hybrid_property def is_low_stock(self): return self.quantity self.warning_threshold hybrid_property def is_expiring_soon(self, days90): return self.expiry_date datetime.now().date() timedelta(daysdays)用 hybrid property 的好处是在 Python 侧访问drug.is_low_stock时走 Python 比较在查询里用Drug.is_low_stock True时 SQLAlchemy 会把它翻译成对应的 SQL 条件。同一个业务规则两处使用一处定义不会出现「列表页和详情页的预警逻辑不一致」这种低级但常见的 bug。3.3 库存流水聚合与药品列表联查如果项目需要展示每种药的入库总量、出库总量、净库存直接用group_by聚合是最稳妥的。StockRecord表里的change_type区分in和out聚合时用CASE WHEN分别求和from sqlalchemy import func, case drug_bp.route(/api/drugs/stats, methods[GET]) def drug_stats(): stats db.session.query( Drug.id, Drug.drug_name, func.sum(case((StockRecord.change_type in, StockRecord.change_qty), else_0)).label(total_in), func.sum(case((StockRecord.change_type out, StockRecord.change_qty), else_0)).label(total_out) ).join(StockRecord, StockRecord.drug_id Drug.id) \ .group_by(Drug.id) \ .all() return jsonify([{ drug_name: s.drug_name, total_in: int(s.total_in or 0), total_out: int(s.total_out or 0), current_stock: int(s.total_in or 0) - int(s.total_out or 0) } for s in stats])case 表达式在这里相当于 SQL 里的SUM(CASE WHEN ... THEN ... ELSE 0 END)是聚合统计的标准写法。这里要特别注意的是total_in or 0当某条流水的change_qty为空或聚合结果为空时数据库返回NULLPython 里的sum拿到None后做减法会直接报TypeError所以or 0是必要的防护。实际调接口时如果发现 JSON 里出现了null而不是数字十有八九是这个位置没做空值兜底。4. 批量导入导出与数据校验Excel/CSV 场景的编码与格式坑4.1 CSV 导入最常见的编码灾难ERP 导出的药品目录通常是 CSV而中国药品名录里的厂家名称基本都包含中文。CSV 文件最常见的编码是GBKWindows Excel 默认和UTF-8Python 的open()默认按系统语言解码Windows 下大概率用gbk去解 UTF-8 文件直接抛UnicodeDecodeError。这个错不是数据坏了是解码方式选错了。我一般用chardet先探测编码再用探测结果打开文件import chardet import pandas as pd def smart_read_csv(path): 自动探测编码并读取 CSV避免手工指定编码的麻烦 with open(path, rb) as f: raw f.read(20000) # 读前 20KB 足够判断编码了 detected chardet.detect(raw) encoding detected.get(encoding, utf-8) try: df pd.read_csv(path, encodingencoding) except UnicodeDecodeError: # 探测失败时退回 gbk再不行就报错让用户明确指定 df pd.read_csv(path, encodinggbk) return df, encoding参数说明f.read(20000)只读取文件前 20KB 用来做编码探测而不是读整个文件因为编码判断只需要样本就足够chardet.detect返回一个包含encoding和confidence的字典confidence越接近 1 越可信。读完整文件后再用pd.read_csv时仍可能遇到极少数文件在文件中部切换编码的情况这时候可以换errorsignore先看一眼数据量是否对得上再决定是丢弃坏行还是人工修正数据源。4.2 批号格式校验与字段清洗导入药品数据最容易出问题的字段是批号和有效期。批号在不同药厂格式差异很大但至少应该做到「不为空、无首尾空格、长度合理」。用正则做基础约束是性价比最高的方式import re from datetime import datetime BATCH_PATTERN re.compile(r^[A-Za-z0-9\-]{6,32}$) def validate_and_normalize(row): 对单行导入数据进行清洗与格式校验 name str(row.get(药品名称, )).strip() batch str(row.get(批号, )).strip() expiry str(row.get(有效期至, )).strip() if not name: raise ValueError(药品名称不能为空) if not BATCH_PATTERN.match(batch): raise ValueError(f批号 {batch} 不符合格式要求) # 兼容 2024-12-31 与 2024/12/31 两种常见写法 for fmt in (%Y-%m-%d, %Y/%m/%d): try: expiry_date datetime.strptime(expiry, fmt).date() break except ValueError: continue else: raise ValueError(f有效期 {expiry} 无法解析) return { drug_name: name, batch_number: batch, expiry_date: expiry_date, }这段代码里for...else的语义值得再说一遍如果for循环里的break成功执行说明日期解析成功else块不会运行如果循环结束都没能解析出合法日期则执行else块抛出异常。这是 Python 里处理「多个格式逐个尝试全部失败才报错」的标准模式比布尔标志位要简洁得多。4.3 批量导入的失败回滚策略一次性导入几千条数据时逐条校验、逐条插入的坏处是效率低而且中途失败会导致「前一半进了库后一半卡了壳」。正确的做法是先把所有数据校验进内存全部通过后再在一个事务里批量提交from sqlalchemy.exc import IntegrityError from .models import Drug, db def batch_import(path): df, enc smart_read_csv(path) drug_list [] errors [] for idx, row in df.iterrows(): try: item validate_and_normalize(row) drug_list.append(Drug(**item)) except ValueError as e: errors.append(f第 {idx 2} 行: {e}) if errors: return False, errors try: db.session.add_all(drug_list) db.session.commit() return True, len(drug_list) except IntegrityError: db.session.rollback() return False, [批号与库内已有数据冲突整批未导入]把校验和提交拆开的意义是如果第 100 行的批号与库里已有重复IntegrityError会抛出但此时db.session里的状态是坏的必须先rollback()再继续。让整批数据要么全进、要么全不进比逐条插入然后补做「哪些成功了哪些没成功」的排查更让人省心。真实库存系统中批号重复往往是导入模板里混入了两次同一批次的数据提前用set做一次行内去重可以把这种错误挡在校验之前seen_batches set() if batch in seen_batches: errors.append(f第 {idx 2} 行: 批号 {batch} 在本次导入中重复) seen_batches.add(batch)4.3.1 导出模板与示列数据下发的处理批量导入配套的导出功能通常用来生成一张符合模板要求的空表。这里有一个在真实项目中反复出现的坑pandas导出 Excel 依赖openpyxl而openpyxl写入时默认把空单元格置为None如果不去处理导出的模板在用户填完数据再导回时会把整列为空的单元格读成NaN而NaN传给字符串字段时会变成字符串nan污染数据。解决方式是在导出时给模板把必填字段做成下拉选项同时在导入时对NaN做显式处理# 导出模板时用 openpyxl 直接写下拉框 from openpyxl import Workbook from openpyxl.worksheet.datavalidation import DataValidation wb Workbook() ws wb.active ws.append([药品名称, 批号, 剂型, 库存数量, 有效期至]) dv DataValidation( typelist, formula1片剂,胶囊,注射液,颗粒,软膏, allow_blankTrue ) ws.add_data_validation(dv) dv.add(C2:C1000) wb.save(drug_import_template.xlsx)导入侧则对每个字段先判断是否为pandas.isna()是则默认空字符串或None而不是直接str()把NaN转成字符串nan。这个细节在数据量小的时候看不出问题等库存数量从文本变成数字、字符串前后多出空格时就会引发类型不匹配和查询失配。5. 部署验证与冒烟脚本用临时数据库验证整套链路5.1 依赖安装与 Python 环境选择项目依赖由poetry.lock锁定最稳妥的安装方式是用 Poetry 创建独立环境并把依赖装进隔离目录避免污染系统 Python。若本机没有 Poetry则先安装它再执行安装# 安装 PoetrymacOS/Linux 下的常见做法 curl -sSL https://install.python-poetry.org | python3 - # 在项目根目录创建虚拟环境并安装依赖 poetry install # 激活虚拟环境后启动服务 poetry shell python app.py如果所在环境没有 Poetry也可以直接读取pyproject.toml里的依赖列表用 pip 手动安装核心依赖。注意 Python 版本如果低于 3.8flask_sqlalchemy的某些新版本 API 可能不兼容建议先python --version确认版本再决定是否降级依赖版本。系统里同时存在多个 Python 版本时务必用python3 -m venv .venv创建虚拟环境避免把依赖装进全局环境导致版本冲突。5.2 冒烟测试脚本用内存数据库验证查询逻辑我每次拿到新的 Flask 项目习惯先把应用指向一个临时 SQLite 数据库跑一遍冒烟测试确认路由、模型、查询之间的协作没有断裂。这样做的好处是坏了立刻知道不用在浏览器里反复刷新页面猜问题。# 用环境变量覆盖数据库配置 export DATABASE_URLsqlite:////tmp/medicine_test.db # 启动 Flask 开发服务器 flask --app app.py run --port 5000 # 另开终端验证主页和接口返回 curl http://127.0.0.1:5000/api/drugs如果不想手动起服务也可以在 Python 里直接构造测试请求让 Flask 的 test client 代劳import tempfile import os def test_health(): 最小冒烟测试录入一条数据再按名称搜出来 with tempfile.TemporaryDirectory() as tmpdir: test_db os.path.join(tmpdir, test.db) os.environ[DATABASE_URL] fsqlite:///{test_db} from app import create_app from medicine_system.models import db, Drug app create_app() with app.app_context(): db.create_all() d Drug(drug_name阿莫西林胶囊, batch_numberAMX20250101, expiry_date2027-01-01, quantity300) db.session.add(d) db.session.commit() result Drug.query.filter(Drug.drug_name.contains(阿莫西林)).first() assert result.batch_number AMX20250101 assert result.quantity 300 print(冒烟测试通过)这段脚本覆盖了「建表 → 插入 → 查询」的最小闭环。测试完成后临时目录被TemporaryDirectory自动清理不会在项目里留下多余的文件。把这段脚本保存为smoke_test.py每次调整models.py或route.py后跑一遍能过滤掉绝大部分低级失误比如字段名拼错、表没建对、外键关联缺失这类问题。本文还有配套的精品资源点击获取
📝

华诺云谱内容团队

资深建站顾问 · 行业研究员

10年+企业数字化服务经验,专注智能建站、SEO优化与品牌营销,持续输出建站技巧、行业洞察与营销干货,已帮助5000+企业实现数字化增长。

你可能需要的服务

订阅华诺云谱资讯周报

每周一封,精选建站技巧、SEO与营销干货,直达邮箱。已有 8,000+ 企业主订阅,助你少走弯路。