3天搞定小企业做账系统，搞定高频面试题

你刚把网上抄的记账代码跑起来，结果报错“字段缺失”，改了一晚上没思路，这简直是复制来的代码跑不通不知道怎么调的典型。很多后端开发想转全栈或做独立开发，总卡在业务逻辑和底层实现的衔接上，这不仅是实战难点，也是各大厂高频面试题里关于“系统设计与数据一致性”的考察重点。

别慌，今天咱们不整虚的，直接从一个小企业做账的真实场景出发，用 Python 从零手写一个轻量级记账系统。这不仅能解决你“调不通”的困惑，还能让你彻底搞懂复式记账法在代码里怎么落地，顺便把面试里常问的“如何处理并发下的账务一致性”给盘明白。

项目目标

咱们做的不是那种几百万行代码的 SAP 或 Oracle，而是一个面向小微企业的小企业做账核心模块。目标很明确：

1. 支持基础复式记账：有借必有贷，借贷必相等。这是会计铁律，也是代码逻辑的核心约束。
2. 数据强一致性：在多线程或高并发场景下，确保每一笔交易原子性提交，不能出现“借了钱但贷没记上”的脏数据。
3. 审计追踪：每一笔账都要有迹可循，符合财务合规要求。这里我们参考 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.md

• models：定义数据结构，对应数据库表。
• 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(f"Account 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(f"Error 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

逐行解析与避坑：

1. with_for_update()：这是解决并发问题的关键。在查询账户时加上行锁，防止两个线程同时读取同一个账户余额并分别更新，导致数据不一致。这是高频面试题中“乐观锁 vs 悲观锁”的实战应用。
2. 事务的 Commit 与 Rollback：务必在一个 try-except 块中处理。如果中间任何一步失败（比如某个账户不存在），必须回滚所有已修改的数据，保证原子性。
3. 浮点数陷阱：代码中用了 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)

调试技巧： 如果测试跑不通，检查以下几点：

1. 数据库连接：SQLite 在多线程下默认不支持并发写，测试时建议使用 PostgreSQL 或 MySQL，或者在 SQLite 中配置 check_same_thread=False（仅用于测试）。
2. 隔离性：确保每次测试都是干净的数据库环境，或者在 fixture 中做好数据清理。
3. 断言逻辑：注意账户类型的方向。资产类借方增加，负债/权益类贷方增加。上面的测试中 revenue 是收入类，贷方记增加，如果直接减余额，逻辑可能不符合会计恒等式展示习惯，但在底层存储上，只要借贷平衡即可。

优化扩展

这个基础版本能跑，但离生产级还差得远。以下是几个进阶方向，也是面试加分项：

1. 引入消息队列：如果记账量巨大，同步写入数据库会成为瓶颈。可以将记账请求发送到 Kafka 或 RabbitMQ，由消费者异步处理入库。但要保证最终一致性，需要引入对账机制。
2. 审计日志表：除了 journal_entries，还需要一个 audit_log 表，记录谁、在什么时间、修改了什么。这在小企业做账合规中非常重要。
3. 使用 Decimal 替换 Float：

   from decimal import Decimal
   # 在 Model 中
   amount = Column(Numeric(10, 2)) # 精确到小数点后两位
   

   在业务逻辑中，所有金额运算都用 Decimal，最后再转为字符串或浮点数用于展示。
4. API 接口幂等性：参考 RFC 7231 规范，为每个 API 请求生成一个唯一的 Idempotency-Key。如果客户端因网络超时重试，服务端通过 Key 判断是否已处理过，避免重复记账。

小结

通过这个小企业做账系统的实战，我们不仅实现了一个核心的业务模块，更掌握了以下关键点：

• 复式记账的代码实现：借贷平衡校验是基础，事务隔离是保障。
• 并发控制：with_for_update() 悲观锁在财务系统中不可或缺。
• 精度处理：严禁使用 float 处理金额，Decimal 是标配。
• 调试思路：当代码“跑不通”时，先查事务是否回滚，再查数据一致性，最后查精度问题。

这些知识点，无论是做全栈开发，还是应对高频面试题中的系统设计部分，都是硬通货。别只盯着语法看，要把业务逻辑和底层机制结合起来思考。

你在项目里踩过这个坑吗？比如因为浮点数精度导致账务不平，或者因为并发问题导致余额错误？评论区聊聊，咱们一起避坑。