乘风的人 - Python 框架
type: channel
link: https://blog.zhangbiechen.cn/index.php/tag/Python-框架
items: Python 框架 - APScheduler （定时任务调度框架 ）
items: Python框架 - aiohttp 异步HTTP客户端/服务器框架

Python 框架 - APScheduler （定时任务调度框架 ）
type: item
date: 2026-02-06T13:47:42+08:00
description: APScheduler 从零到实战（FastAPI 定时任务全指南）本文面向已掌握 Python 异步基础和 FastAPI 路由用法、无 APScheduler 经验的自学开发者，从零开始、由浅入深讲解 APScheduler 的使用，重点聚焦 FastAPI 异步环境下的定时任务实现，覆盖基础、中级、进阶及集成全流程，每个知识点均包含「原理说明+最小可运行代码+常见陷阱」，解决实际开发中后台定时任务（日志清理、数据同步、自定义提醒等）的核心需求。前言：为什么选择 APScheduler？在 FastAPI 中实现定时任务，常见方案有 APScheduler、Celery、BackgroundTasks 等，其中 APScheduler 具备以下优势，适配大多数中小规模定时任务场景：轻量无依赖（核心功能无需额外组件），部署简单；支持同步/异步任务，完美适配 FastAPI 异步环境；灵活的触发器（interval/cron/date），满足各类定时需求；支持任务持久化、动态管理，可与数据库无缝集成；无需像 Celery 那样依赖 Redis/RabbitMQ 作为消息队列，降低架构复杂度。本文将围绕 APScheduler 展开，全程结合 FastAPI 实战，解决「怎么用」「怎么用好」「怎么避坑」三个核心问题。基础篇：APScheduler 核心入门基础篇目标：掌握 APScheduler 三大核心组件，实现简单的内存级定时任务，理解不同触发器的用法。1.1 核心概念（Job、Trigger、Scheduler）原理说明APScheduler 的核心由 3 个组件构成，三者协同工作完成定时任务的调度与执行：Job（任务）：最基本的执行单元，封装了要执行的函数（同步/异步）、参数、执行策略（如重试次数、执行超时时间）。Trigger（触发器）：定义 Job 的执行时机，决定「什么时候执行任务」，APScheduler 提供 3 种内置触发器（覆盖所有常见场景）。Scheduler（调度器）：核心控制器，负责将 Job 与 Trigger 绑定，管理 Job 的生命周期（添加、删除、暂停、执行），同时负责任务的存储、线程/进程管理。补充：基础篇使用「内存存储」（默认），即任务信息仅保存在内存中，服务重启后任务丢失，适合临时测试。最小可运行代码（核心组件演示）#!/usr/bin/env python3 # -*- coding: utf-8 -*- # 基础篇：核心组件（Job + Trigger + Scheduler）演示 from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger # 1. 定义要执行的任务函数（同步函数，基础篇默认用同步） def print_hello(name: str = "默认用户"): """简单的任务函数，打印问候语""" print(f"Hello, {name}! 这是定时任务的输出") # 2. 初始化调度器（基础篇用 BlockingScheduler，阻塞式调度器，适合单独运行的定时任务脚本） # BlockingScheduler 会阻塞当前线程，直到手动停止（Ctrl+C） scheduler = BlockingScheduler() # 3. 创建 Trigger（触发器）：每 5 秒执行一次 trigger = IntervalTrigger(seconds=5) # 4. 创建 Job（任务），并绑定到调度器 # add_job 方法：将任务、触发器绑定到调度器，返回 Job 对象（可用于后续管理） job = scheduler.add_job( func=print_hello, # 要执行的任务函数 trigger=trigger, # 触发器（执行时机） args=["APScheduler 新手"], # 任务函数的位置参数（可选） kwargs={"name": "APScheduler 新手"}, # 任务函数的关键字参数（可选） id="hello_job", # 任务唯一ID（必填，用于后续动态管理） replace_existing=True # 如果存在相同id的任务，替换它（避免重复添加） ) # 5. 启动调度器（阻塞当前线程，开始执行任务） if __name__ == "__main__": try: print("调度器已启动，按 Ctrl+C 停止...") scheduler.start() except (KeyboardInterrupt, System…

Python框架 - aiohttp 异步HTTP客户端/服务器框架
type: item
date: 2026-02-04T14:40:00+08:00
description: aiohttp 从基础到进阶学习指南（适配 FastAPI 异步开发）适用人群：已掌握 Python 基础、FastAPI 路由基本用法，无 asyncio 实战经验，计划用 FastAPI + aiohttp 进行异步 Web 开发的自学开发者。核心定位：aiohttp 是 Python 主流异步 HTTP 客户端/服务器框架，本文重点聚焦 aiohttp 客户端（适配 FastAPI 路由中调用第三方 API 的场景），全程贴合 FastAPI 异步开发最佳实践，规避实战陷阱。基础篇：async/await 核心概念 + aiohttp 客户端最简用法一、async/await 核心原理（极简版）原理说明Python 异步编程的核心是“非阻塞等待”，解决 CPU 等待 I/O（如 HTTP 请求、数据库操作）时的资源浪费问题，关键语法就是 async 和await：async def：定义“异步函数”（协程函数），该函数调用后不会立即执行，而是返回一个“协程对象”（coroutine）。await：只能用在异步函数内部，用于“等待”另一个协程完成（如等待 HTTP 请求响应），等待期间 CPU 可执行其他任务，实现“并发”。补充：异步函数不能直接调用执行（如 func() 无效），必须通过 asyncio.run(func())（Python 3.7+）启动，或交给事件循环（event loop）调度。极简示例（先理解语法，再结合 aiohttp）import asyncio # 1. 用 async def 定义异步函数（协程函数） async def async_task(): print("异步任务开始") # 模拟 I/O 等待（如 HTTP 请求、数据库查询），用 asyncio.sleep 替代 # 注意：不能用 time.sleep()，它是阻塞的，会卡住整个事件循环 await asyncio.sleep(1) # 等待 1 秒，期间 CPU 可做其他事 print("异步任务结束") # 2. 用 asyncio.run() 启动异步函数（Python 3.7+ 推荐写法） if __name__ == "__main__": asyncio.run(async_task()) # 自动创建事件循环、运行协程、关闭循环 常见陷阱陷阱1：在异步函数内用阻塞函数（如 time.sleep()、requests.get()），会导致整个事件循环卡住，所有异步任务都无法执行。陷阱2：直接调用异步函数（如 async_task()），不会执行任务，只会返回协程对象（无任何效果）。二、aiohttp 客户端最简用法（GET/POST）原理说明aiohttp 是基于 asyncio 实现的异步 HTTP 客户端，核心优势是“异步发送 HTTP 请求”，与 FastAPI 配合时，能让整个 Web 应用保持非阻塞，提升并发处理能力（对比 requests：requests 是阻塞式，会卡住 FastAPI 的事件循环）。aiohttp 客户端的核心是 ClientSession：相当于一个“异步请求会话”，所有 HTTP 请求（GET/POST 等）都通过它发送，会话内可复用连接，提升效率。1. GET 请求（最简可运行示例）import asyncio import aiohttp # 异步函数：用 aiohttp 发送 GET 请求 async def fetch_get(url: str): # 1. 创建 ClientSession 实例（会话），推荐用 async with 自动关闭会话（避免资源泄漏） async with aiohttp.ClientSession() as session: # 2. 用 session.get() 发送 GET 请求，await 等待响应 # session.get() 是异步方法，必须用 await 等待其完成 async with session.get(url) as response: # 3. 解析响应：text() 异步获取文本内容，json() 异步获取 JSON 内容（按需选择） # 注意：response.text()/json() 也是异步方法，必须加 await response_text = await response.text() response_json = await response.json() # 若接口返回 JSON 格式，可用这个 print(f"GET 请求状态码：{response.status}") print(f"GET 请求响应文本（前100字符）：{response_text[:100]}") return resp…