Python

现代加密替代方案：AES‑GCM 与 ChaCha20‑Poly1305 实战指南（附 Python 示例） 副标题 / 摘要 这篇延伸读聚焦现代 AEAD 算法，解释为什么 AES‑GCM 与 ChaCha20‑Poly1305 是 RC4 的安全替代，并提供可运行的 Python 示例、常见陷阱与最佳实践。 建议先阅读配套文章《用 Python 还原 RC4 + JWT + 自定义 SSO Token 加解密》，理解遗留方案，再迁移到本篇的现代实践。 目标读者 后端/安全工程师（中级以上） 需要在服务间或 Web 客户端安全传输数据的工程团队 计划从自研/过时算法迁移到现代 AEAD 的项目负责人 背景 / 动机 RC4 等过时算法存在结构性弱点，且难以正确、安全地使用。现代 AEAD（Authenticated Encryption with Associated Data）算法在保证“机密性”的同时还能“认证完整性”，有效防止篡改与重放，API 更易用，错误空间更小——因此成为主流推荐。 核心概念 AEAD：同时提供加密（Confidentiality）与认证（Integrity/Authenticity）的模式。 Nonce/IV（随机数）：每次加密必须唯一（对同一密钥）。常用长度：12 字节。 AAD（Associated Data）：不加密但要认证的额外上下文（例如请求头、资源标识）。 Tag（认证标签）：解密时必须验证；任何修改都会导致校验失败。 Key Derivation（密钥派生）：通过 HKDF/Argon2/Scrypt 将口令或主密钥派生为会话密钥，避免直接使用弱口令。 实践指南 / 步骤 安装依赖 pip install cryptography 生成或派生密钥 服务到服务：使用随机 16/32 字节密钥（AES‑128/256），KMS 管理与轮换。 口令到密钥：使用 HKDF（或 Argon2/Scrypt）派生固定长度密钥，避免直接使用口令。 选择算法 AES‑GCM：硬件加速广泛（x86 AES‑NI），在服务端通用、高性能。 ChaCha20‑Poly1305：对移动/无 AES 加速的设备更友好，性能稳定。 Nonce 策略 每条消息使用唯一 Nonce（12 字节），推荐 os.urandom(12)，将 Nonce 与密文一起存储/传输（前缀写入）。 AAD 的使用 将上下文信息（版本、用户ID、消息类型等）作为 AAD 提供，增强完整性绑定。 密钥轮换 引入 kid（Key ID），支持多活密钥与平滑迁移。 可运行示例 以下示例仅演示用法。请结合 KMS、密钥轮换、权限隔离与 TLS，构建完整的生产级方案。 ...

用 Python 还原 RC4 + JWT + 自定义 SSO Token 加解密（含可运行示例） 副标题 / 摘要 这篇文章带你从 0 拆解 RC4 流加密、Base64/Hex 编码，以及基于 JWT 与自定义 SSO 的鉴权设计，并给出可以复制运行的 Python 示例。示例中的密钥与发行方均为占位值，切勿用于生产。 目标读者 Python 后端/测试工程师（中级） 对鉴权、令牌与基础加密流程感兴趣的开发者 想理解 RC4 工作方式与替代方案的安全入门读者 背景 / 动机 在实际项目中，我们常需要为 HTTP 或 WebSocket 请求附带令牌进行身份校验。常见做法包括使用 JWT（对称/非对称签名）或自定义的 SSO Token（例如对某段明文进行对称加密后再以 Hex/Base64 编码）。本文整理并复现一种组合方案：RC4+Base64/Hex 与 JWT/SSO 的加解密与校验流程，帮助你在测试或 PoC 中快速上手，同时理解其安全取舍。 核心概念 RC4：经典流加密（已不再安全）。通过密钥调度（KSA）与伪随机序列（PRGA）生成密钥流，与明文字节按位异或得到密文。解密过程与加密一致（同一函数）。 Base64 与 Hex：两种将二进制数据编码为可传输文本的方式。Base64 更紧凑；Hex 可读、调试直观。 JWT：JSON Web Token。Header.Payload.Signature。常包含 iss（发行方）、aud（受众/自定义）、iat/exp（签发/过期）。 自定义 SSO Token：一种自定义明文格式（示例采用 issuer_expire_ts_userSeqId_userId），经对称加密后再编码为 Hex，便于在 HTTP 头中传输。 限制与风险：RC4 已过时且不建议用于生产；如需兼容遗留系统，应仅在测试/过渡场景，且搭配 TLS、短周期、签名与回放防护。 实践指南 / 步骤 安装依赖 pip install pyjwt 设定占位常量（不要使用真实密钥/发行方） ISSUER = "demo-issuer" SECRET = "demo-secret-change-me" RC4KEY = "demo-rc4-key-change-me" UTE_ISSUER = "ute-demo" 实现 RC4 与常用编码包装 encrypt_string/decrypt_string：RC4 后 Base64 encrypt_hex_string/decrypt_hex_string：RC4 后 Hex 构造与校验 JWT（x-auth-token） aud = [Base64(RC4(user_id)), Base64(RC4(user_seq_id))] iss/iat/exp 等标准字段 构造与校验 SSO Token（x-sso-token） 明文 UTE_ISSUER_expire_ts_userSeqId_userId → RC4 → Hex 校验发行方与过期时间 运行演示，观察生成与校验结果 可运行示例（完整代码） 仅用于学习与测试，切勿将 RC4 用于生产环境。请优先使用现代 AEAD（AES‑GCM/ChaCha20‑Poly1305）。 ...

标题 用一段优雅的 Python 代码，把 SQLAlchemy 模型安全、高效地序列化成字典 副标题 / 摘要 SQLAlchemy 模型转字典（dict）看似简单，却暗藏字段格式、关系递归、循环引用等坑。本文通过一段实战代码，带你实现一个可复用的 _to_dict 序列化工具，并分析其设计取舍与改进方向，适合正在用 SQLAlchemy 写后端接口的你。 目标读者 这篇文章适合以下读者： 使用 SQLAlchemy 做 ORM 的后端开发者 想把 ORM 模型转换为 JSON/dict 的 Python 工程师 对 模型序列化规范化 有需求的中级开发者 使用 Flask/FastAPI/Django + SQLAlchemy 的同学 一、背景 / 动机：为什么要自己写 _to_dict？ 在 Web 开发中，我们几乎每天都要做一件事： 把数据库里的 ORM 对象，转成可以 JSON 响应给前端的数据结构（通常是 dict / list）。 乍一看好像只是 obj.__dict__ 或用个 asdict 就完事，但现实中的问题包括： 日期时间字段无法直接 JSON 化： datetime / date 对象不能直接 JSON 序列化，必须格式化成字符串。 关系字段怎么处理？ 一对多 / 多对多（uselist=True） 一对一 / 多对一（uselist=False） 避免递归爆炸： 两个模型互相关联，很容易序列化时陷入无限递归。 ...

标题 别让 Pydantic 占领你的整个项目：聊聊 API 校验、Domain 模型和数据库之间的边界 副标题 / 摘要 很多用 FastAPI/Pydantic 的 Python 工程师，会不知不觉让 Pydantic Model 贯穿 API、业务、数据库所有层。本文用一个清晰的分层思路和完整代码示例，帮你搞清楚：Pydantic 适合用在什么地方，Domain / ORM 又应该怎么配合。 目标读者 这篇文章适合： 正在使用 FastAPI / Pydantic / SQLAlchemy / SQLModel 的 Python 后端工程师 刚入行 0–3 年、开始关心“分层、架构、领域模型”的开发者 想从“会写接口”进阶到“懂业务建模、懂分层”的工程师 对 “Pydantic 要不要进 Domain / 要不要用于 DB 模型” 有疑惑的人 一、背景 / 动机：为什么 Pydantic 容易“长满全项目”？ 如果你是从 FastAPI 入门后端，很可能经历过这样的路径： 用 Pydantic 定义请求体、响应体：太好用了，自动校验 + 文档 + 类型提示。 觉得既然 Pydantic 这么香，那干脆： 直接拿 Pydantic Model 当“业务对象”传来传去 甚至顺手拿它去做“数据库模型” 渐渐地，你的项目变成： ...

标题 从写路由到写“大脑”：Python 工程师如何先搞定核心逻辑，再考虑 API 副标题 / 摘要 刚入行时，我们常常一上来就写路由、设计接口、想 chat_id / message_id 怎么存，却发现真正的“智力活”——核心逻辑——总是拖到后面。这篇文章带你从「先写接口」的思维，升级到「先写大脑，再接外壳」，并串起来六边形架构、Clean Architecture、DDD 等背后的经典理念。 目标读者 适合这些同学阅读： 1–3 年经验 的 Python 后端工程师 / AI 应用开发者 正在用 FastAPI / Django / Flask 等框架写 API 的工程师 想从“CRUD 搬砖工”进化为“懂设计、能抽象”的工程师 对 六边形架构 / Clean Architecture / DDD 有点好奇但没系统看过书的人 一、背景 / 动机：为什么“先写接口”会卡死自己？ 很多刚入行的 Python 工程师（包括你我）会有这样的流程： 产品提一个新需求：做一个 AI 聊天功能。 打开编辑器，第一反应就是： 设计 URL：POST /api/chat/send_message 开始写 router：@app.post("/chat/send") 想 request body 参数长什么样：chat_id / message_id / user_id / content 想数据库表结构：chats，messages 写了一堆 API、schema、model、迁移脚本之后，才想起来： “那 AI 回复到底是怎么生成的？” ...

🚀 从阻塞到异步：为什么上传接口不该等文件处理完？ —— 用异步任务和状态跟踪构建高性能文件处理系统 🧭 副标题 / 摘要 在现代 Web 系统中，文件上传只是起点，真正的挑战在于后续的解析、索引和处理。本文带你理解为什么“上传接口不等待处理完成”是现代架构的核心理念，以及如何通过异步任务 + 状态查询实现稳定、可扩展的后台处理系统。 👥 目标读者 有一定 Web 开发经验的工程师（Python/FastAPI/Node.js 等） 想优化后端性能、提高可扩展性的中级开发者 对架构设计、异步系统感兴趣的工程师或技术负责人 🎯 背景 / 动机 很多初学者写上传接口时会这样做： @app.post("/upload") def upload_file(file: UploadFile): parse_and_store(file) # 阻塞操作 return {"status": "completed"} 表面简单，实则隐藏问题： ⏱ 超时风险高（解析/embedding/OCR可能几分钟） 🧵 阻塞主线程，拖慢整个 API 服务 💥 请求中断即任务丢失 😕 用户只能干等着，无法看到进度 解决方案就是：上传与处理分离。上传只负责“投递任务”，处理由后台 worker 异步执行，状态存储在数据库中供前端查询。 🔍 核心概念 概念 说明 异步任务（Async Job） 文件解析、OCR、embedding 等耗时操作独立运行，不阻塞主线程。 任务队列（Task Queue） 临时存放待执行的任务，如 Redis、RabbitMQ、Celery。 状态持久化（State Persistence） 将任务状态（pending / processing / completed / failed）写入数据库。 SSE（Server-Sent Events） 一种轻量的实时推送机制，前端可实时接收状态更新。 ⚙️ 实践指南 / 实现步骤 1️⃣ 上传文件接口（只负责入队） @router.post("/upload") async def upload(file: UploadFile, user=Depends(get_verified_user)): file_id = Files.create(file, user.id) # 异步提交任务（Celery、RQ、线程池等） background_tasks.add_task(process_file, file_id) return {"file_id": file_id, "status": "pending"} 2️⃣ 异步任务（后台 worker 执行） def process_file(file_id: str): file = Files.get(file_id) Files.update_status(file_id, "processing") try: parse_and_vectorize(file) Files.update_status(file_id, "completed") except Exception as e: Files.update_status(file_id, "failed", error=str(e)) 3️⃣ 状态查询接口 @router.get("/{id}/process/status") async def get_status(id: str, stream: bool = False): file = Files.get(id) if stream: async def event_stream(): while True: status = Files.get_status(id) yield f"data: {json.dumps({'status': status})}\n\n" if status in ("completed", "failed"): break await asyncio.sleep(1) return StreamingResponse(event_stream(), media_type="text/event-stream") return {"status": file.data.get("status", "pending")} 💻 可运行示例 前端轮询： async function checkStatus(fileId) { let status = 'pending'; while (status === 'pending' || status === 'processing') { const res = await fetch(`/api/files/${fileId}/process/status`); const data = await res.json(); status = data.status; console.log("当前状态:", status); await new Promise(r => setTimeout(r, 1000)); } if (status === 'completed') alert("解析完成！"); } 前端 SSE 实时监听： const evtSource = new EventSource(`/api/files/${fileId}/process/status?stream=true`); evtSource.onmessage = (e) => { const { status } = JSON.parse(e.data); console.log("文件状态:", status); if (status === "completed") evtSource.close(); }; 🧠 原理解释与取舍 模式 特点 适用场景 同步上传+处理 实现简单，但阻塞主线程 小文件、低并发、离线脚本 异步上传+状态查询（推荐） 非阻塞、可恢复、可扩展 Web 应用、后台任务 消息队列驱动 支持分布式任务、重试机制 大规模系统、微服务架构 取舍原则： ...

🔌 为什么让前端执行 Chat Completion：一套通用的多模型流式对话架构设计 副标题 / 摘要 在现代 AI 聊天系统中，很多人会问：为什么不直接在后端调用 OpenAI API？ 本文将带你理解一种更灵活的架构——让前端承担推理执行，后端负责调度和状态同步。适合需要支持多模型、本地推理或用户自带 API Key 的开发者。 目标读者 AI 聊天应用开发者 WebSocket / Socket.IO 实践者 想构建多模型、多端协作聊天系统的架构师 🧠 背景 / 动机 传统的聊天后端往往直接在服务器调用 OpenAI API： resp = client.chat.completions.create(model="gpt-4o", messages=messages) 虽然简单，但带来几个现实问题： 所有请求都消耗服务器的 Key，成本高且难追踪； 无法支持用户自定义 Key（BYOK 模式）； 无法连接用户本地推理（如 Ollama、LM Studio）； 无法切换不同模型或 API Base URL； 前后端状态不同步，不利于流式消息推送。 为了解决这些问题，一些开源系统（如 Open-WebUI、Chatbot-UI 增强版）采用了更灵活的 Socket.IO 双向通信架构。 服务端负责「调度与状态流」，前端负责「执行与回传」。 🧩 核心概念 概念 说明 Socket.IO 基于 WebSocket 的实时双向通信库，支持事件与回调。 event_emitter 服务端向前端广播事件（推送消息/状态）。 event_caller (sio.call) 服务端请求前端执行任务（RPC），并等待前端 callback 返回。 request:chat:completion 一种自定义事件类型，用于请求前端执行 chat completion。 BYOK 模式 “Bring Your Own Key”，用户使用自己的 OpenAI Key 调用 API。 Executor 架构 前端承担推理任务的执行者，后端作为协调者。 🧭 实践指南 / 步骤 1️⃣ 服务端发送调用请求 res = await event_caller({ "type": "request:chat:completion", "data": { "form_data": form_data, "model": models[form_data["model"]], "channel": channel, "session_id": session_id, }, }) 这里的 event_caller 使用 sio.call() 发送事件给指定客户端，并等待 callback 返回。 ...

🛰️ WebSocket 深入理解：为什么要保持一个“永远在线”的连接？ ✨ 副标题 / 摘要 这篇文章带你彻底搞懂 WebSocket： 它和 HTTP 的根本区别、为什么需要“长连接”、连接是如何建立和保持的、以及它在实时应用中的意义。 适合想从“知道是什么”到“理解为什么”的开发者。 👩‍💻 目标读者 Web 前后端初级到中级开发者 想实现实时聊天、AI 流式输出、协作系统的工程师 想从 HTTP 模型过渡到实时架构思维的学习者 🧭 背景 / 动机：为什么这个问题重要？ 几乎每个现代 Web 应用都涉及“实时”功能： 聊天对话（ChatGPT、Slack） 实时通知（邮箱、消息提醒） 在线协作（Notion、Google Docs） 数据看板（实时指标、监控） 然而，传统的 HTTP 是“一问一答”的协议， 无法满足服务器主动通知客户端、低延迟双向通信的需求。 WebSocket 的出现，彻底改变了这种单向关系， 让 Web 应用第一次真正拥有了“实时对话”的能力。 🧠 核心概念与术语解释 名称 说明 HTTP 一问一答型协议。客户端发请求，服务器回响应，然后断开。 长连接 一条保持不关闭的 TCP 连接，可反复收发数据。 WebSocket 一种基于 TCP 的双向通信协议，能让服务器主动推送消息。 握手 (Handshake) 客户端通过 HTTP 请求告诉服务器：“我想升级为 WebSocket 协议”。 帧 (Frame) WebSocket 传输的最小数据单元，比 HTTP header 更轻量。 心跳 (Ping/Pong) 定期发送的小数据包，防止连接超时断开。 🪜 实践指南：WebSocket 建立的全过程 1️⃣ 浏览器发起请求（HTTP 阶段） ...

🚀 从 Pip 到 UV：一站式 Python 包管理与依赖同步指南 💡 副标题 / 摘要 想让你的 Python 环境更干净、更快、更可靠？本文将带你从传统的 pip + venv + requirements.txt 迁移到现代的 uv 包管理系统，并教你如何在两者之间无缝同步。 🎯 目标读者 适合 Python 开发者（初学者到中级）、数据科学家、后端工程师，以及希望提升开发环境一致性、减少依赖地狱的读者。 🔥 背景 / 动机 在日常 Python 开发中，我们经常遇到以下痛点： 环境混乱、包冲突； pip install 太慢； 不同机器、团队成员环境不一致； requirements.txt 手动维护麻烦。 而 uv 是一个由 Astral 团队推出的新一代包管理工具， 用 Rust 编写，集成了： 包安装（比 pip 快数倍）； 虚拟环境管理； 锁文件机制（可复现环境）； 与 PyPI 完全兼容。 一句话：uv = pip + virtualenv + pip-tools + poetry 的融合体。 🧩 核心概念 概念 说明 pyproject.toml 现代 Python 项目的依赖与元信息文件 uv.lock 锁文件，记录所有依赖的精确版本，保证可复现 uv sync 根据锁文件同步环境（自动创建/更新虚拟环境） uv add / remove 添加或删除依赖，并自动更新锁文件 uv export 导出为 requirements.txt，兼容传统 pip 流程 🛠 实践指南 / 步骤 一、从 pip 项目迁移到 uv 假设你已有一个项目： ...

🧩 如何高效审核 FastAPI 后端项目的 Pull Request（PR） 副标题 / 摘要： 本文为你系统梳理了在 Python FastAPI 项目中如何进行专业的代码审核流程，从逻辑正确性到安全、性能与架构一致性，附带实用审查清单与示例，助你成为团队中更高效的 Reviewer。 👥 目标读者 使用 Python + FastAPI 的中高级后端开发者 初入团队、需要学习代码审查流程的工程师 负责代码质量与合并决策的 Tech Lead / Reviewer 💡 背景与动机 在多人协作的后端项目中，代码审查（Code Review） 是保障系统稳定、提升团队代码质量的关键环节。 但许多工程师在面对 PR 时往往只“浏览一下改动”，忽略了逻辑、性能和安全的隐患。 尤其在 FastAPI 项目中，接口结构简洁、异步特性突出，但也因此容易出现： 不当的 async/await 用法导致阻塞； 不安全的输入校验； 不一致的 Schema 与返回模型； 难以维护的业务逻辑。 因此，本文将教你如何 系统化、标准化地审查 FastAPI PR。 🧠 核心概念 概念 说明 PR (Pull Request) 在 Git 平台上发起代码合并请求，等待他人审核后合并到主分支。 Code Review 同事间对代码进行质量和设计审查的过程。 FastAPI 高性能、异步的 Python Web 框架，基于 Pydantic 和 Starlette。 Pydantic Schema FastAPI 的数据验证与序列化模型系统。 Depends() FastAPI 的依赖注入机制，用于数据库连接、认证等。 🧭 实践指南：PR 审核流程 1️⃣ 阅读 PR 描述 明确改动目的、功能范围、对应 issue。 判断是否为修复、功能新增、重构或优化。 2️⃣ 浏览改动文件 注意核心目录：routers/, schemas/, models/, services/, core/。 检查是否包含依赖变更、配置修改或多余文件。 3️⃣ 深入逻辑代码 重点审查： ...