800 行 Python 手写 CI/CD 平台：从 Zip 上传到 Docker 部署的完整实现

2301_76700054

501人浏览 · 2026-08-03 11:34:05

2301_76700054 · 2026-08-03 11:34:05 发布

摘要：Py_cicd 是一个约 800 行 Python 代码从零构建的轻量级 CI/CD Web 平台，不依赖 Jenkins 等第三方 CI 系统。用户通过网页上传项目 Zip 包并点击构建，平台自动完成依赖安装、自动化测试、Docker 镜像构建和容器化部署全流程，最终返回可访问的服务地址。文章从技术架构、数据模型、五步构建流水线、实时日志系统到开发中踩过的六个典型坑，系统梳理了完整实现思路，适合对 Python 后端、CI/CD 或 Docker 自动化感兴趣的开发者阅读。

做过中小型 Python 项目部署的人大概都有这种体验：开发机上的代码跑得好好的，一到服务器就要手动 pip install、手动改端口、手动配环境变量。如果再加上「每次发布都要重复这套流程」，运维就变成了纯粹的体力劳动。

CI/CD 这个词大家都熟，但落到实际往往绕不开 Jenkins 那一套——装插件、写 Groovy、配节点、管权限，学习曲线本身就超过了大部分小项目本身的价值。对于三五个人的团队，或者一个初版的 SaaS 原型，需要的其实不是 Jenkins 那种级别的「重型武器」，而是一个简单直给的工具：把代码丢进去，自动帮我把环境装好、测试跑完、服务启动，给我一个能直接访问的地址。

这就是 Py_cicd 解决的问题。

Py_cicd 是一个用 Python 从零构建的轻量级 CI/CD Web 平台，核心代码约 800 行，不依赖 Jenkins 等第三方 CI 系统。用户在网页上新建项目、上传 zip 包、点击构建，平台自动完成整个交付链路：

这篇文章会对项目的技术架构、核心实现、关键设计决策和开发中踩过的坑做一次系统性的梳理，既有完整的代码逻辑分析，也有可复用的工程实践，适合对 Python 后端、CI/CD 或 Docker 自动化感兴趣的朋友阅读。

技术架构

整体设计分为三层：浏览器交互层、Flask 服务层、Docker 引擎层。服务层内部通过 queue.Queue 实现日志的生产者-消费者解耦。

技术选型分析

层	选型	决策依据
Web 框架	Flask 3.x	轻量、路由简洁、不需要 Django 那种重量级 ORM 和中间件体系
模板引擎	Jinja2（Flask 内置）	配合 Bootstrap 5 做服务端渲染，零前端构建工具链
ORM	SQLAlchemy 2.x	用 `joinedload` 等特性解决懒加载问题，同时保留后期切数据库的灵活性
数据库	SQLite	单机工具不需要 MySQL，一个文件零配置；有 ORM 兜底，未来改一行连接串即可升级
容器引擎	Docker SDK for Python (docker-py)	Python 原生调用，流式读取构建日志，比 `subprocess` 调 CLI 更稳定
实时通信	SSE (Server-Sent Events)	日志是单向推送场景，SSE 比 WebSocket 更轻，浏览器 `EventSource` API 内置断线重连
并发模型	threading + queue.Queue	生产者-消费者解耦构建与推送，`daemon=True` 确保主进程退出时清理
生产部署	Gunicorn (2 worker × 4 thread)	Flask 自带开发服务器不适合生产，Gunicorn 提供并发能力和超时控制

数据模型设计

数据库只有两张表，用 SQLAlchemy 的 ORM 映射。设计原则是保持足够简单，但能完整追踪每个项目的构建历史。

# Project — 项目主表
class Project(Base):
    __tablename__ = "projects"
    id            = Column(String(12), primary_key=True)   # uuid4 前8位，如 "a2d0d0ba"
    name          = Column(String(200), nullable=False)     # 项目名称
    created_at    = Column(DateTime, default=datetime.utcnow)
    status        = Column(String(20), default="stopped")   # stopped / building / running / failed
    port          = Column(Integer, nullable=True)          # 分配的宿主机端口
    container_id  = Column(String(64), nullable=True)       # Docker 容器 ID
    image_tag     = Column(String(200), nullable=True)      # 镜像标签
    error_msg     = Column(Text, nullable=True)             # 失败原因

    builds = relationship("BuildRecord", back_populates="project",
                          order_by="BuildRecord.created_at.desc()",
                          cascade="all, delete-orphan")


# BuildRecord — 构建记录表
class BuildRecord(Base):
    __tablename__ = "builds"
    id            = Column(Integer, primary_key=True, autoincrement=True)
    project_id    = Column(String(12), ForeignKey("projects.id"), nullable=False)
    build_num     = Column(Integer, default=1)              # 第 N 次构建
    status        = Column(String(20), default="running")   # running / success / failed
    log           = Column(Text, default="")                # 完整构建日志
    created_at    = Column(DateTime, default=datetime.utcnow)
    finished_at   = Column(DateTime, nullable=True)

    project = relationship("Project", back_populates="builds")

几个值得注意的设计细节：

id 使用 uuid4().hex[:8] 而非自增整数，避免 URL 中暴露项目数量，同时防止并发创建时的主键冲突；
builds 关系配置了 cascade="all, delete-orphan"，删除项目时自动级联删除构建记录；
BuildRecord.project_id 使用外键关联，且 nullable=False——构建记录必须有归属项目，这与级联删除的配合在后文「踩坑记录」中会展开；
查询仪表盘列表时用了 joinedload(Project.builds) 预加载，避免模板渲染时触发懒加载导致 DetachedInstanceError。

构建流水线：五步实现「上传即部署」

构建引擎是项目的核心，位于 pycicd/builder.py。每次构建由 Flask 路由触发，在独立的后台守护线程中执行五步流水线。

Step 1: 文件检测

这一步负责在解压后的工作目录中定位项目的关键文件：

def _find_main(ws: str) -> str | None:
    # 先强制提平嵌套目录（双重保险）
    _ensure_flattened(ws)
    # 优先级：根目录 → 一级子目录
    for name in ["main.py", "app.py", "server.py", "wsgi.py", "run.py", "manage.py"]:
        if os.path.isfile(os.path.join(ws, name)):
            return name
    # 未匹配到已知命名，搜索包含框架特征的 py 文件
    for name in sorted(os.listdir(ws)):
        if name.endswith(".py") and not name.startswith("_"):
            content = open(os.path.join(ws, name)).read()
            if any(kw in content for kw in ("app.run", "Flask(", "FastAPI(", "gunicorn", "django")):
                return name
    # 继续搜索一级子目录（适用于项目源码放在子包中的情况）
    ...

入口文件的查找有明确的优先级：先文件名匹配，再内容特征匹配，再子目录搜索。这样做是为了兼容不同项目结构——有人的 app.py 在根目录，有人的主模块放在 src/ 下面。

另外，_ensure_flattened() 在构建阶段会再次执行目录提平。这个逻辑在上传阶段已经执行过一次，构建阶段再执行一次作为保险，处理上传时未触发提平的边缘情况（例如手动放置文件而非通过网页上传）。

Step 2: 依赖安装与测试

依赖安装使用 subprocess.Popen 而非 subprocess.run，目的是逐行捕获输出并实时推送到前端：

pip_cmd = [
    sys.executable, "-m", "pip", "install", "-r", req_file,
    "--target", os.path.join(ws, ".deps"),          # 安装到项目本地目录
    "-i", "https://pypi.tuna.tsinghua.edu.cn/simple",  # 清华镜像
    "--trusted-host", "pypi.tuna.tsinghua.edu.cn",
    "--progress-bar", "off",  # 关键：强制逐行输出
]
proc = subprocess.Popen(pip_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                        text=True, cwd=ws)
for line in proc.stdout:
    if line.strip():
        log(f"  {line.strip()}")   # 实时写入 Queue
proc.wait(timeout=600)

这里有三个关键决策：

--target .deps 将依赖装到项目工作目录而非系统路径，方便 Docker 构建时 COPY 使用，也避免污染全局 Python 环境；
清华镜像加速依赖下载，在国内网络环境下从分钟级降到秒级；
--progress-bar off 禁用 pip 的进度条动画——这是保证日志流不断的关键，\r 刷新不产生 \n，会导致 for line in proc.stdout 长时间读不到数据。

测试阶段的行为：

如果存在 tests/ 目录，自动执行 pytest -q --tb=short；
测试失败则标记构建失败并通过 return 终止流水线，不会进入 Docker 构建——坏代码不允许上线；
测试超时设为 120 秒，防止无限循环的测试用例卡死构建线程。

Step 3: Docker 镜像构建

这一阶段需要做一个判断：项目是否自带 Dockerfile？

没有的话，平台根据检测到的项目类型自动生成一个。生成逻辑考虑了 Django 的特殊性：

def _generate_dockerfile(ws, req_file, main_file):
    lines = ["FROM python:3.11-slim", "WORKDIR /app"]
    if req_file and os.path.isfile(req_file):
        lines.append("COPY requirements.txt .")
        lines.append("RUN pip install --no-cache-dir -r requirements.txt")
    lines.append("COPY . .")
    if main_file in ("manage.py",) or main_file.endswith("/manage.py"):
        # Django 项目需要指定 runserver 参数
        lines.append("EXPOSE 5000")
        lines.append('CMD ["python", "manage.py", "runserver", "0.0.0.0:5000"]')
    else:
        lines.append(f'CMD ["python", "{main_file}"]')
    ...

最早版本的 Dockerfile 对 Django 项目只写了 CMD ["python", "manage.py"]，不带任何参数，导致容器启动时只打印 Django 帮助信息就退出了。修正为在检测到 manage.py 时自动拼接 runserver 0.0.0.0:5000，所有容器统一监听 5000 端口，对外由宿主机做端口映射。

镜像构建通过 Docker SDK 的低层 API 调用 client.api.build(decode=True) 逐行获取构建输出，同样流式写入日志管道，保证前端能看到完整的构建过程。

Step 4: 停止旧容器

在启动新容器之前，需要先清理旧的。container_mgr.stop_project() 采用双重查找策略：

def stop_project(project: Project):
    client = docker.from_env()
    # 策略 1: 用 container_id 查找
    if project.container_id:
        try:
            c = client.containers.get(project.container_id)
            c.stop()
            c.remove()
        except (docker.errors.NotFound, docker.errors.APIError):
            pass
    # 策略 2: 用容器名称查找（兜底）
    try:
        c = client.containers.get(f"pycicd-{project.id}")
        c.stop()
        c.remove()
    except (docker.errors.NotFound, docker.errors.APIError):
        pass
    project.container_id = None
    project.status = "stopped"

双重查找的理由：数据库中的 container_id 可能因为容器被手动删除、Docker 重启等原因失效，但容器名称是平台统一命名的，用名称查找作为兜底能提高清理的成功率。

Step 5: 启动新容器与端口分配

端口分配逻辑扫描 5001–5099 范围，找到第一个未被占用的端口：

def _allocate_port(client) -> int:
    used = set()
    for c in client.containers.list(all=True):
        bindings = c.attrs.get("HostConfig", {}).get("PortBindings") or {}
        for bind_list in bindings.values():
            if bind_list:
                for b in bind_list:
                    try:
                        used.add(int(b["HostPort"]))
                    except (ValueError, KeyError):
                        pass
    for p in range(5001, 5100):
        if p not in used:
            return p
    return 5080  # 兜底端口

容器启动时附加了几个生产环境必要的配置：

client.containers.run(
    image=project.image_tag,
    name=f"pycicd-{project_id}",
    detach=True,
    restart_policy={"Name": "unless-stopped"},  # 容器异常退出自动重启
    ports={"5000/tcp": host_port},              # 容器:5000 → 宿主机:动态端口
    environment={"PYTHONUNBUFFERED": "1"},       # Python 输出不缓冲
    mem_limit="256m",                           # 内存限制，防止单个容器耗尽主机
)

实时日志系统：从构建线程到浏览器的完整链路

实时日志是用户感知最深的功能——构建过程如果不能实时看到，和「点了按钮干等」没有区别。整个链路涉及三条线程的协作。

日志双写机制

构建线程在写日志时做的是「双写」——同时写入 queue.Queue（给 SSE 推送）和一个 Python list（最终存数据库）：

def _log(log_queue, msg: str, log_lines: list | None = None):
    log_queue.put(msg)          # 路径 A: 推送给浏览器
    if log_lines is not None:
        log_lines.append(msg)   # 路径 B: 累积到列表，构建结束后写入 DB

这个设计解决了早期版本的一个重要 bug：构建结束时的 finally 块在关闭 session 之前把队列清空，导致 SSE 路由还没来得及读取就被消费掉了。现在两条路径互不干扰，日志存在哪里由各自路径决定。

SSE 长连接的保活机制

Flask 的 SSE 端点实现了一个带心跳的消费循环：

def generate():
    fail_count = 0
    while True:
        try:
            msg = q.get(timeout=15)       # 最多等 15 秒
            yield f"data: {json.dumps({'text': msg})}\n\n"
            fail_count = 0
            if msg == "__DONE__":
                break
        except Exception:
            fail_count += 1
            if fail_count >= 8:           # 8 × 15 = 120 秒无数据才真断开
                yield f"data: {json.dumps({'text': ''})}\n\n"
                break
            yield ":keepalive\n\n"        # 心跳注释行，SSE 协议不计入 data

timeout=15 和 fail_count >= 8 的组合意味着连接可以容忍 2 分钟的静默，期间每 15 秒发送一个 SSE 协议的注释行做心跳。这配合前端 EventSource 的内置重连，让日志流在 pip 下载大包等场景下也能保持连接。

前端的自动重连与状态同步

前端使用浏览器原生的 EventSource API，天然支持断线重连。SSE 流结束后，前端再通过一个额外的 REST 请求获取最终状态和完整日志——这是一个降级设计，确保即使 SSE 中途断开，用户也能看到完整的构建结果。

容器管理与生命周期

container_mgr.py 虽然代码量最小，但每个方法都处理了多个边界情况：

def remove_project(project: Project):
    stop_project(project)                        # 1. 停容器
    shutil.rmtree(os.path.join(..., project.id), # 2. 删工作目录
                  ignore_errors=True)
    client.images.remove(project.image_tag,      # 3. 删镜像
                         force=True)
    # 4. 手动删构建记录 + 项目（避免外键冲突）
    with Session() as s:
        p = s.query(Project).filter_by(id=project.id).first()
        if p:
            for b in p.builds:
                s.delete(b)
            s.delete(p)
            s.commit()

删除顺序是先 Docker 资源后数据库记录，先子表后主表。镜像删除用了 force=True，因为同一镜像可能被多个容器引用。工作目录删除用了 ignore_errors=True，因为可能存在权限问题，不应阻塞主流程。

安全与防护策略

项目不是企业级产品，但在几个关键点上做了防护：

Session 鉴权：login_required 装饰器拦截所有页面和 API 路由，密码通过环境变量 PLATFORM_PASSWORD 注入；
上传限制：Flask MAX_CONTENT_LENGTH = 500MB，配合前端提示排除 .git、node_modules 等大目录；
目录过滤：zip 解压时自动跳过 .git、__pycache__、node_modules、.venv、.idea、__MACOSX 和以 ._ 开头的 macOS 隐藏文件；
容器隔离：每个用户的每个项目跑在独立容器中，mem_limit=256m 限制内存使用；
端口隔离：每个项目独立端口，容器之间网络隔离。

开发中解决的几个典型问题

下面几个问题是开发过程中真实遇到的，不同程度上反映了 ORM、Docker SDK、子进程处理和前端通信的常见陷阱。

1. DetachedInstanceError——Session 生命周期与懒加载

仪表盘页面打开就报 500，Flask 捕获到 Parent instance is not bound to a Session。根因是：session.query(Project).all() 之后 session 自动关闭了，但 Jinja2 模板里访问 p.builds 时触发了 ORM 的懒加载，尝试回数据库查询却发现 session 已经关闭。

SQLAlchemy 默认关系是懒加载（lazy load），这在 Web 请求-响应模型里是常见的坑。修复方式是在查询时用 joinedload 预加载关联对象：

projects = s.query(Project).options(
    joinedload(Project.builds)
).order_by(Project.created_at.desc()).all()

SQLAlchemy 会在同一条 SQL 中 LEFT JOIN builds 表，一次性取回所有数据。session 关闭后模板直接使用内存中的对象，不会再发起查询。

2. SSE 日志频繁断开——pip 的 \r 进度条

表现是构建还在进行，前端日志流先提示断开。追下去发现链路有三层：

pip 的下载进度条使用 \r（回车）在不换行的情况下原地刷新百分比，不会产生 \n 换行符；
for line in proc.stdout 按换行符分割行，下载大包时可能几分钟都没有新行产出；
SSE 连接没有新数据超时被断开。

解决：--progress-bar off 强制 pip 逐行输出；SSE 增加心跳保活机制；日志双写解耦推送与存储。

3. Flask reloader 导致 ERR_CONNECTION_RESET

上传 zip 文件时浏览器直接报连接重置。原因是 Flask debug 模式的 watchdog（watchfiles）监控了整个项目目录，zip 文件的写入和删除触发了服务进程重启，当前请求被中断。

app.run(use_reloader=False) 关闭自动重载后解决。debug 模式本身的错误页面和 traceback 打印仍然保留。

4. Zip 嵌套目录——提平时机

GitHub 下载的 zip 解压后往往有一层 project-main/ 的包装目录，直接使用会导致找不到 requirements.txt。上传阶段会调用 _flatten_single_dir() 做提平，但这个函数在 zip 未删除时不会触发（因为 zip 文件本身被算作一个普通文件，条件 len(files) == 0 不成立）。

修复：先 os.remove(zip_path)，再调用 _flatten_single_dir(ws)。这两个操作的顺序至关重要。

5. Django 容器启动即退出

自动生成的 Dockerfile 里 CMD ["python", "manage.py"] 没带参数，Django 的 manage.py 不带参数执行只输出帮助信息然后退出，容器启动后立刻终止。修复是在检测到 manage.py 时生成 CMD ["python", "manage.py", "runserver", "0.0.0.0:5000"]。

6. 删除项目时的级联失败

删除 Project 时 SQLAlchemy 先尝试把 BuildRecord 的 project_id 更新为 NULL（SET NULL），但这个字段定义了 nullable=False，数据库拒绝执行。修复分两层：关系上配置 cascade="all, delete-orphan" 声明级联删除；删除代码里先手动 s.delete(b) 所有子记录，再删父记录，做双重保障。

项目目录结构

Py_cicd/
├── server.py                # Flask 主应用（路由、SSE、鉴权、文件上传）
├── pycicd/
│   ├── __init__.py
│   ├── models.py            # SQLAlchemy 数据模型与数据库初始化
│   ├── builder.py           # CI/CD 构建引擎（5 步流水线 + 日志双写）
│   └── container_mgr.py     # Docker 容器生命周期管理
├── templates/
│   ├── base.html            # Bootstrap 5 基础布局
│   ├── login.html           # 登录页
│   ├── dashboard.html       # 项目卡片列表（含状态、操作按钮）
│   ├── new_project.html     # 新建项目 + 文件上传
│   └── build_log.html       # 构建日志页（EventSource SSE）
├── sample/
│   ├── app.py               # 示例 Flask 应用
│   ├── requirements.txt     # flask
│   └── tests/
│       └── test_app.py      # pytest 用例
├── scripts/
│   └── rollback.sh          # 一键回滚脚本（Harbor + docker compose）
├── Dockerfile               # 自部署镜像（Gunicorn）
├── docker-compose.yml       # 一键自部署（挂载 docker.sock）
├── requirements.txt         # flask / sqlalchemy / docker / gunicorn
├── LICENSE                  # MIT
└── README.md

本地运行与部署

环境要求：

Python 3.11+
Docker Engine（Docker Desktop 或 Docker CE）
不需要安装数据库（SQLite 内置于 Python 标准库）

开发模式启动：

git clone https://github.com/okudhdheheuus/Py_cicd.git
cd Py_cicd
pip install -r requirements.txt
python server.py
# → http://localhost:5000，默认密码 admin

生产部署（推荐）：

docker compose up -d
# → http://localhost:5000
# 平台自身以 Gunicorn 运行在容器中，通过 docker.sock 挂载控制宿主 Docker

体验完整流程：

cd sample && zip -r ../demo.zip .
# 浏览器：新建项目 → 上传 demo.zip → 点击构建
# 构建完成 → 应用运行在 http://localhost:5001

一键回滚（配合 Harbor 镜像仓库）：

bash scripts/rollback.sh harbor.example.com py-cicd/flask-app v41 prod

技术价值与适用场景

对于正在学 Python 后端的同学，这个项目是一个不错的综合练手案例。代码量不大，但把 Web 开发、ORM 使用、Docker 操作、多线程和实时通信这几块串成了一条完整的链路。建议的学习路径是：先把项目跑起来，用 sample/ 下的示例体验一遍完整流程，然后从头读 server.py 的路由设计，再看 builder.py 的流水线逻辑，最后看 container_mgr.py 的资源管理。读完之后自己改一个版本，比如加一种新语言的构建支持，会比看教程有更深的印象。

对于实际团队使用，当前版本适合作为中小型 Python 项目的内部部署工具。几个自然的增强方向：