Python+Django运维平台实战：从脚本堆到自动化任务下发与权限审计

简介：这是一套面向Python Web开发初学者与运维工程师的Django运维平台实战源码，适合想通过完整项目理解前后端不分离架构、掌握自动化运维工具开发的学习者。资源包共164个文件，约3.63MB，以39个py文件承载Django模型、视图与业务逻辑，38个js与23个css构建前端交互与样式，另有11个html模板、18个png及多种字体图标文件，整体结构清晰，便于按模块阅读。平台功能覆盖主机管理、服务监控、日志查看、任务调度、权限控制、资源统计、脚本执行、故障告警、审计追踪与知识库等典型运维场景，读者可借此理解Django后台管理、数据模型定义与模板渲染的配合方式，并学习Linux运维与Web开发结合的实践思路。目前已有2714人学习下载，适合作为课程设计、毕业项目或运维工具二次开发的参考案例。

1. 从脚本堆到平台：Python-基于django的运维平台到底在解决什么

很多运维同行的日常是这样的：十几台机器靠 SSH 手工敲，几十台靠一堆散落的 shell 脚本，上百台就开始出现「脚本比机器还难管」的局面。Python-基于django的运维平台，本质上是把这堆散落的脚本、定时任务、资产台账和权限控制收进一个 Web 系统里，让「谁在什么时候对哪台机器做了什么」变成可查、可控、可复现的动作。它解决的不是「能不能连上机器」，而是「连上之后怎么不失控」。适合两类人：一是手里已经有几十台以上服务器、脚本开始互相打架的运维；二是想从写脚本进阶到做内部工具的后端开发者。这一章先把边界划清楚，后面才好动手。

Django 在这里的角色不是「顺便用一下」，而是承担了这套系统里最烦人的三件事：用户与权限、ORM 与数据一致性、后台管理。运维平台天然是多角色的——管理员能下发任务，普通运维只能看自己负责的机器，审计只能读日志。这些用 Django 自带的 auth 和 permission 体系能省掉大量重复代码。而资产、任务、执行记录这些结构化数据，用 ORM 建模比手写 SQL 拼接安全得多，尤其是涉及批量更新和状态流转时。

需要提前说清楚的是，这个方向不是让你去造一个商业级监控系统。监控有 Prometheus、Zabbix 这类成熟方案，运维平台的价值在于「操作编排」和「资产与权限的收口」。把边界想明白，选型就不会跑偏：Django 负责 Web 层和业务逻辑，真正连机器执行命令的部分交给独立的执行器，两者用消息或任务表解耦。这个分层是后面所有章节的地基。

2. 技术选型与最小可跑骨架：Django 怎么和运维场景对上

2.1 为什么是 Django 而不是 Flask 或 FastAPI

选型这件事，我一般看三个维度：内置能力、团队熟悉度、后期维护成本。Flask 轻，但用户体系、后台、ORM 都要自己拼；FastAPI 异步性能好，适合纯 API 服务，但运维平台大量是「表单 + 列表 + 权限」的 CRUD 场景，异步优势用不上。Django 自带 admin、auth、ORM、表单校验和迁移工具，一个中等规模的运维平台，光 admin 就能顶掉一半的后台开发量。

具体到运维场景，Django 有几个点特别对味。第一是 ORM 的事务支持，批量下发任务时，任务记录和状态更新必须在一个事务里，否则会出现「任务发了但记录没写」的脏数据。第二是中间件机制，做操作审计时，一个中间件就能把所有写操作的请求体、用户、时间落库。第三是 signals，资产变更时自动触发同步逻辑，不用在每个视图里手动调用。

代价也要讲清楚：Django 的同步模型在高并发下发场景下会吃力，所以执行层必须拆出去。常见做法是 Django 只负责「创建任务记录」，真正的执行交给独立的 worker 进程轮询任务表或消费消息队列。这个拆分不是过度设计，是这套架构能不能扛住批量操作的关键。

2.2 用 venv 和 requirements 固定住运行环境

第一步永远是环境隔离。运维平台依赖多，版本冲突是血泪经验里最常见的一类翻车。用 venv 建虚拟环境，把依赖写进 requirements.txt，换台机器也能一键还原。

# 创建并激活虚拟环境，Python 3.10 及以上 python3 -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装核心依赖，版本按实际锁定 pip install django==4.2.* psycopg2-binary celery redis paramiko # 冻结当前环境，方便复现 pip freeze > requirements.txt

这里几个依赖各有分工：django 是框架本体；psycopg2-binary 是 PostgreSQL 驱动，生产环境别用 SQLite；celery 加 redis 负责异步任务，也就是前面说的执行层解耦；paramiko 是 SSH 执行命令的库。版本号我习惯用4.2.*这种写法，锁大版本不锁小版本，既能拿到安全补丁，又不会因为小版本跳变踩坑。

提示：requirements.txt 里不要出现-e本地路径依赖，部署到别的机器会直接失败。所有依赖都从包索引装。

2.3 建项目、建应用、连数据库的最小命令

环境好了就开始搭骨架。Django 的项目结构是「项目 + 多个应用」，运维平台一般拆成资产、任务、审计三个应用，职责清晰，后期好维护。

# 创建项目，注意结尾的点，避免多套一层目录 django-admin startproject opsplatform . # 创建三个核心应用 python manage.py startapp assets # 资产管理 python manage.py startapp tasks # 任务编排 python manage.py startapp audit # 操作审计

建完之后要在 settings.py 里注册应用并配置数据库。数据库这块我强烈建议直接上 PostgreSQL，原因很实际：运维平台会有大量并发写任务状态，SQLite 的写锁在批量场景下会直接卡死。配置时把数据库连接信息走环境变量，别硬编码在代码里。

# settings.py 关键片段 import os INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'assets', 'tasks', 'audit', ] DATABASES = { 'default': { 'ENGINE': 'django.db.backends.postgresql', 'NAME': os.environ.get('DB_NAME', 'opsplatform'), 'USER': os.environ.get('DB_USER', 'ops'), 'PASSWORD': os.environ.get('DB_PASSWORD', ''), 'HOST': os.environ.get('DB_HOST', '127.0.0.1'), 'PORT': os.environ.get('DB_PORT', '5432'), 'CONN_MAX_AGE': 60, # 复用连接，减少握手开销 } }

CONN_MAX_AGE这个参数值得单独说。默认是 0，每个请求结束就关连接，批量操作时连接建立的开销很可观。设成 60 秒复用，能明显降低数据库压力。但要注意，如果前面挂了连接池中间件，这个值要相应调整，否则会出现连接被两边同时管理的情况。

2.4 资产模型：把「机器」抽象成可查询的数据

运维平台的核心数据是资产。资产模型设计得好不好，直接决定后面任务下发和权限过滤顺不顺手。我一般会拆成「主机」和「业务分组」两张表，用外键关联，而不是把所有信息塞一张表。

# assets/models.py from django.db import models class BusinessGroup(models.Model): """业务分组，用于权限隔离和批量操作""" name = models.CharField(max_length=64, unique=True) owner = models.CharField(max_length=32, blank=True) def __str__(self): return self.name class Host(models.Model): """主机资产""" ip = models.GenericIPAddressField(unique=True) hostname = models.CharField(max_length=128) group = models.ForeignKey( BusinessGroup, on_delete=models.PROTECT, related_name='hosts' ) ssh_port = models.PositiveIntegerField(default=22) os_type = models.CharField(max_length=32, default='linux') is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: indexes = [models.Index(fields=['group', 'is_active'])]

几个设计点解释一下。on_delete=models.PROTECT而不是 CASCADE，是因为删分组时如果还有主机挂着，应该报错而不是把主机一起删掉，这是运维数据的基本安全底线。related_name='hosts'让反向查询写成group.hosts.all()，可读性好。Meta 里的联合索引针对「按分组查活跃主机」这个高频查询，不加索引的话主机上千台后列表页会明显变慢。

is_active字段是软删除的替代方案。运维场景里机器下线很常见，但历史任务记录还要能关联到这台机器，所以不能真删，用状态位标记。这个习惯能省掉很多「记录关联丢失」的排查时间。

3. 任务下发与执行解耦：从 Web 请求到真正跑在机器上

3.1 任务模型与状态机设计

任务下发最容易翻车的地方，是把「执行」直接写在视图里。用户点一下按钮，视图里同步 SSH 连过去跑命令，机器一多请求就超时，页面转圈转到用户以为系统挂了。正确做法是视图只创建任务记录，执行交给后台 worker。这就需要一个带状态机的任务模型。

# tasks/models.py from django.db import models from assets.models import Host class Task(models.Model): STATUS_CHOICES = [ ('pending', '待执行'), ('running', '执行中'), ('success', '成功'), ('failed', '失败'), ] name = models.CharField(max_length=128) command = models.TextField() hosts = models.ManyToManyField(Host, related_name='tasks') status = models.CharField( max_length=16, choices=STATUS_CHOICES, default='pending' ) created_by = models.CharField(max_length=32) created_at = models.DateTimeField(auto_now_add=True) finished_at = models.DateTimeField(null=True, blank=True) class TaskResult(models.Model): """每台主机的执行结果，和 Task 一对多""" task = models.ForeignKey(Task, on_delete=models.CASCADE, related_name='results') host = models.ForeignKey(Host, on_delete=models.PROTECT) exit_code = models.IntegerField(null=True) stdout = models.TextField(blank=True) stderr = models.TextField(blank=True) duration = models.FloatField(null=True) # 秒

状态机只有四个状态，但流转规则要卡死：pending 只能到 running，running 只能到 success 或 failed，不允许从 success 回退。这个约束建议在模型层用方法封装，而不是散在各个视图里判断。

# tasks/models.py 续 def mark_running(self): if self.status != 'pending': raise ValueError(f'非法状态流转: {self.status} -> running') self.status = 'running' self.save(update_fields=['status'])

update_fields这个参数是性能细节。只更新状态字段，不触发整行写，在高频状态变更时能减少锁竞争。很多人忽略它，任务一多数据库就开始抖。

3.2 用 Celery 把执行层拆出去

执行层用 Celery 是常见做法，原因是它和 Django 集成成熟，任务重试、超时、并发控制都有现成机制。配置分三块：broker、backend 和任务路由。

# settings.py 追加 CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0' CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1' CELERY_TASK_TIME_LIMIT = 300 # 单任务硬超时 5 分钟 CELERY_TASK_SOFT_TIME_LIMIT = 240 # 软超时，留出清理时间 CELERY_WORKER_CONCURRENCY = 8 # 并发 worker 数

TIME_LIMIT和SOFT_TIME_LIMIT的区别要搞清楚：软超时先触发，任务可以捕获异常做清理；硬超时直接杀进程。SSH 执行命令最怕卡死，这两个值必须设，否则一个卡住的连接会占着 worker 不放。

执行任务本身用 paramiko 连过去跑命令，关键是超时和输出截断。

# tasks/tasks.py import paramiko from celery import shared_task from .models import Task, TaskResult @shared_task(bind=True, max_retries=2) def run_task(self, task_id): task = Task.objects.get(id=task_id) task.mark_running() for host in task.hosts.filter(is_active=True): result = TaskResult.objects.create(task=task, host=host) try: client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( host.ip, port=host.ssh_port, username='ops', timeout=10, banner_timeout=10 ) _, stdout, stderr = client.exec_command(task.command, timeout=60) result.exit_code = stdout.channel.recv_exit_status() # 输出截断，防止单条记录撑爆数据库 result.stdout = stdout.read().decode(errors='ignore')[:10000] result.stderr = stderr.read().decode(errors='ignore')[:10000] client.close() except Exception as exc: result.stderr = str(exc)[:10000] result.exit_code = -1 finally: result.save() task.status = 'success' if not task.results.filter(exit_code__lt=0).exists() else 'failed' task.save(update_fields=['status'])

逻辑说明：bind=True让任务能访问 self，配合 max_retries 做重试。timeout=10是连接超时，exec_command的timeout=60是命令执行超时，两个都要设。输出截断到 10000 字符是必须的，我见过cat一个大日志文件直接把数据库写爆的情况。recv_exit_status()必须在读 stdout 之前调用，否则可能拿到空值，这是个经典坑。

3.3 视图层只做三件事：校验、建记录、投递

视图层要克制，别把业务逻辑堆进来。一个下发任务的视图，职责就是校验参数、创建 Task 和关联主机、把任务 ID 投给 Celery。

# tasks/views.py from django.http import JsonResponse from django.views.decorators.http import require_POST from django.db import transaction from .models import Task from .tasks import run_task from assets.models import Host @require_POST @transaction.atomic def create_task(request): command = request.POST.get('command', '').strip() host_ids = request.POST.getlist('host_ids') if not command or not host_ids: return JsonResponse({'error': '命令和主机不能为空'}, status=400) task = Task.objects.create( name=request.POST.get('name', '未命名任务'), command=command, created_by=request.user.username, ) task.hosts.set(Host.objects.filter(id__in=host_ids, is_active=True)) # 事务提交后再投递，避免 worker 读到未提交数据 transaction.on_commit(lambda: run_task.delay(task.id)) return JsonResponse({'task_id': task.id, 'status': task.status})

transaction.on_commit是这里的关键。如果直接在事务里run_task.delay，worker 可能在事务提交前就去查 Task，查不到直接报错。这个坑很隐蔽，本地单机测试经常发现不了，一上生产就偶发失败。@transaction.atomic保证 Task 和主机关联要么都成功要么都回滚。

4. 权限、审计与批量操作的避坑清单

4.1 权限过滤必须落在查询集上

运维平台的权限不是「能不能进这个页面」，而是「能看到哪些机器」。这个过滤必须落在 queryset 层，不能只在模板里隐藏按钮。模板隐藏是障眼法，接口一调照样能操作别人的机器。

# assets/views.py from django.contrib.auth.decorators import login_required from django.shortcuts import render from .models import Host @login_required def host_list(request): # 管理员看全部，普通用户只看自己分组 if request.user.is_superuser: qs = Host.objects.filter(is_active=True) else: groups = request.user.groups.values_list('name', flat=True) qs = Host.objects.filter(is_active=True, group__name__in=groups) return render(request, 'assets/host_list.html', {'hosts': qs})

用 Django 自带的 group 和业务分组做映射，是最省事的方案。关键是每个涉及资产的视图都要走这个过滤逻辑，建议抽成一个get_visible_hosts(user)函数，别在每个视图里重复写，重复就意味着迟早有一处漏掉。

4.2 审计中间件：把写操作全记下来

审计是运维平台的后悔药。出问题时能查到「谁在几点对哪台机器执行了什么命令」，比事后扯皮有用得多。用中间件统一记录，比在每个视图里手动写日志可靠。

# audit/middleware.py import json from .models import OperationLog class AuditMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): response = self.get_response(request) # 只记录写操作，读操作量太大 if request.method in ('POST', 'PUT', 'DELETE') and request.user.is_authenticated: OperationLog.objects.create( user=request.user.username, path=request.path, method=request.method, body=json.dumps(request.POST.dict(), ensure_ascii=False)[:2000], status_code=response.status_code, ) return response

request.POST.dict()会把多值字段压成一个值，如果表单里有 checkbox 多选，记录会丢数据。更严谨的做法是用request.POST.lists()保留多值。另外 body 截断到 2000 字符，防止有人提交大文本把日志表撑大。审计日志表增长很快，建议按月分表或定期归档，别让它无限膨胀拖慢查询。

4.3 批量下发的三个必调参数

批量操作是运维平台最容易出事故的地方。一次对几百台机器执行命令，参数没调好就是灾难。下面三个参数我每次上线前都会确认。

并发数不是越高越好。目标机器如果是同一批，并发太高相当于对它们发起一次小型压测，轻则负载飙升，重则服务不可用。稳妥做法是分批，比如每批 20 台，批间加个短延迟。这个逻辑可以在 Celery 任务里用chunks或者自己控制节奏。

4.4 避坑清单：五个真实踩过的坑

现象：任务状态一直停在 running，永远不结束。原因：SSH 连接建立后目标机器网络抖动，exec_command卡住，而没设命令超时。解决：exec_command必须带timeout参数，同时 Celery 配SOFT_TIME_LIMIT兜底，双保险。

现象：本地测试一切正常，部署后任务偶发查不到。原因：视图在事务里直接投递 Celery 任务，worker 在事务提交前查询数据库。解决：用transaction.on_commit延迟投递，确保数据已落库。

现象：审计日志表几个月后查询变得极慢。原因：日志表没有索引也没有归档，几百万行全表扫描。解决：给user和created_at建索引，同时写个定时任务把三个月前的日志归档到冷表。

现象：批量任务里部分主机成功部分失败，但整体状态显示成功。原因：状态判断逻辑只看了 Task 本身，没聚合 TaskResult。解决：任务结束时统计exit_code < 0的结果数，有失败就标记整体 failed，别让用户误以为全成功。

现象：命令输出里有中文，存进数据库变成乱码。原因：decode()没指定编码，或者目标机器 locale 不是 UTF-8。解决：decode(errors='ignore')先保证不抛异常，同时在 SSH 执行前export LANG=en_US.UTF-8，统一编码环境。

5. 让平台真正可用：从能跑到敢用的几个进阶技巧

平台能跑起来只是及格线，敢不敢交给团队日常用，取决于几个细节。第一个是幂等性。同一个任务被重复投递（网络重试、用户连点）时，不能对机器重复执行危险命令。我的做法是在 Task 上加一个idempotency_key字段，视图层用get_or_create保证同一个 key 只创建一个任务。这个 key 可以由「用户 + 命令哈希 + 时间窗口」生成，简单有效。

第二个是灰度能力。批量命令下发前，先在一台机器上试跑，确认输出符合预期再全量。实现上不难，在 Task 里加一个dry_run标志，worker 遇到这个标志就只连第一台机器执行，把结果返回给用户确认。这个功能上线后，误操作导致的事故能少一大半，是性价比最高的一个改动。

第三个是执行结果的检索。任务多了以后，用户最常问的是「上周那条命令在哪台机器上跑失败了」。这要求 TaskResult 的 stdout 和 stderr 可搜索。PostgreSQL 的全文检索或者简单的icontains都能用，但要注意给字段加 GIN 索引，否则数据量上来后搜索会拖垮数据库。我一般会单独建一张搜索结果表，用异步任务把关键输出提取出来，避免在主表上做重查询。

第四个是连接复用。每台机器每次执行都新建 SSH 连接，开销不小。可以在 worker 进程内维护一个连接池，按主机 IP 缓存 SSHClient，设置空闲超时自动关闭。但这里有个坑：连接长时间不用会被目标机器断开，复用时必须先做一次心跳检测，否则会拿到失效连接直接报错。稳妥做法是复用前发一个echo探测，失败就重建。

最后说一个验证方法。平台做完别急着全量推，先拿一个测试分组跑两周，重点看三件事：任务成功率、平均执行耗时、审计日志完整性。成功率低于 95% 说明连接或超时参数有问题；耗时突然变长说明有机器或网络异常；审计日志有缺口说明中间件有漏记的路径。这三项稳定了，再逐步放开到生产分组。

我自己踩过最深的一个坑，是早期图省事把执行逻辑写在视图里，结果一次对两百台机器下发命令，请求超时后用户以为没发出去，又点了一次，命令被执行了两遍。从那以后我定了个习惯：任何涉及远程执行的操作，视图层只允许创建记录，执行一律走异步，并且必须带幂等键。这个习惯救过我很多次。希望帮到你。

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