FastAPI Real-Time Features 是一个针对 FastAPI 项目的技能/代码模式集合,主要覆盖实时通信与外部服务集成三类场景:WebSocket、文件上传到 S3、邮件与通知。资料中提供了可直接参考的代码结构与实现方式。
一、WebSocket 实时通信 该部分给出连接管理器(ConnectionManager)的实现,用于统一管理 WebSocket 连接:
- 连接与断开:connect 方法接受 WebSocket、user_id 和可选的 room_id,先 accept 连接,再按 user_id 保存连接,若指定房间则加入对应房间集合;disconnect 方法从用户连接表和房间集合中移除连接,并清理空房间。
- 消息发送:send_personal 向指定用户发送 JSON 消息;broadcast_to_room 向房间内所有连接广播;broadcast_all 向所有用户连接广播。广播对单个连接异常做了捕获忽略处理(连接已关闭)。
- WebSocket 端点:示例路由 /ws 通过 Query 参数 token 进行认证,使用 decode_token 解析 payload 取得 user_id,失败则以 4001 关闭连接。连接后进入循环接收 JSON 消息并交由 handle_message 处理,断开时调用 manager.disconnect。
- 消息处理:示例支持 join_room(加入房间并广播 user_joined)、message(向房间广播用户消息内容)、typing(广播正在输入状态)。
二、文件上传到 S3
- S3Storage 服务:使用 boto3 客户端(签名版本 s3v4)初始化,读取 aws_access_key_id、aws_secret_access_key、region_name、s3_bucket 等配置。
- 上传文件:upload_file 根据文件名扩展名与 uuid 生成唯一 key,以指定 folder 前缀上传,设置 ContentType 和 ACL 为 private,返回 key。
- 预签名 URL:get_presigned_url 生成下载用预签名 URL,默认有效期 3600 秒;get_presigned_upload_url 生成客户端直传用的预签名 POST,包含字段与条件(Content-Type 匹配、内容长度 1 到 10485760 字节即 10MB 上限)。
- 删除文件:delete_file 调用 delete_object 删除指定 key。
- 上传接口:POST /uploads/ 校验 content_type 是否在允许类型(image/jpeg、image/png、image/gif、application/pdf)内,读取内容校验不超过 10MB,随后上传到 users/{user.id} 目录,并把文件名、s3_key、content_type、size 等元数据写入数据库,返回 id、filename 和预签名 URL。
- 预签名上传接口:GET /uploads/presigned-upload 接收 filename 与 content_type,校验类型后生成 key 并返回 upload_url、fields 和 key。
三、邮件服务
- EmailService:基于 SMTP 配置(smtp_host、smtp_port、smtp_username、smtp_password、from_email、from_name)和 Jinja2 模板环境(模板目录 app/templates/email)构建。
- send 方法:接收收件人(字符串或列表)、主题、模板名、上下文和附件参数,渲染 HTML 模板并组装 MIMEMultipart 消息,然后通过事件循环的 run_in_executor 异步执行 SMTP 发送。
- _send_smtp:使用 smtplib.SMTP,调用 starttls、login、sendmail 完成发送。
- 便捷方法:send_welcome 发送欢迎邮件(welcome 模板,传入 name);send_password_reset 发送密码重置邮件(password_reset 模板,传入 reset_link)。
- 邮件模板:给出 welcome.html 示例,包含 header、content、button 等样式与占位变量(name、dashboard_url)。
四、推送通知
- NotificationService:create 方法创建通知记录(user_id、title、message、type、data),并通过 manager.send_personal 以 type 为 notification 的消息实时推送;mark_read 在校验归属后把通知标记为已读;get_unread 查询指定用户的未读通知列表。
五、附加资源 资料提到可参考的参考文件与示例文件:references/websocket-auth.md、references/s3-advanced.md、references/email-templates.md,以及 examples/websocket_chat.py、examples/file_upload_service.py、examples/notification_worker.py。
适用场景与限制:该技能适合在 FastAPI 后端需要加入实时聊天、实时通知推送、文件上传存储或事务性邮件的项目。其依赖包括 boto3、botocore、Jinja2、smtplib 等;示例中涉及数据库模型(Notification、Upload)、配置项和安全函数(decode_token、get_current_user),需要结合具体项目补齐。价格、平台支持范围与运行效果资料中未作说明,暂未提供。