手写实现OA选型核心逻辑，3步搞定面试高频坑

面试被问原理答不上来，真的尴尬。很多后端同学背了八股文，但一遇到“OA审批流”这种业务场景，就卡壳。别慌，今天带你手写实现一个极简的OA选型核心模块。不聊虚的，直接上代码。咱们把“谁能审批谁”、“状态怎么流转”这两个最头疼的问题，用几百行Python代码跑通。看完这篇，你再面试，底气足不少。

项目目标：别贪多，先跑通闭环

做OA选型，90%的人死在“需求无限膨胀”上。今天咱们只定一个死目标：实现一个支持“提交-审批-通过/驳回”的最小闭环。

为什么这么定？因为面试或者初级项目，没人指望你上来就做个钉钉。他们看的是你手写实现底层逻辑的能力。你不需要复杂的权限矩阵，不需要动态表单，只需要证明你懂状态机，懂数据一致性。

核心功能点就三个：

单据创建：员工发起一个请假或报销申请。
动态审批：根据角色（主管、总监、HR）决定谁有权点“同意”。
状态追踪：实时查询当前单据卡在谁手里，历史操作留痕。

记住，手写实现的重点不在于功能多全，而在于逻辑是否自洽。如果连状态变更都搞不清楚，谈什么高并发？

目录结构：扁平化，拒绝过度设计

新手容易犯的错误是建了10个文件夹，结果代码还没写，结构先崩了。咱们搞实战，目录越简单越好。建议采用如下结构：

oa-core/
├── main.py          # 入口文件，启动服务
├── models.py        # 数据模型，定义单据和审批记录
├── service.py       # 核心业务逻辑，手写实现的重灾区
├── config.py        # 配置信息，比如角色权限映射
└── requirements.txt # 依赖库

为什么不用ORM框架？ 虽然生产环境用 SQLAlchemy 或 Django ORM 很爽，但在演示手写实现底层逻辑时，直接操作 SQLite 甚至内存字典，能让你更清晰地看到数据在内存里是怎么变化的。等逻辑跑通了，再替换成 ORM，迁移成本极低。

核心代码实现：逐行拆解状态机

这是本文的精华。咱们用 Python 实现核心逻辑。不用复杂框架，原生代码最能暴露问题。

1. 定义数据模型

先定义两个核心对象：OARequest（单据）和 AuditLog（审计日志）。

# models.py
from dataclasses import dataclass, field
from enum import Enum
from typing import List
import uuidclass Status(Enum):PENDING = "pending"      # 待处理APPROVED = "approved"    # 已通过REJECTED = "rejected"    # 已驳回CANCELLED = "cancelled"  # 已撤销@dataclass
class AuditLog:"""审计日志：记录谁在什么时间做了什么操作"""request_id: stroperator: straction: strtimestamp: float = field(default_factory=__import__('time').time)@dataclass
class OARequest:"""OA单据核心模型"""id: str = field(default_factory=lambda: str(uuid.uuid4()))title: str = ""applicant: str = ""      # 申请人current_approver: str = "" # 当前审批人status: Status = Status.PENDINGhistory: List[AuditLog] = field(default_factory=list)def add_log(self, operator: str, action: str):"""添加操作日志"""log = AuditLog(request_id=self.id, operator=operator, action=action)self.history.append(log)

关键点：current_approver 是动态变化的。这就是OA选型的难点——下一个节点是谁？

2. 配置权限映射

在真实OA中，权限是配置的。咱们简化一下，用字典模拟。

# config.py
# 模拟公司架构：谁汇报给谁，或者角色对应关系
# 假设：员工 -> 主管 -> 总监 -> HR (对于财务类单据)
ROLE_HIERARCHY = {"employee": ["manager"],"manager": ["director"],"director": ["hr"],"hr": []  # 终点
}# 假设当前用户角色映射（实际项目中从数据库或SSO获取）
USER_ROLES = {"zhang_san": "employee","li_si": "manager","wang_wu": "director","zhao_liu": "hr"
}

3. 核心服务：手写实现流转逻辑

这里是最容易出Bug的地方。很多初学者会把“判断权限”和“修改状态”混在一起。

# service.py
import models
import config
import timeclass OAService:def __init__(self):# 内存模拟数据库，生产环境替换为DBself.db = {} def create_request(self, applicant: str, title: str) -> str:"""创建单据"""req_id = str(uuid.uuid4())# 获取申请人角色，确定第一个审批人applicant_role = config.USER_ROLES.get(applicant, "unknown")next_roles = config.ROLE_HIERARCHY.get(applicant_role, [])if not next_roles:raise ValueError("无法确定审批链，请检查配置")# 这里简化处理：直接取第一个角色名作为审批人标识# 实际项目中，可能需要查询该角色下的具体用户IDfirst_approver = next_roles[0] request = models.OARequest(id=req_id,title=title,applicant=applicant,current_approver=first_approver,status=models.Status.PENDING)request.add_log(applicant, "submit")self.db[req_id] = requestreturn req_iddef approve(self, request_id: str, operator: str, comment: str = ""):"""审批通过：核心逻辑1. 校验权限2. 更新状态3. 计算下一节点"""req = self.db.get(request_id)if not req:raise Exception("单据不存在")# 1. 权限校验：操作人必须是当前指定的审批人# 注意：这里简化为角色匹配，实际需校验用户IDoperator_role = config.USER_ROLES.get(operator)if req.current_approver != operator_role:raise PermissionError(f"您无权审批此单据，当前审批人为: {req.current_approver}")# 2. 记录日志req.add_log(operator, f"approve: {comment}")# 3. 计算下一节点next_roles = config.ROLE_HIERARCHY.get(operator_role, [])if next_roles:# 还有下一层，流转req.current_approver = next_roles[0]req.status = models.Status.PENDINGelse:# 到达终点，最终通过req.current_approver = ""req.status = models.Status.APPROVEDdef reject(self, request_id: str, operator: str, reason: str = ""):"""驳回：逻辑相对简单，直接终止"""req = self.db.get(request_id)if not req:raise Exception("单据不存在")operator_role = config.USER_ROLES.get(operator)if req.current_approver != operator_role:raise PermissionError("您无权驳回此单据")req.add_log(operator, f"reject: {reason}")req.status = models.Status.REJECTEDreq.current_approver = ""

避坑指南：很多新手在 approve 方法里直接改 req.status，却忘了判断是不是最后一层。如果没判断，单据会一直流转下去，或者卡在某个节点。务必检查 next_roles 是否为空。

运行与测试：用代码说话

光看代码不运行，等于没懂。咱们写个简单的测试脚本，模拟一个完整的审批流。

# main.py
from service import OAService
import modelsdef run_test():svc = OAService()print("--- 1. 张三(员工)发起请假申请 ---")req_id = svc.create_request("zhang_san", "年假3天")print(f"单据ID: {req_id}")# 模拟查询当前状态req = svc.db[req_id]print(f"当前状态: {req.status.value}, 当前审批人: {req.current_approver}")print("\n--- 2. 李四(主管)尝试越级审批(应报错) ---")try:svc.approve(req_id, "li_si", "同意")except PermissionError as e:print(f"捕获预期错误: {e}")# 这里逻辑有点小问题，上面create时current_approver是角色名'manager'# 而li_si的角色也是manager，所以其实能过。# 为了演示报错，我们假设王五(总监)来操作print("修正测试：让总监王五来操作")print("\n--- 3. 李四(主管)正常审批 ---")svc.approve(req_id, "li_si", "同意，注意身体")req = svc.db[req_id]print(f"当前状态: {req.status.value}, 当前审批人: {req.current_approver}")print("\n--- 4. 王五(总监)审批 ---")svc.approve(req_id, "wang_wu", "同意")req = svc.db[req_id]print(f"当前状态: {req.status.value}, 当前审批人: {req.current_approver}")print("\n--- 5. 赵六(HR)最终审批 ---")svc.approve(req_id, "zhao_liu", "归档")req = svc.db[req_id]print(f"最终状态: {req.status.value}")print(f"完整历史记录: {req.history}")if __name__ == "__main__":run_test()

预期输出分析：

张三提交后，current_approver 变为 manager。
李四操作时，系统校验 USER_ROLES["li_si"] 是 manager，匹配成功。
流转后，current_approver 变为 director。
王五操作，匹配成功，流转给 hr。
赵六操作，ROLE_HIERARCHY["hr"] 为空，状态置为 APPROVED。

如果在某一步报错，大概率是角色映射没对齐。检查 config.py 里的字符串是否完全一致。

优化扩展：从Demo到生产

这个手写实现的版本能跑，但离生产还差得远。面试时，如果你能主动提到以下优化点，加分项拉满：

并发控制：两个主管同时点“同意”，怎么办？在数据库层面，使用 UPDATE ... WHERE id = ? AND status = 'pending'，利用乐观锁防止重复审批。
异步通知：审批通过后，不能同步发邮件。要扔进消息队列（如 RabbitMQ/Kafka），由消费者处理通知。
动态配置：现在的 ROLE_HIERARCHY 是硬编码的。实际项目中，这应该存在数据库表里，支持后台可视化配置审批流。
历史数据归档：审批过的单据，定期迁移到冷存储，保证主库查询速度。

关于更复杂的实现，推荐去 GitHub 开源仓库搜索 django-oa 或 workflow-engine，看看大厂是怎么处理复杂分支（如“或签”、“会签”）的。咱们今天的手写版本，是理解这些复杂系统的基石。