Python资产管理系统实战：从数据模型到状态机与定时任务

简介：一套基于Python实现的资产管理系统完整源码包，面向Python初中级开发者、企业IT运维人员及需要快速搭建资产管理原型的学生。系统围绕资产录入、分类、状态追踪、查询报告与权限管理等核心功能展开，覆盖硬件、软件、办公设备等多类资产的日常管理场景，方便用户掌握资产管理类Web应用的典型实现路径。

资源共459个文件、压缩包约4.35MB。文件构成以Python源码（78个py）和前端资源为主，含93个HTML页面、128个JavaScript脚本、57个CSS样式表，以及少量图片、字体、文档和SQLite数据库文件。整体结构兼顾后端业务逻辑、前端展示层与静态资源，目录划分清晰，适合按模块阅读和二次开发。

目前已有633人学习下载。基于源码可了解资产信息建模、分类组织、状态流转、统计查询及权限控制等模块的设计思路，也能学习到前后端数据交互与界面样式组织方式，适合用于课程设计、毕业设计或企业内训参考。

1. 资产管理系统不是增删改查，先把生命周期画出来

很多团队把资产管理系统做成一张 Excel 数据表，能增删改查就算上线；等设备多起来，问题全在状态对不上、分类随口写、借还没有记录。基于 Python 实现的这套资产管理系统，前端用 Bootstrap、jQuery UI 和 Select2 搭界面，后端则把重点压在数据模型、状态机、权限和定时任务上。它不是那种只能录入和搜索的 demo，而是从资产录入、分类、状态追踪到统计报告的完整闭环。适合一个人维护百来件资产的中小团队，也适合想用 Flask 或 Django 做实战项目的开发者。对于看过 Snipe-IT 这类开源资产系统的熟手，这套代码的价值在怎么用更轻的结构把同样的规则落地。

2. 资产数据模型与分类树：从 Python 字典到 SQLAlchemy 映射

开发时我没有直接建表，而是先把资产卡片需要的信息列成 Python 字典，确认每个字段的业务含义后再映射到 ORM。这样做的原因是：字段名改起来最贵的是数据库迁移，而字典阶段改成本几乎为零。Python 的字典在这里不是最终存储结构，而是用来对齐前后端字段的草稿。

2.1 先用 Python 字典把字段定下来

新建项目的第一步，先把一条资产记录写成可读的字典，让业务人员看一眼就能确认字段。

# assets_demo.py sample_assets = [ { "asset_no": "IT-2024-0001", "name": "ThinkPad X1 Carbon", "category": "笔记本", "model": "X1 Carbon G10", "supplier": "联强国际", "purchase_date": "2024-03-18", "warranty_months": 36, "department": "研发部", "status": "in_use", "owner": "zhang.san", } ]

这段字典把资产录入的核心字段列全了。asset_no 是业务里的唯一标识，打印标签、扫码核对都要靠它；purchase_date 用字符串只是临时做法，后面要转成 date；status 一开始就用英文枚举而不是中文文案，否则报表排序和多语言都麻烦。负责录入的人可以直接在资产录入表单里看到这些字段，后端接口也就不需要临时做字段映射。

2.2 用 SQLAlchemy 声明式模型固化资产表

字典能看清字段，但做不了唯一约束、外键和索引，所以要落到 SQLAlchemy 的声明式模型上。

# models.py from datetime import datetime from sqlalchemy import Column, Integer, String, Date, DateTime, ForeignKey from sqlalchemy.orm import declarative_base, relationship Base = declarative_base() class AssetStatus: NEW = "new" IN_USE = "in_use" MAINTENANCE = "maintenance" RETIRED = "retired" class Category(Base): __tablename__ = "asset_category" id = Column(Integer, primary_key=True) name = Column(String(64), unique=True, nullable=False) parent_id = Column(Integer, ForeignKey("asset_category.id"), nullable=True) children = relationship("Category", backref="parent", remote_side=[id]) class Asset(Base): __tablename__ = "asset" id = Column(Integer, primary_key=True) asset_no = Column(String(32), unique=True, index=True, nullable=False) name = Column(String(128), nullable=False) category_id = Column(Integer, ForeignKey("asset_category.id"), nullable=False) model = Column(String(128), default="") supplier = Column(String(128), default="") purchase_date = Column(Date, nullable=True) warranty_months = Column(Integer, default=0) department = Column(String(64), index=True, default="") owner = Column(String(64), default="") status = Column(String(16), default=AssetStatus.NEW, index=True) created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

asset_no 的唯一索引保证不会出现两条相同的资产编号；category_id 外键把分类单独拆表，避免字典里那种手写字符串造成的分类漂移；status 存英文字符串而不是数字，日志和接口可读性更好。created_at 用数据库默认时间，而不是让应用层每次传入。

2.2.1 字段选型细节

warranty_months 保留为 Integer，之后计算到期日要用 relativedelta 或 dateutil，而不是简单加天数。department 这种维度在小型系统里可以直接存字符串，如果后面要做组织树，再拆成 department 表。SQLite 开发时无需额外配置；切 MySQL 只要把连接串改成 mysql+pymysql://user:pass@host/dbname。先跑 SQLite 起步并不会限制后续切换。

新环境要确认 Python 3.8 以上版本，Windows 上常见问题是 PATH 里的 Python 版本不一致，导致 SQLAlchemy 装到了另一个解释器。装完依赖后先执行一遍模型建表代码，比直接跑前端页面更能提前暴露环境问题。

2.3 分类树：自关联查询与物化路径的取舍

分类是层级结构：信息中心 -> 研发组 -> 硬件设备。最容易的做法是给每条资产存一个 category_path 字符串，比如信息中心/研发组/硬件设备；查询方便，但改层级时所有子项要一起改。另一个做法是用 parent_id 自关联，查询时递归向下。

# category_service.py from sqlalchemy.orm import Session def get_children(session: Session, parent_id): return session.query(Category).filter(Category.parent_id == parent_id).all() def build_tree(session: Session, parent_id=None, depth=0): tree = [] for row in get_children(session, parent_id): tree.append({ "id": row.id, "name": row.name, "depth": depth, "children": build_tree(session, row.id, depth + 1), }) return tree

build_tree 用递归把整棵分类树拼出来，depth 参数记录层级，前端下拉框里可以按缩进展示。这个写法在分类数量少时非常直观，但如果分类超过一千层，递归深度和查询次数会成为问题。实际项目中更常见的是自关联加缓存，第一次查完后把树结构放进 Redis，分类变更时再清掉。

3. 资产状态机与审计日志：状态字段背后的规则

资产管理里最容易出 bug 的地方不是表单，而是状态。一张设备从新购到报废，中间可能经过在用、维护、借用、遗失；如果前端直接把状态字段放开给用户随意选，就会出现新购设备直接变成遗失、报废设备又改成在用的脏数据。状态机的作用就是把这套流转规则用代码显式表达出来。

3.1 状态字段为什么不能随便更新

状态不是一列字符串，而是一张有向图。先用 Python 字典定义节点和允许的边，比在视图函数里写一堆 if 更清晰。

# status_machine.py ALLOWED_TRANSITIONS = { "new": {"in_use", "retired"}, "in_use": {"maintenance", "retired", "lost"}, "maintenance": {"in_use", "retired"}, "retired": set(), "lost": set(), } def can_transit(old_status: str, new_status: str) -> bool: return new_status in ALLOWED_TRANSITIONS.get(old_status, set())

用集合描述路径，要加一条路径时只需改这里。注意 retired 和 lost 是终态，不能继续流转。这个规则必须写在后端，不能只写在下拉框的 onChange 事件里。前端校验只能改善体验，后端校验才是数据保证。

3.2 把状态机放进 Service 层

状态变更还要带上操作人和原因，所以我把 change_asset_status 放在 service 层，而不是让视图直接改 asset.status。

# asset_service.py from datetime import datetime from sqlalchemy.orm import Session # 实际项目中 AssetLog 在 models.py 定义，这里先引用 from models import Asset, AssetLog from status_machine import can_transit def change_asset_status(session: Session, asset: Asset, target_status: str, operator: str, reason: str = ""): if not can_transit(asset.status, target_status): raise ValueError(f"不允许的状态变更: {asset.status} -> {target_status}") old_status = asset.status asset.status = target_status session.add(AssetLog( asset_id=asset.id, from_status=old_status, to_status=target_status, operator=operator, reason=reason, created_at=datetime.utcnow(), )) session.commit()

先保存 old_status 再更新 asset.status，这样日志里的 from 和 to 才是真实路径。别小看这个顺序，问题排查时日志比代码时序更早暴露问题。operator 参数应该来自登录会话，而不是前端表单传入的字符串；否则任何人都可以伪造操作人。

3.3 审计日志：业务日志与 Python logging 双通道

业务日志要落库，运行日志要落文件。AssetLog 模型负责记录每一次状态变更，方便以后按资产编号回放生命周期。

# audit_models.py from datetime import datetime from sqlalchemy import Column, Integer, String, DateTime, ForeignKey class AssetLog(Base): __tablename__ = "asset_log" id = Column(Integer, primary_key=True) asset_id = Column(Integer, ForeignKey("asset.id"), index=True, nullable=False) from_status = Column(String(16)) to_status = Column(String(16)) operator = Column(String(64)) reason = Column(String(255), default="") created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

asset_id 建索引，因为查询设备历史时最常按这个维度过滤。reason 字段不是必填，但一旦填了，后续审计时能看到状态变更的真实原因。

3.3.1 logging 的 JSON 化输出

Python 的 logging 模块除了打印文本，也可以直接把每条日志格式化成 JSON 行，方便后续接入日志平台。

# logging_config.py import json import logging def setup_logging(): handler = logging.StreamHandler() handler.setFormatter(logging.Formatter( json.dumps({ "time": "%(asctime)s", "level": "%(levelname)s", "module": "%(module)s", "message": "%(message)s", }) )) root = logging.getLogger() root.handlers = [handler] root.setLevel(logging.INFO)

业务审计记录留在 asset_log 表里，查询设备历史状态时用 SQL 就能回放；运行日志输出到终端或日志文件，排查接口慢、任务失败时才用得到。两个通道互不替代。

4. 查询、报告、权限与前端资源：把 Bootstrap 表单接回 Flask

资产系统的前端页面通常是一张大表单加一张表格。Bootstrap 负责栅格和基础样式，jQuery UI 负责日期选择器，Select2 负责下拉搜索。要让这些组件真正跑起来，后端必须提供三个能力：动态条件查询、聚合统计、角色权限控制。

4.1 动态条件查询：参数没来就跳过条件

前端表单会提交多个筛选条件，但用户不一定填满每个字段。查询函数要做的是：条件存在才加入 filter，避免盲目构造 SQL。

# asset_query.py from datetime import datetime from sqlalchemy import or_ def query_assets(session, filters: dict): q = session.query(Asset) if filters.get("keyword"): like = f"%{filters['keyword']}%" q = q.filter(or_(Asset.name.like(like), Asset.asset_no.like(like))) if filters.get("status"): q = q.filter(Asset.status == filters["status"]) if filters.get("department"): q = q.filter(Asset.department == filters["department"]) if filters.get("category_id"): q = q.filter(Asset.category_id == int(filters["category_id"])) if filters.get("purchase_from"): q = q.filter(Asset.purchase_date >= datetime.strptime( filters["purchase_from"], "%Y-%m-%d").date()) return q.order_by(Asset.asset_no.asc()).all()

用 .get() 避免前端少传参数时抛 KeyError；purchase_from 从 query string 拿到的是字符串，必须先用 strptime 转成 date，否则 SQLAlchemy 比较 date 和 str 在 SQLite 下会静默出错。category_id 同理要 int()，这是表单提交里最常见的 Python 类型转换问题。

4.2 统计报告：group_by 之后还能直接导出

统计报表不需要把每一行资产传到前端再去数，直接在数据库端聚合。

# report_service.py import csv from io import StringIO from sqlalchemy import func def count_by_status(session): rows = session.query(Asset.status, func.count(Asset.id)).group_by(Asset.status).all() return {status: count for status, count in rows} def export_report_csv(session): data = count_by_status(session) buf = StringIO() writer = csv.DictWriter(buf, fieldnames=["status", "count"]) writer.writeheader() for status, count in data.items(): writer.writerow({"status": status, "count": count}) return buf.getvalue()

group_by 聚合在数据库端完成，比把所有行 load 进 Python 再统计省内存。如果还要看部门资产分布，再加一个 department 字段进 group_by。用 pandas 做数据分析与可视化时，可以直接把 rows 转成 DataFrame，再画堆叠图或导出 Excel。

4.3 权限控制：角色装饰器只放行该放行的人

资产数据属于内部信息，不能每个登录用户都能改。Flask-Login 管理会话，角色装饰器只做一件事：当前用户角色不在白名单里就返回 403。

# auth.py from functools import wraps from flask_login import current_user def role_required(*roles): def decorator(fn): @wraps(fn) def wrapper(*args, **kwargs): if not current_user.is_authenticated: return {"error": "未登录"}, 401 if current_user.role not in roles: return {"error": "无权限"}, 403 return fn(*args, **kwargs) return wrapper return decorator @app.route("/api/assets/<int:asset_id>", methods=["PUT"]) @role_required("admin", "asset_manager") def update_asset(asset_id): return {"ok": True}

常见误用是把权限判断写在视图函数中间，导致每个接口重复一套逻辑。角色名保持常量，不要散落在字符串字面量里。前端如果换成 Vue 或 React，把 current_user 换成 JWT payload 再判断 role 即可，装饰器结构不用变。

4.4 Bootstrap、jQuery UI 与 Select2 的加载顺序

这套系统的静态资源里有 bootstrap.min.css、jquery-ui.css、select2-bootstrap.css、font-awesome.css 和 style.css。很多人直接把文件全塞进 HTML，结果下拉框样式错乱、图标不显示，基本都是顺序问题。

<link rel="stylesheet" href="css/bootstrap.min.css"> <link rel="stylesheet" href="css/jquery-ui.min.css"> <link rel="stylesheet" href="css/select2-bootstrap.css"> <link rel="stylesheet" href="css/font-awesome.min.css"> <link rel="stylesheet" href="css/style.css">

bootstrap.min.css 在前，让后续组件都基于它覆盖；select2-bootstrap.css 必须放在 select2 的核心样式之后，否则主题类名不会生效；jquery-ui.css 和 jquery-ui.min.css 是重复文件，线上只保留 min 版本，避免加载两套样式。

4.4.1 Select2 的 AJAX 回填

资产录入表单里的分类下拉，通常用 Select2 的 AJAX 模式搜索后端接口。

// asset_form.js $('#category_id').select2({ placeholder: '选择资产分类', allowClear: true, ajax: { url: '/api/categories', dataType: 'json', delay: 250, data: function (params) { return { q: params.term }; }, processResults: function (data) { return { results: data.items.map(function (item) { return { id: item.id, text: item.name }; }) }; } } });

delay 250ms 是防抖，避免每敲一个字符就发请求；q 是 Select2 传给后端的搜索词；processResults 必须返回 { results: [...] } 结构，字段名错了下拉列表不会渲染。后端的 /api/categories 接口也要注意：q 参数长度超过 64 就截掉，防止用户输入超长关键词拖垮查询。

5. APScheduler 盘点提醒：用定时任务验证资产管理闭环

资产管理不能只靠人点按钮，系统应该每天自动去检查哪些资产保修快到期、哪些已经过了报废期。定时任务就是那种没人点击也被执行的验证逻辑，它逼迫数据模型和状态机必须真实可用。

5.1 用 BackgroundScheduler 做每日盘点任务

# scheduler.py from apscheduler.schedulers.background import BackgroundScheduler def start_scheduler(): scheduler = BackgroundScheduler(timezone="Asia/Shanghai") scheduler.add_job( check_warranty_soon, trigger="cron", hour=9, minute=0, id="warranty_check", replace_existing=True, misfire_grace_time=300, ) scheduler.start() return scheduler

cron trigger 表示每天 9 点运行一次，misfire_grace_time 300 秒的意思是任务如果因为系统休眠迟到不超过 5 分钟仍然执行，避免错过。replace_existing=True 可以防止 Flask debug 模式重复启动调度器。

5.2 用保修计算和通知任务验证状态闭环

# tasks.py from datetime import date, timedelta def check_warranty_soon(session, days=30): today = date.today() deadline = today + timedelta(days=days) rows = session.query(Asset).filter( Asset.status.in_(["in_use", "maintenance"]) ).all() for asset in rows: end_date = add_months(asset.purchase_date, asset.warranty_months) if end_date and end_date <= deadline: notify_manager(asset.asset_no, asset.name, end_date)

筛选条件限定 in_use 和 maintenance，是因为报废和遗失资产不需要提醒。add_months 要处理月份跨年的问题，常见做法是用 dateutil.relativedelta，或者先计算 months 总数再整除 12 和取模加 1。第一次跑定时任务前，先不要依赖调度器，直接在命令行里调用一次 check_warranty_soon(session, 365) 看看输出。确认会提醒哪些资产后，再把任务交给 APScheduler，这样至少能排除任务到底有没有执行、提醒逻辑对不对两件事。

定时任务跑起来之后，再去查 asset_log 表里有没有对应的状态变更记录。下一次有人问这批设备有没有到期未处理，答案不是靠口头汇报，而是从这张表里一条条回放出来的。

本文还有配套的精品资源，点击获取