手写实现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, fapprove: {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, freject: {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，看看大厂是怎么处理复杂分支（如“或签”、“会签”）的。咱们今天的手写版本，是理解这些复杂系统的基石。
小结
OA选型的核心，不是选哪个框架，而是梳理清楚业务状态流转。
通过手写实现这个极简版本，你掌握了：如何用状态机管理单据生命周期。
如何解耦“权限校验”与“状态变更”。
如何通过审计日志实现全链路追踪。下次面试再被问“OA怎么做的”，别背概念。直接说：“我手写实现过一个基于状态机的核心模块，解决了并发审批和数据一致性问题……” 这就叫懂行。
你更常用哪种写法？是倾向于用状态机库（如 Transitions）还是像上面这样手写逻辑？评论区交流，看看哪种更适合你的项目场景。
