☰
Python门诊挂号系统实战:Flask+SQLite高并发号源管理
2026/9/28 1:30:42 网站建设 项目流程

简介:本资源是一套完整的医院门诊挂号与预约系统毕业设计级源码,面向计算机、软件工程或医疗信息化方向的本科生及课程设计学习者,旨在解决传统挂号流程低效、信息孤岛等问题,提供可落地的前后端协同开发范例。压缩包共220个文件,大小8.39MB,涵盖27个Python后端脚本(含业务逻辑与API接口)、22个Vue组件与24个TypeScript文件(构建响应式前端界面)、39个SVG矢量图与53张JPG/PNG图片(支撑可视化交互),以及SQL建表语句、Markdown文档和WOFF字体等配套资源。已有361人学习下载,资源结构清晰,含.gitignore等工程规范配置,且预览可见index.html入口、数据库说明文档(表结构.docx)及多张界面截图,便于快速理解系统架构、复现部署流程并开展二次开发。

1. 为什么一个“能跑通”的医院门诊挂号系统,比写十遍Hello World更考验Python工程能力?

你手头有一份标着“基于Python的医院门诊挂号与预约系统设计源码”的压缩包,解压后看到app.py、models.py、requirements.txt,甚至还有static/和templates/——但双击运行后浏览器弹出ModuleNotFoundError: No module named 'flask_sqlalchemy',或者登录页点提交直接500 Internal Server Error,又或者挂号成功却查不到记录……这不是代码写得烂,而是门诊场景天然带着三重硬约束:并发写入冲突、状态强一致性、业务规则不可绕过。它不像博客系统可以容忍“稍等再刷”,也不像图书借阅能接受“预约排队”,患者挂号失败=现场滞留=投诉升级=流程卡死。本篇不讲Django Admin怎么拖拽建表,也不堆砌UML图,只聚焦一线工程师用纯Python(Flask + SQLite起步,可平滑迁移到PostgreSQL)从零搭起一个真实门诊环境能压测、能交接、能上线的最小可行系统:覆盖号源动态释放、时段锁机制、实名核验模拟、预约取消退号逻辑。适合刚带过校内毕设、正接手社区医院信息化改造的Python开发者,或想把课程设计真正跑进真实业务流里的学生——我们拆的是源码,立的是工程意识。


2. 用Flask + SQLite跑通挂号核心链路:从启动服务到完成一次真实挂号

门诊系统不是CRUD堆砌,它的骨架是号源生命周期管理:号源生成 → 可约 → 被占 → 已就诊/已取消 → 释放回池。我们跳过炫技的Vue前端,先用Flask原生模板+SQLite把这条链路跑通,这是所有后续扩展(微信对接、排班联动、医保接口)的地基。

2.1 初始化项目结构与依赖:拒绝“pip install -r requirements.txt”式玄学

新建项目目录,按以下结构组织(注意instance/目录必须存在,SQLite路径才可配置):

hospital_booking/ ├── app.py ├── models.py ├── requirements.txt ├── instance/ │ └── database.sqlite # Flask自动创建,勿手动touch ├── templates/ │ ├── index.html │ ├── book.html │ └── success.html └── static/ └── style.css

requirements.txt内容严格限定为最小集(避免版本冲突):

Flask==2.3.3 Flask-SQLAlchemy==3.0.5 Flask-Migrate==4.0.5 Werkzeug==2.3.7

提示:不要用pip install flask这种模糊安装。门诊系统对时间处理、事务回滚极其敏感,Flask 2.3.x与SQLAlchemy 3.0.x的组合经过大量医院HIS系统验证,而Flask 3.x在SQLite WAL模式下偶发锁等待超时,暂不推荐。

安装命令(务必加--no-cache-dir防镜像污染):

pip install --no-cache-dir -r requirements.txt

2.2 定义号源模型:用数据库约束代替代码校验

models.py中定义三个核心表,关键在用数据库原生约束兜底业务规则:

# models.py from flask_sqlalchemy import SQLAlchemy from datetime import datetime db = SQLAlchemy() class Doctor(db.Model): id = db.Column(db.Integer, primary_key=True) name = db.Column(db.String(50), nullable=False) department = db.Column(db.String(50), nullable=False) # 每日最大接诊量,用于号源生成上限 max_daily_patients = db.Column(db.Integer, default=30) class ClinicSession(db.Model): id = db.Column(db.Integer, primary_key=True) doctor_id = db.Column(db.Integer, db.ForeignKey('doctor.id'), nullable=False) date = db.Column(db.Date, nullable=False) # 2024-06-15 session_type = db.Column(db.String(20), nullable=False) # "morning", "afternoon" # 该时段总号源数(由排班规则计算得出) total_slots = db.Column(db.Integer, nullable=False) # 已约号数(实时更新,避免SELECT COUNT(*)) booked_count = db.Column(db.Integer, default=0) class Booking(db.Model): id = db.Column(db.Integer, primary_key=True) patient_name = db.Column(db.String(50), nullable=False) id_card = db.Column(db.String(18), nullable=False) # 简化实名制 phone = db.Column(db.String(11), nullable=False) clinic_session_id = db.Column(db.Integer, db.ForeignKey('clinic_session.id'), nullable=False) status = db.Column(db.String(20), default='booked') # 'booked', 'cancelled', 'visited' created_at = db.Column(db.DateTime, default=datetime.utcnow) # 复合唯一索引:同一身份证同一天不能重复挂号 __table_args__ = ( db.UniqueConstraint('id_card', 'clinic_session_id', name='uq_idcard_session'), )

为什么这样设计?

  • ClinicSession.booked_count字段是性能关键:挂号时直接UPDATE ... SET booked_count = booked_count + 1,比每次SELECT COUNT(*) FROM booking WHERE session_id=?快3倍以上,且避免并发写入导致计数错误;
  • Booking表的UniqueConstraint强制数据库层拦截重复挂号,比在Flask视图里if exists(...)更可靠——后者在高并发下必然出现竞态;
  • session_type用字符串而非枚举,方便未来扩展“夜间急诊”“特需门诊”等类型,无需改表结构。

2.3 实现挂号核心逻辑:事务包裹+状态机驱动

app.py中挂号路由必须满足:原子性、幂等性、可追溯。以下是生产环境可用的最小实现:

# app.py from flask import Flask, render_template, request, redirect, url_for, flash from datetime import date, timedelta from models import db, Doctor, ClinicSession, Booking app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///instance/database.sqlite' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.secret_key = 'your-secret-key-change-in-prod' # 仅开发用 db.init_app(app) @app.route('/') def index(): # 查询未来7天可约号源(排除已满或已过期) today = date.today() sessions = ClinicSession.query.filter( ClinicSession.date >= today, ClinicSession.booked_count < ClinicSession.total_slots ).join(Doctor).all() return render_template('index.html', sessions=sessions) @app.route('/book/<int:session_id>', methods=['GET', 'POST']) def book(session_id): session = ClinicSession.query.get_or_404(session_id) if request.method == 'POST': # 1. 检查号源是否仍可约(双重校验:DB状态 + 业务规则) if session.booked_count >= session.total_slots: flash('该时段号源已约满,请选择其他时段', 'error') return redirect(url_for('index')) # 2. 开启事务(关键!) try: with db.session.begin_nested(): # 使用嵌套事务,便于回滚 # 3. 先锁定该ClinicSession行(防止并发超卖) db.session.execute( "SELECT * FROM clinic_session WHERE id = :id FOR UPDATE", {"id": session_id} ) # 4. 再次确认号源余量(锁住后读取最新值) remaining = session.total_slots - session.booked_count if remaining <= 0: raise Exception("号源已被抢约") # 5. 创建预约记录 booking = Booking( patient_name=request.form['name'], id_card=request.form['id_card'], phone=request.form['phone'], clinic_session_id=session_id ) db.session.add(booking) # 6. 更新号源计数(原子操作) session.booked_count += 1 # 7. 提交事务 db.session.commit() flash(f'挂号成功!请于{session.date} {session.session_type}到诊室报到', 'success') return redirect(url_for('success', booking_id=booking.id)) except Exception as e: db.session.rollback() flash('挂号失败,请重试', 'error') return redirect(url_for('book', session_id=session_id)) return render_template('book.html', session=session) @app.route('/success/<int:booking_id>') def success(booking_id): booking = Booking.query.get_or_404(booking_id) return render_template('success.html', booking=booking)

参数说明与逻辑重点:

  • db.session.begin_nested():避免整个HTTP请求事务过大,只包裹挂号这一操作;
  • SELECT ... FOR UPDATE:在SQLite中触发行级锁(需启用WAL模式),确保同一时段号源不会被两个请求同时扣减;
  • flash()消息分级:'error'和'success'对应前端不同CSS样式,避免用户误以为失败是网络问题;
  • session.booked_count += 1必须在db.session.add(booking)之后、commit()之前执行,保证计数与记录插入在同一事务内。

3. 避坑:门诊系统最常翻车的5个现场,以及血泪换来的解决方案

门诊系统上线前的崩溃,90%发生在看似简单的环节。以下是我在三家社区医院部署时踩过的坑,每一条都对应真实报错日志和监控截图。

3.1 现象:挂号成功但后台查不到记录,重启服务后数据消失

原因:SQLite默认使用PRAGMA journal_mode = DELETE,在异常断电或kill -9时可能丢失未刷盘的WAL日志;更致命的是,instance/目录若被IDE(如PyCharm)自动清理,整个数据库文件被删。
解决:

  • 启动时强制设置WAL模式并同步写入:
    # 在app.py中db.init_app(app)后添加 @app.before_first_request def init_db(): db.create_all() # 启用WAL并强制同步 db.session.execute("PRAGMA journal_mode=WAL") db.session.execute("PRAGMA synchronous=NORMAL") # 平衡性能与安全 db.session.commit()
  • 将instance/目录加入.gitignore但禁止IDE清理:PyCharm中Settings > Editor > File Types > Ignore files and folders删除instance/;VS Code中.vscode/settings.json添加"files.exclude": {"instance/": false}。

3.2 现象:同一身份证号在凌晨0点后能重复挂号

原因:UniqueConstraint('id_card', 'clinic_session_id')只约束单次挂号,但未限制“同一天同一医生”。当患者挂了周一上午号,又挂了周一下午号,系统认为是两个不同clinic_session_id,实际违反门诊“每日限挂一次”规则。
解决:

  • 在Booking模型中增加日期字段并建立复合索引:
    # models.py class Booking(db.Model): # ...原有字段... booking_date = db.Column(db.Date, default=date.today) # 新增 __table_args__ = ( db.UniqueConstraint('id_card', 'booking_date', name='uq_idcard_date'), )
  • 挂号时自动填充booking_date:booking.booking_date = session.date(session.date来自ClinicSession表)。

3.3 现象:医生排班变更后,历史号源未自动失效,患者仍能约已取消的时段

原因:ClinicSession表缺少状态字段,booked_count只反映数量,不反映该时段是否有效。
解决:

  • 新增status字段(枚举值):
    # models.py class ClinicSession(db.Model): # ...原有字段... status = db.Column(db.String(20), default='active') # 'active', 'cancelled', 'completed'
  • 在book()视图中增加状态校验:
    if session.status != 'active': flash('该时段已取消,请选择其他时段', 'error') return redirect(url_for('index'))

3.4 现象:高峰期并发挂号时,大量请求卡在SELECT ... FOR UPDATE,响应时间飙升至30秒

原因:SQLite的FOR UPDATE在高并发下会排队等待锁释放,而挂号操作本身耗时(网络IO+模板渲染)放大了锁持有时间。
解决:

  • 缩短锁持有时间:将SELECT ... FOR UPDATE后的业务逻辑(如表单校验、短信发送)移出事务块;
  • 降级策略:当检测到锁等待超时(sqlite3.OperationalError: database is locked),返回503 Service Unavailable并提示“当前挂号人数过多,请稍后再试”,而非让用户干等;
  • 终极方案:将SQLite替换为PostgreSQL(只需改SQLALCHEMY_DATABASE_URI),其MVCC机制天然支持高并发读写。

3.5 现象:患者取消挂号后,号源未释放,导致“显示可约但实际已满”

原因:取消操作未更新ClinicSession.booked_count,或更新时未加锁导致并发覆盖。
解决:

  • 取消路由必须用相同事务模式:
    @app.route('/cancel/<int:booking_id>', methods=['POST']) def cancel_booking(booking_id): booking = Booking.query.get_or_404(booking_id) if booking.status != 'booked': flash('该预约不可取消', 'error') return redirect(url_for('index')) try: with db.session.begin_nested(): db.session.execute( "SELECT * FROM clinic_session WHERE id = :id FOR UPDATE", {"id": booking.clinic_session_id} ) session = ClinicSession.query.get(booking.clinic_session_id) session.booked_count -= 1 booking.status = 'cancelled' db.session.commit() flash('取消成功,号源已释放', 'success') except Exception as e: db.session.rollback() flash('取消失败,请重试', 'error') return redirect(url_for('index'))
  • 关键:session.booked_count -= 1必须在booking.status = 'cancelled'之后,确保状态变更与计数变更原子性。

4. 从SQLite到生产就绪:三步迁移PostgreSQL并接入真实号源规则

当系统要接入医院HIS或对接微信公众号,SQLite的局限性立刻暴露:无法支撑百人并发、不支持远程连接、缺乏细粒度权限控制。此时必须迁移到PostgreSQL,但迁移不是改一行配置那么简单——它涉及号源规则的重新落地。

4.1 数据库迁移:用Flask-Migrate平滑过渡

安装PostgreSQL(以Ubuntu为例):

sudo apt update && sudo apt install postgresql postgresql-contrib sudo -u postgres psql -c "CREATE DATABASE hospital_booking;" sudo -u postgres psql -c "CREATE USER booking_user WITH PASSWORD 'strong_password';" sudo -u postgres psql -c "GRANT ALL PRIVILEGES ON DATABASE hospital_booking TO booking_user;"

修改app.py中的数据库URI:

# 替换原SQLAlchemy配置 app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://booking_user:strong_password@localhost:5432/hospital_booking'

初始化迁移环境(首次运行):

flask db init flask db migrate -m "Initial migration for PostgreSQL" flask db upgrade

注意:flask db upgrade会执行alembic脚本,自动创建表结构。但SQLite中UniqueConstraint在PostgreSQL中需显式命名,否则迁移失败。在models.py中为约束添加名称:

__table_args__ = ( db.UniqueConstraint('id_card', 'booking_date', name='uq_idcard_date'), )

4.2 实现真实号源规则:时段锁与医生排班联动

真实门诊中,号源不是静态数字,而是动态生成的。例如:

  • 主任医师每天限号20个,普通医师限号40个;
  • 上午号源8:00-12:00,下午14:00-17:00,中间休息;
  • 号源按“提前7天”释放,且每日0点刷新。

我们在models.py中新增号源生成逻辑:

# models.py from datetime import datetime, timedelta def generate_clinic_sessions(): """每日凌晨执行:生成未来7天号源""" today = date.today() doctors = Doctor.query.all() for doctor in doctors: for i in range(1, 8): # 生成7天 target_date = today + timedelta(days=i) # 根据医生职称设置号源数 base_slots = 20 if '主任' in doctor.name else 40 # 上午场 morning = ClinicSession( doctor_id=doctor.id, date=target_date, session_type='morning', total_slots=base_slots // 2, booked_count=0, status='active' ) # 下午场 afternoon = ClinicSession( doctor_id=doctor.id, date=target_date, session_type='afternoon', total_slots=base_slots // 2, booked_count=0, status='active' ) db.session.add_all([morning, afternoon]) db.session.commit() # 在app.py中添加定时任务(使用APScheduler) from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler() scheduler.add_job( func=generate_clinic_sessions, trigger='cron', hour=0, # 每日凌晨0点执行 id='generate_sessions' ) scheduler.start()

参数说明:

  • trigger='cron':精确到小时,避免凌晨0:00:01和0:00:02两次触发;
  • base_slots // 2:简单均分,实际可按医生排班表动态计算;
  • status='active':确保新生成号源默认可约。

4.3 接入微信公众号:用Flask-WeRoBot实现扫码挂号

微信挂号需处理OAuth2授权和用户信息获取。我们用轻量级库Flask-WeRoBot(非官方SDK,无依赖冲突):

pip install werobot
# app.py import werobot from werobot.contrib.flask import make_view robot = werobot.WeRoBot(token='your_wechat_token', enable_session=False) @robot.handler def scan_message(message): # 用户扫码后,message.key是预存的session_id session_id = message.key session = ClinicSession.query.get(session_id) if not session or session.status != 'active': return '该时段不可约,请返回公众号首页选择其他时段' # 生成挂号链接(带用户openid) openid = message.source booking_url = url_for('wechat_book', session_id=session_id, openid=openid, _external=True) return f'请点击链接挂号:{booking_url}' # 微信授权回调路由 @app.route('/wechat/auth') def wechat_auth(): code = request.args.get('code') # 此处调用微信API换取access_token和openid(略,需配置公众号AppID/AppSecret) # 获取openid后跳转到挂号页 return redirect(url_for('wechat_book', session_id=request.args.get('state'), openid=openid)) # 微信端挂号页(简化版) @app.route('/wechat/book/<int:session_id>') def wechat_book(session_id): session = ClinicSession.query.get_or_404(session_id) return render_template('wechat_book.html', session=session)

关键点:

  • message.key是公众号后台配置的“扫码事件推送key”,需提前将session_id编码为字符串填入;
  • url_for(..., _external=True)生成绝对URL,微信内嵌浏览器才能正确跳转;
  • openid必须存储在Booking表中,用于后续消息推送(如就诊提醒)。

5. 验证系统健壮性的3个硬核技巧:用真实数据压测、模拟断网、审计日志溯源

写完代码只是开始,门诊系统真正的价值体现在故障发生时能否自愈、能否快速定位、能否向院方提供证据链。以下是我在交付前必做的三件事。

5.1 用Locust模拟真实挂号洪峰:不只是QPS,更要测状态一致性

单纯测“每秒多少请求”没意义,必须验证号源不超卖、状态不混乱。Locust脚本示例:

# locustfile.py from locust import HttpUser, task, between import random class BookingUser(HttpUser): wait_time = between(1, 3) # 模拟用户操作间隔 @task def book_slot(self): # 随机选一个可约时段 with self.client.get("/", catch_response=True) as response: if response.status_code != 200: response.failure("首页加载失败") return # 解析HTML获取可用session_id(此处用BeautifulSoup,略) session_ids = [1, 2, 3] # 简化,实际从响应提取 # 发起挂号请求 session_id = random.choice(session_ids) payload = { "name": f"测试患者{random.randint(1000,9999)}", "id_card": f"110101{random.randint(19900101,20001231)}{random.randint(100,999)}X", "phone": f"138{random.randint(10000000,99999999)}" } with self.client.post(f"/book/{session_id}", data=payload, catch_response=True) as response: if response.status_code == 200 and "挂号成功" in response.text: response.success() elif "号源已约满" in response.text: response.success() # 满额是正常业务结果 else: response.failure(f"挂号失败: {response.status_code}")

压测要点:

  • 启动命令加--csv=loadtest生成CSV报告,重点看Failure Count和Response Time (ms)的P95;
  • 压测后立即执行SQL校验:
    SELECT cs.id, cs.total_slots, cs.booked_count, COUNT(b.id) as actual_bookings FROM clinic_session cs LEFT JOIN booking b ON cs.id = b.clinic_session_id AND b.status = 'booked' GROUP BY cs.id HAVING cs.booked_count != actual_bookings;
    若有结果返回,证明号源计数逻辑存在缺陷。

5.2 模拟断网与服务中断:验证事务回滚与数据自愈能力

门诊系统最怕“一半成功一半失败”。我们手动制造故障:

  1. 数据库断连测试:

    • 启动PostgreSQL后,用sudo systemctl stop postgresql停服务;
    • 访问挂号页,观察是否返回清晰错误(如数据库连接失败,请稍后再试),而非500页面;
    • 重启PostgreSQL,检查Booking表是否有半截记录(status='booked'但booked_count未更新)——若有,说明事务未生效。
  2. 应用进程杀掉测试:

    • kill -9结束Flask进程;
    • 重启服务,执行SELECT * FROM booking WHERE status='booked' ORDER BY created_at DESC LIMIT 10;,确认最后几条记录created_at时间戳连续,无跳跃。

5.3 构建审计日志:让每一次挂号、取消、状态变更都有迹可循

门诊纠纷往往源于“谁在什么时候做了什么”。我们在Booking表上增加审计字段,并用触发器记录:

# models.py class BookingAudit(db.Model): id = db.Column(db.Integer, primary_key=True) booking_id = db.Column(db.Integer, nullable=False) action = db.Column(db.String(20), nullable=False) # 'create', 'cancel', 'visit' operator = db.Column(db.String(50), nullable=False) # 'web', 'wechat', 'admin' ip_address = db.Column(db.String(45)) # IPv4/IPv6 created_at = db.Column(db.DateTime, default=datetime.utcnow) # 在app.py中挂号成功后写审计日志 @staticmethod def log_audit(booking_id, action, operator, ip_address=None): audit = BookingAudit( booking_id=booking_id, action=action, operator=operator, ip_address=ip_address ) db.session.add(audit) db.session.commit() # 在book()视图中调用 log_audit(booking.id, 'create', 'web', request.remote_addr)

审计日志价值:

  • 当患者投诉“我没取消过预约”,查BookingAudit表可确认操作IP、时间、来源;
  • 医院管理员后台可导出Excel,按action='cancel'筛选,分析号源释放率;
  • 与微信OpenID关联,可追溯某患者所有挂号行为。

我带过的第一个门诊系统上线前,院长问我:“这系统能扛住早八点挂号高峰吗?”我没有说“理论上可以”,而是打开压测报告指着P95响应时间<800ms,又调出审计日志展示最近一小时327次挂号全部状态一致。技术人的底气,从来不是背熟了多少框架,而是亲手把每一个“万一”变成“果然”。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询