首页/资讯中心/详情

Python Django/Flask医疗预约系统开发实践

Python Django/Flask医疗预约系统开发实践
📅 发布时间：2026/8/3 16:41:39

1. 项目概述

医疗预约与诊断系统是医疗机构数字化转型的核心基础设施。这个基于Python Django/Flask框架开发的系统，本质上解决了传统医疗场景中的三大痛点：患者排队时间长、医生资源分配不均、病历信息管理混乱。我在三甲医院信息化建设项目中，曾亲眼见证过一套优秀的预约系统如何将门诊效率提升40%以上。

选择Django+Flask的组合方案，是经过实际项目验证的成熟架构。Django自带完善的Admin后台和ORM，能快速搭建基础数据模型；Flask的轻量级特性则非常适合开发灵活的API接口。这种组合既保证了开发效率，又能满足医疗系统对稳定性和扩展性的严苛要求。

2. 系统架构设计

2.1 技术栈选型分析

核心框架采用Django 4.1 + Flask 2.2的组合方案，主要基于以下考量：

  • Django的认证系统（django.contrib.auth）可直接复用，满足医疗系统严格的权限控制需求
  • Flask的Blueprint功能可以模块化开发预约、诊断、报表等业务组件
  • 使用Django ORM管理基础数据模型（患者、医生、科室等）
  • 通过Flask-RESTful构建REST API供移动端调用

数据库选用PostgreSQL 14，因其具有：

  • 完善的JSON字段支持（存储动态病历模板）
  • 良好的地理空间扩展（用于分院区管理）
  • 优于MySQL的事务处理能力（挂号支付场景必需）

2.2 核心模块划分

系统采用微服务架构设计，主要包含以下服务：

服务模块技术实现关键功能
用户中心Django患者/医生注册、权限管理
预约服务Flask + Celery号源管理、智能分诊
诊断系统Django + Vue.js电子病历、处方开具
支付网关Flask + Alipay SDK挂号费/诊费在线支付
数据分析Django ORM + Pandas就诊量统计、医生绩效分析

3. 关键功能实现

3.1 智能预约调度算法

核心算法采用改进的时间片轮转策略，在services/scheduling.py中实现：

def generate_slots(doctor_id, date): """ 生成可预约时间段 算法逻辑： 1. 读取医生基础坐班时间 2. 排除已预约时段 3. 动态调整时段长度（初诊15min，复诊10min） 4. 保留20%号源用于现场挂号 """ base_slots = DoctorSchedule.objects.get(doctor=doctor_id) appointments = Appointment.objects.filter( doctor=doctor_id, date=date ).values_list('time_slot', flat=True) available_slots = [] for slot in base_slots: if slot not in appointments: # 动态调整时段长度 duration = 15 if is_first_visit(patient) else 10 available_slots.append({ 'start': slot, 'end': slot + timedelta(minutes=duration), 'type': 'online' if len(available_slots) < 0.8*base_slots else 'onsite' }) return available_slots

重要提示：医疗系统必须考虑线下场景，永远不要将号源100%开放线上预约，需保留现场挂号通道。

3.2 电子病历模板系统

采用Django的JSONField实现动态表单，关键模型设计：

class MedicalRecord(models.Model): PATIENT_TYPE_CHOICES = [ ('OUT', '门诊'), ('IN', '住院'), ] patient = models.ForeignKey(Patient, on_delete=models.CASCADE) doctor = models.ForeignKey(Doctor, on_delete=models.PROTECT) record_type = models.CharField(max_length=3, choices=PATIENT_TYPE_CHOICES) template = models.ForeignKey(RecordTemplate, on_delete=models.SET_NULL) content = models.JSONField() # 存储结构化病历数据 created_at = models.DateTimeField(auto_now_add=True) def render_html(self): """将JSON数据渲染为HTML病历""" return render_template( f'records/{self.template.name}.html', data=self.content )

4. 安全与合规实现

4.1 医疗数据加密方案

采用双层加密策略确保数据安全：

  1. 数据库层面：使用PostgreSQL的pgcrypto扩展

    CREATE EXTENSION pgcrypto; UPDATE patients SET id_card = pgp_sym_encrypt(id_card, 'encryption_key');

  2. 应用层面：Django信号机制自动加密

    @receiver(pre_save, sender=Patient) def encrypt_sensitive_data(sender, instance, **kwargs): if instance.id_card: instance.id_card = encrypt(instance.id_card)

4.2 审计日志实现

符合医疗信息系统等保三级要求：

class AuditLog(models.Model): ACTION_CHOICES = [ ('VIEW', '查看'), ('EDIT', '修改'), ('DELETE', '删除'), ] user = models.ForeignKey(User, on_delete=models.PROTECT) action = models.CharField(max_length=6, choices=ACTION_CHOICES) model = models.CharField(max_length=50) object_id = models.CharField(max_length=36) ip_address = models.GenericIPAddressField() timestamp = models.DateTimeField(auto_now_add=True) @classmethod def log(cls, request, action, obj): cls.objects.create( user=request.user, action=action, model=obj.__class__.__name__, object_id=str(obj.pk), ip_address=get_client_ip(request) )

5. 性能优化实践

5.1 预约高并发处理

使用Redis + Celery实现：

# tasks.py @app.task(bind=True, rate_limit='100/m') def make_appointment(self, patient_id, slot_id): try: with transaction.atomic(): slot = TimeSlot.objects.select_for_update().get(pk=slot_id) if slot.status == 'AVAILABLE': Appointment.objects.create( patient_id=patient_id, time_slot=slot, status='PENDING_PAYMENT' ) slot.status = 'RESERVED' slot.save() return True except Exception as e: self.retry(exc=e, countdown=60)

5.2 数据库查询优化

针对高频查询的优化措施：

  1. 使用select_related预加载外键关系

    Appointment.objects.select_related('patient', 'doctor').filter(date=today)

  2. 对科室表添加复合索引

    class Department(models.Model): class Meta: indexes = [ models.Index(fields=['hospital', 'name']), ]

  3. 使用Django的prefetch_related优化多对多查询

    Doctor.objects.prefetch_related('specialties').filter(department=dept)

6. 部署架构建议

6.1 生产环境配置

推荐使用Docker Swarm或Kubernetes部署：

# docker-compose.prod.yml services: web: image: registry.example.com/medical-app environment: - DATABASE_URL=postgres://user:pass@db:5432/medical - REDIS_URL=redis://redis:6379/0 deploy: replicas: 3 resources: limits: cpus: '2' memory: 2G celery: image: registry.example.com/medical-app command: celery -A core worker -l INFO deploy: replicas: 2

6.2 监控方案

使用Prometheus + Grafana监控关键指标：

  1. Django应用指标：django-prometheus

    INSTALLED_APPS += ['django_prometheus'] MIDDLEWARE.insert(0, 'django_prometheus.middleware.PrometheusBeforeMiddleware')

  2. 自定义业务指标：

    from prometheus_client import Counter APPOINTMENT_CREATED = Counter( 'appointment_created_total', 'Total created appointments', ['department'] ) @api_view(['POST']) def create_appointment(request): APPOINTMENT_CREATED.labels(department=dept.name).inc()

7. 常见问题解决方案

7.1 号源冲突处理

采用乐观锁机制防止超卖：

def reserve_slot(slot_id): slot = TimeSlot.objects.get(pk=slot_id) if slot.status == 'AVAILABLE': rows = TimeSlot.objects.filter( pk=slot.pk, status='AVAILABLE' ).update(status='RESERVED') if rows == 0: raise ConcurrentModificationError()

7.2 医保对接方案

典型医保接口封装示例：

class MedicalInsurance: def __init__(self, config): self.wsdl = config['wsdl'] def verify_patient(self, id_card, medical_card): """实名认证""" client = zeep.Client(wsdl=self.wsdl) return client.service.verify( id_card=id_card, medical_card=medical_card ) def submit_bill(self, appointment_id): """医保结算""" appointment = Appointment.objects.get(pk=appointment_id) items = [{ 'item_code': 'REG', 'fee': appointment.fee }] return client.service.submit( patient_id=appointment.patient.medical_card, items=items )

这套系统在实际部署时需要特别注意医疗行业的特殊要求：所有数据库操作必须开启事务，关键业务日志至少保留5年，患者敏感信息需要在前端展示时自动脱敏（如身份证号显示为110**********1234）。我在某三甲医院实施时，曾因为忽略医保系统的异步回调机制导致账目不平，后来通过添加补偿事务机制解决了这个问题。