3天搞定小企业做账系统，搞定高频面试题
3天搞定小企业做账系统，搞定高频面试题
你刚把网上抄的记账代码跑起来，结果报错“字段缺失”，改了一晚上没思路，这简直是复制来的代码跑不通不知道怎么调的典型。很多后端开发想转全栈或做独立开发，总卡在业务逻辑和底层实现的衔接上，这不仅是实战难点，也是各大厂高频面试题里关于“系统设计与数据一致性”的考察重点。
别慌，今天咱们不整虚的，直接从一个小企业做账的真实场景出发，用 Python 从零手写一个轻量级记账系统。这不仅能解决你“调不通”的困惑，还能让你彻底搞懂复式记账法在代码里怎么落地，顺便把面试里常问的“如何处理并发下的账务一致性”给盘明白。
项目目标
咱们做的不是那种几百万行代码的 SAP 或 Oracle，而是一个面向小微企业的小企业做账核心模块。目标很明确：支持基础复式记账：有借必有贷，借贷必相等。这是会计铁律，也是代码逻辑的核心约束。
数据强一致性：在多线程或高并发场景下，确保每一笔交易原子性提交，不能出现“借了钱但贷没记上”的脏数据。
审计追踪：每一笔账都要有迹可循，符合财务合规要求。这里我们参考 RFC 7231 (Hypertext Transfer Protocol -- HTTP/1.1) 中关于幂等性和状态码的设计思想，确保接口调用的可重复性和错误处理的标准化，虽然记账系统通常是内部调用，但遵循通用的网络协议规范能让系统更健壮，尤其是在未来扩展 API 供外部 ERP 对接时。目录结构
为了保持代码的可维护性，我们采用分层架构。别小看目录结构，高频面试题里经常问“如何设计一个高可用的财务系统”，目录结构的合理性直接体现你对模块解耦的理解。
accounting_system/
├── app/
│ ├── __init__.py
│ ├── main.py # 入口文件
│ ├── models/
│ │ ├── __init__.py
│ │ ├── account.py # 账户模型
│ │ └── entry.py # 记账分录模型
│ ├── services/
│ │ ├── __init__.py
│ │ └── bookkeeper.py # 核心记账服务
│ └── utils/
│ ├── __init__.py
│ └── db.py # 数据库连接
├── tests/
│ └── test_bookkeeper.py # 单元测试
├── requirements.txt
└── README.mdmodels：定义数据结构，对应数据库表。
services：核心业务逻辑，这里是“重灾区”，也是调试最频繁的地方。
utils：工具类，如数据库连接池管理。核心代码实现
这是最关键的环节。很多新手写的代码，问题出在事务处理和数据校验上。我们一步步来。
1. 数据模型定义
使用 SQLAlchemy 作为 ORM，它能让代码更 Pythonic。
# app/models/account.py
from sqlalchemy import Column, Integer, String, Float, DateTime
from datetime import datetime
from app.utils.db import Baseclass Account(Base):__tablename__ = 'accounts'id = Column(Integer, primary_key=True, index=True)name = Column(String(50), unique=True, nullable=False)type = Column(String(10), nullable=False) # 'asset', 'liability', 'equity', 'income', 'expense'balance = Column(Float, default=0.0)created_at = Column(DateTime, default=datetime.utcnow)# app/models/entry.py
from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from datetime import datetime
from app.utils.db import Baseclass JournalEntry(Base):__tablename__ = 'journal_entries'id = Column(Integer, primary_key=True, index=True)description = Column(String(100))created_at = Column(DateTime, default=datetime.utcnow)# 一对多关系：一个分录包含多条明细lines = relationship(JournalEntryLine, back_populates=entry, cascade=all, delete-orphan)class JournalEntryLine(Base):__tablename__ = 'journal_entry_lines'id = Column(Integer, primary_key=True, index=True)entry_id = Column(Integer, ForeignKey('journal_entries.id'))account_id = Column(Integer, ForeignKey('accounts.id'))amount = Column(Float, nullable=False) # 正数为借方，负数为贷方，或者用两个字段，这里简化用正负direction = Column(String(10), nullable=False) # 'debit' or 'credit'entry = relationship(JournalEntry, back_populates=lines)account = relationship(Account)注意：这里 direction 和 amount 的设计是为了清晰。在实际生产环境中，建议金额字段始终为正，通过 direction 区分借贷，避免负数带来的逻辑混淆。
2. 核心记账服务：解决“跑不通”的关键
很多代码跑不通，是因为在内存里算对了，往库里写的时候没加锁或没开事务。看这个 bookkeeper.py：
# app/services/bookkeeper.py
from sqlalchemy.orm import Session
from app.models.account import Account
from app.models.entry import JournalEntry, JournalEntryLine
from datetime import datetime
import logginglogger = logging.getLogger(__name__)class BookkeeperService:def __init__(self, db: Session):self.db = dbdef create_journal_entry(self, description: str, lines_data: list) - JournalEntry:创建记账分录lines_data: list of dict, e.g., [{'account_id': 1, 'amount': 100, 'direction': 'debit'}, ...]# 1. 前置校验：借贷平衡检查total_debit = 0.0total_credit = 0.0for line in lines_data:if line['direction'] == 'debit':total_debit += line['amount']elif line['direction'] == 'credit':total_credit += line['amount']else:raise ValueError(Invalid direction: must be 'debit' or 'credit')# 浮点数精度问题处理：在实际财务系统中，严禁使用 float，应使用 Decimal# 这里为了演示方便使用 float，但面试时要指出这一点！if abs(total_debit - total_credit) 0.01: raise ValueError(Debit and Credit must be equal)# 2. 开启事务try:# 创建分录头entry = JournalEntry(description=description, created_at=datetime.utcnow())self.db.add(entry)# 创建分录行并更新账户余额for line_data in lines_data:account = self.db.query(Account).filter_by(id=line_data['account_id']).with_for_update().first()if not account:raise ValueError(fAccount ID {line_data['account_id']} not found)line = JournalEntryLine(account_id=account.id,amount=line_data['amount'],direction=line_data['direction'])entry.lines.append(line)# 更新账户余额if line_data['direction'] == 'debit':account.balance += line_data['amount']else:account.balance -= line_data['amount']# 3. 提交事务self.db.commit()self.db.refresh(entry)return entryexcept Exception as e:# 4. 异常回滚self.db.rollback()logger.error(fError creating journal entry: {str(e)})raisedef get_balance(self, account_id: int) - float:account = self.db.query(Account).filter_by(id=account_id).first()if not account:return 0.0return account.balance逐行解析与避坑：with_for_update()：这是解决并发问题的关键。在查询账户时加上行锁，防止两个线程同时读取同一个账户余额并分别更新，导致数据不一致。这是高频面试题中“乐观锁 vs 悲观锁”的实战应用。
事务的 Commit 与 Rollback：务必在一个 try-except 块中处理。如果中间任何一步失败（比如某个账户不存在），必须回滚所有已修改的数据，保证原子性。
浮点数陷阱：代码中用了 abs(total_debit - total_credit) 0.01 来做平衡检查。这是因为 0.1 + 0.2 != 0.3 在计算机里是常识。在真实财务项目中，必须使用 decimal.Decimal 类来存储金额，否则你会因为几分钱的误差导致账务不平，这在审计时是致命错误。运行与测试
写完代码，别急着跑，先写测试。单元测试能帮你提前发现 90% 的逻辑错误，而不是等到上线后用户投诉。
# tests/test_bookkeeper.py
import pytest
from app.services.bookkeeper import BookkeeperService
from app.utils.db import SessionLocal
from app.models.account import Account
from app.models.entry import JournalEntry
from sqlalchemy import create_engine
from app.utils.db import Baseengine = create_engine(sqlite:///./test_accounting.db)
Base.metadata.create_all(bind=engine)@pytest.fixture
def client():db = SessionLocal()try:yield BookkeeperService(db)finally:db.close()def test_create_balanced_entry(client):# 准备数据：创建一个现金账户cash = Account(name=Cash, type=asset, balance=0)client.db.add(cash)client.db.commit()cash_id = cash.id# 创建一个收入账户revenue = Account(name=Revenue, type=income, balance=0)client.db.add(revenue)client.db.commit()revenue_id = revenue.id# 执行记账：借现金 100，贷收入 100lines = [{'account_id': cash_id, 'amount': 100, 'direction': 'debit'},{'account_id': revenue_id, 'amount': 100, 'direction': 'credit'}]entry = client.create_journal_entry(Initial Cash, lines)# 断言assert entry.id is not Noneassert client.get_balance(cash_id) == 100assert client.get_balance(revenue_id) == -100 # 注意：收入类账户贷方增加，余额逻辑需根据科目类型调整，此处简化def test_unbalanced_entry_raises_error(client):cash = Account(name=Cash, type=asset, balance=0)client.db.add(cash)client.db.commit()cash_id = cash.idlines = [{'account_id': cash_id, 'amount': 100, 'direction': 'debit'},# 缺少贷方]with pytest.raises(ValueError) as excinfo:client.create_journal_entry(Bad Entry, lines)assert Debit and Credit must be equal in str(excinfo.value)调试技巧：
如果测试跑不通，检查以下几点：数据库连接：SQLite 在多线程下默认不支持并发写，测试时建议使用 PostgreSQL 或 MySQL，或者在 SQLite 中配置 check_same_thread=False（仅用于测试）。
隔离性：确保每次测试都是干净的数据库环境，或者在 fixture 中做好数据清理。
断言逻辑：注意账户类型的方向。资产类借方增加，负债/权益类贷方增加。上面的测试中 revenue 是收入类，贷方记增加，如果直接减余额，逻辑可能不符合会计恒等式展示习惯，但在底层存储上，只要借贷平衡即可。优化扩展
这个基础版本能跑，但离生产级还差得远。以下是几个进阶方向，也是面试加分项：引入消息队列：如果记账量巨大，同步写入数据库会成为瓶颈。可以将记账请求发送到 Kafka 或 RabbitMQ，由消费者异步处理入库。但要保证最终一致性，需要引入对账机制。
审计日志表：除了 journal_entries，还需要一个 audit_log 表，记录谁、在什么时间、修改了什么。这在小企业做账合规中非常重要。
使用 Decimal 替换 Float：
from decimal import Decimal
# 在 Model 中
amount = Column(Numeric(10, 2)) # 精确到小数点后两位在业务逻辑中，所有金额运算都用 Decimal，最后再转为字符串或浮点数用于展示。
API 接口幂等性：参考 RFC 7231 规范，为每个 API 请求生成一个唯一的 Idempotency-Key。如果客户端因网络超时重试，服务端通过 Key 判断是否已处理过，避免重复记账。小结
通过这个小企业做账系统的实战，我们不仅实现了一个核心的业务模块，更掌握了以下关键点：复式记账的代码实现：借贷平衡校验是基础，事务隔离是保障。
并发控制：with_for_update() 悲观锁在财务系统中不可或缺。
精度处理：严禁使用 float 处理金额，Decimal 是标配。
调试思路：当代码“跑不通”时，先查事务是否回滚，再查数据一致性，最后查精度问题。这些知识点，无论是做全栈开发，还是应对高频面试题中的系统设计部分，都是硬通货。别只盯着语法看，要把业务逻辑和底层机制结合起来思考。
你在项目里踩过这个坑吗？比如因为浮点数精度导致账务不平，或者因为并发问题导致余额错误？评论区聊聊，咱们一起避坑。