手写实现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 uuid
class Status(Enum):
PENDING = pending # 待处理
APPROVED = approved # 已通过
REJECTED = rejected # 已驳回
CANCELLED = cancelled # 已撤销
@dataclass
class AuditLog:
审计日志：记录谁在什么时间做了什么操作
request_id: str
operator: str
action: str
timestamp: 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.PENDING
history: 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 time
class OAService:
def __init__(self):
# 内存模拟数据库，生产环境替换为DB
self.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(无法确定审批链，请检查配置)
# 这里简化处理：直接取第一个角色名作为审批人标识
# 实际项目中，可能需要查询该角色下的具体用户ID
first_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] = request
return req_id
def approve(self, request_id: str, operator: str, comment: str = ):
审批通过：核心逻辑
1. 校验权限
2. 更新状态
3. 计算下一节点
req = self.db.get(request_id)
if not req:
raise Exception(单据不存在)
# 1. 权限校验：操作人必须是当前指定的审批人
# 注意：这里简化为角色匹配，实际需校验用户ID
operator_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.PENDING
else:
# 到达终点，最终通过
req.current_approver =
req.status = models.Status.APPROVED
def 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.REJECTED
req.current_approver =
避坑指南：
很多新手在 approve 方法里直接改 req.status，却忘了判断是不是最后一层。如果没判断，单据会一直流转下去，或者卡在某个节点。务必检查 next_roles 是否为空。
运行与测试：用代码说话
光看代码不运行，等于没懂。咱们写个简单的测试脚本，模拟一个完整的审批流。
# main.py
from service import OAService
import models
def 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）还是像上面这样手写逻辑？评论区交流，看看哪种更适合你的项目场景。 
长河君 · 2026/9/23 7:39:49 · 阅读 1.2k 手写实现OA选型核心逻辑，3步搞定面试高频坑
手写实现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 uuid
class Status(Enum):
PENDING = pending # 待处理
APPROVED = approved # 已通过
REJECTED = rejected # 已驳回
CANCELLED = cancelled # 已撤销
@dataclass
class AuditLog:
审计日志：记录谁在什么时间做了什么操作
request_id: str
operator: str
action: str
timestamp: 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.PENDING
history: 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 time
class OAService:
def __init__(self):
# 内存模拟数据库，生产环境替换为DB
self.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(无法确定审批链，请检查配置)
# 这里简化处理：直接取第一个角色名作为审批人标识
# 实际项目中，可能需要查询该角色下的具体用户ID
first_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] = request
return req_id
def approve(self, request_id: str, operator: str, comment: str = ):
审批通过：核心逻辑
1. 校验权限
2. 更新状态
3. 计算下一节点
req = self.db.get(request_id)
if not req:
raise Exception(单据不存在)
# 1. 权限校验：操作人必须是当前指定的审批人
# 注意：这里简化为角色匹配，实际需校验用户ID
operator_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.PENDING
else:
# 到达终点，最终通过
req.current_approver =
req.status = models.Status.APPROVED
def 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.REJECTED
req.current_approver =
避坑指南：
很多新手在 approve 方法里直接改 req.status，却忘了判断是不是最后一层。如果没判断，单据会一直流转下去，或者卡在某个节点。务必检查 next_roles 是否为空。
运行与测试：用代码说话
光看代码不运行，等于没懂。咱们写个简单的测试脚本，模拟一个完整的审批流。
# main.py
from service import OAService
import models
def 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）还是像上面这样手写逻辑？评论区交流，看看哪种更适合你的项目场景。