FastAPI-utils定时任务终极教程：轻松实现周期性后台任务
【免费下载链接】fastapi-utilsReusable utilities for FastAPI项目地址: https://gitcode.com/gh_mirrors/fa/fastapi-utils
FastAPI-utils是一款强大的FastAPI增强工具库，其中的定时任务功能能够帮助开发者轻松实现周期性后台任务。本文将详细介绍如何使用FastAPI-utils的@repeat_every装饰器，让你快速掌握在FastAPI应用中创建定时任务的方法。
为什么选择FastAPI-utils定时任务？
在开发FastAPI应用时，我们经常需要执行一些周期性的后台任务，例如定期清理缓存、删除过期数据或生成统计报告等。虽然可以通过启动事件触发循环来实现，但这需要处理诸多问题：确保启动事件在循环结束前完成、避免阻塞事件循环、处理任务异常等。
FastAPI-utils的fastapi_utils.tasks.repeat_every装饰器完美解决了这些问题，提供了便捷、可靠的定时任务实现方式。
快速入门：@repeat_every装饰器
@repeat_every装饰器是FastAPI-utils定时任务功能的核心。当被装饰的函数被调用时，会启动一个循环，按照指定的时间间隔周期性地调用该函数。如果配合@app.event("startup")装饰器使用，FastAPI会在服务器启动时调用该函数，从而实现服务器运行期间的周期性任务执行。
下面是一个定期清理过期令牌的示例：
from fastapi import FastAPI from fastapi_utils.tasks import repeat_every from fastapi_utils.session import FastAPISessionMaker app = FastAPI() sessionmaker = FastAPISessionMaker("sqlite:///mydb.db") @app.on_event("startup") @repeat_every(seconds=60 * 60) # 1 hour def remove_expired_tokens_task() -> None: with sessionmaker.context_session() as session: # 在这里执行删除过期令牌的操作 session.query(Token).filter(Token.expires < datetime.now()).delete() session.commit()通过设置seconds=60 * 60，我们确保被装饰的函数每小时被调用一次。
@repeat_every装饰器的关键参数
@repeat_every装饰器提供了多个关键字参数，以满足不同的定时任务需求：
seconds: float
指定连续调用之间的等待秒数。这是创建定时任务最基本的参数，用于控制任务的执行频率。
wait_first: float | None = None
如果不为None，函数将在第一次调用前等待指定的秒数。这对于需要延迟启动的任务非常有用。
max_repetitions: int | None = None
指定函数被调用的最大次数。如果为None（默认值），函数将永远重复执行；否则，在达到指定次数后停止。
on_complete: NoArgsNoReturnAnyFuncT | None = None
在最后一次重复执行被装饰函数后调用的函数。可以用于执行清理或最终报告生成等操作。
on_exception: ExcArgNoReturnAnyFuncT | None = None
当被装饰的函数引发异常时调用的函数。用于自定义异常处理逻辑，确保任务异常不会被默默忽略。
处理定时任务中的异常
在定时任务执行过程中，妥善处理异常至关重要。FastAPI-utils提供了灵活的异常处理机制：
- 使用
on_exception参数指定异常处理函数，该函数将在任务抛出异常时被调用。 - 异常处理函数可以是同步或异步函数，接受一个
Exception参数。
例如，我们可以创建一个异常处理函数来记录错误并尝试恢复：
def handle_task_exception(exc: Exception) -> None: logger.error(f"定时任务执行失败: {exc}") # 这里可以添加恢复逻辑，如重新连接数据库等 @app.on_event("startup") @repeat_every(seconds=3600, on_exception=handle_task_exception) def my_periodic_task() -> None: # 执行可能抛出异常的操作 ...同步与异步任务支持
@repeat_every装饰器同时支持同步和异步函数：
- 对于异步函数（使用
async def定义），装饰器将直接在事件循环中执行。 - 对于同步函数（使用
def定义），装饰器会在线程池中执行，避免阻塞事件循环，就像处理同步端点一样。
这使得处理各种类型的任务变得灵活方便，无论是IO密集型还是CPU密集型操作。
最佳实践与注意事项
任务函数无参数：被
@repeat_every装饰的函数不应接受任何必需参数。如果需要传递参数，可以使用functools.partial或其他方式包装目标函数。避免长时间运行的任务：尽量确保定时任务能够在指定的时间间隔内完成，避免任务重叠执行。如果任务执行时间可能超过间隔时间，考虑增加时间间隔或优化任务性能。
合理设置异常处理：始终为定时任务设置适当的异常处理机制，避免任务因未处理的异常而意外停止。
使用日志记录：结合日志记录功能，跟踪任务执行情况和异常信息，便于问题排查和系统监控。
考虑任务依赖：如果多个定时任务之间存在依赖关系，需要谨慎设计执行顺序和时间间隔，避免出现竞态条件。
总结
FastAPI-utils的@repeat_every装饰器为FastAPI应用提供了简单而强大的定时任务功能。通过本文的介绍，你已经了解了如何使用这个装饰器创建、配置和管理周期性后台任务，以及如何处理任务执行过程中的异常。
无论是简单的定期清理工作，还是复杂的周期性数据处理，@repeat_every都能帮助你轻松实现。开始使用FastAPI-utils定时任务，提升你的FastAPI应用功能和可靠性吧！
要开始使用FastAPI-utils，你可以通过以下命令克隆仓库：
git clone https://gitcode.com/gh_mirrors/fa/fastapi-utils更多详细信息，请参考官方文档中的repeated-tasks.md和源代码fastapi_utils/tasks.py。
【免费下载链接】fastapi-utilsReusable utilities for FastAPI项目地址: https://gitcode.com/gh_mirrors/fa/fastapi-utils
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
