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.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 Base
class 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 Base
class 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 logging
logger = logging.getLogger(__name__)
class BookkeeperService:
def __init__(self, db: Session):
self.db = db
def 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.0
total_credit = 0.0
for 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 entry
except Exception as e:
# 4. 异常回滚
self.db.rollback()
logger.error(fError creating journal entry: {str(e)})
raise
def get_balance(self, account_id: int) - float:
account = self.db.query(Account).filter_by(id=account_id).first()
if not account:
return 0.0
return 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 Base
engine = 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，贷收入 100
lines = [
{'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 None
assert client.get_balance(cash_id) == 100
assert 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.id
lines = [
{'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 是标配。
调试思路：当代码“跑不通”时，先查事务是否回滚，再查数据一致性，最后查精度问题。
这些知识点，无论是做全栈开发，还是应对高频面试题中的系统设计部分，都是硬通货。别只盯着语法看，要把业务逻辑和底层机制结合起来思考。
你在项目里踩过这个坑吗？比如因为浮点数精度导致账务不平，或者因为并发问题导致余额错误？评论区聊聊，咱们一起避坑。 
长河君 · 2026/9/22 18:18:22 · 阅读 1.2k 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.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 Base
class 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 Base
class 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 logging
logger = logging.getLogger(__name__)
class BookkeeperService:
def __init__(self, db: Session):
self.db = db
def 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.0
total_credit = 0.0
for 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 entry
except Exception as e:
# 4. 异常回滚
self.db.rollback()
logger.error(fError creating journal entry: {str(e)})
raise
def get_balance(self, account_id: int) - float:
account = self.db.query(Account).filter_by(id=account_id).first()
if not account:
return 0.0
return 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 Base
engine = 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，贷收入 100
lines = [
{'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 None
assert client.get_balance(cash_id) == 100
assert 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.id
lines = [
{'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 是标配。
调试思路：当代码“跑不通”时，先查事务是否回滚，再查数据一致性，最后查精度问题。
这些知识点，无论是做全栈开发，还是应对高频面试题中的系统设计部分，都是硬通货。别只盯着语法看，要把业务逻辑和底层机制结合起来思考。
你在项目里踩过这个坑吗？比如因为浮点数精度导致账务不平，或者因为并发问题导致余额错误？评论区聊聊，咱们一起避坑。