fastapi-notification-module-skill

FastAPI Notification Module Skill

面向已有 FastAPI 项目的开发者，快速集成通知能力。

能力清单

能力	说明
短信发送	验证码/通知短信
邮件推送	HTML/文本邮件/模板邮件
站内消息	系统通知/未读消息
推送通知	极光/FCM
消息模板	模板管理和渲染

触发场景

用户说"帮我加短信"或"集成通知"时触发。

核心实现

依赖配置

# 短信
pip install aliyun-python-sdk-core

# 邮件
pip install aiosmtpd jinja2

# 站内消息
pip install redis

配置

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # 短信配置
    sms_access_key: str = ""
    sms_secret_key: str = ""
    sms_sign_name: str = ""
    sms_template_code: str = ""
    
    # 邮件配置
    mail_host: str = "smtp.qq.com"
    mail_port: int = 587
    mail_username: str = ""
    mail_password: str = ""
    
    # Redis 配置
    redis_host: str = "localhost"
    redis_port: int = 6379

数据模型

# models.py
from sqlalchemy import Column, BigInteger, String, Text, Integer, DateTime, Enum as SQLEnum
from database import Base
import enum

class NotificationType(str, enum.Enum):
    SYSTEM = "system"
    SMS = "sms"
    EMAIL = "email"
    PUSH = "push"

class Notification(Base):
    __tablename__ = "wg_notification"
    
    id = Column(BigInteger, primary_key=True, autoincrement=True)
    user_id = Column(String(50), nullable=False, index=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    type = Column(SQLEnum(NotificationType), default=NotificationType.SYSTEM)
    link = Column(String(500))
    is_read = Column(Integer, default=0)
    created_at = Column(DateTime, server_default=func.now())

class SmsCode(Base):
    __tablename__ = "wg_sms_code"
    
    id = Column(BigInteger, primary_key=True, autoincrement=True)
    phone = Column(String(20), nullable=False, index=True)
    code = Column(String(10), nullable=False)
    template_code = Column(String(50))
    expire_time = Column(DateTime)
    used = Column(Integer, default=0)
    created_at = Column(DateTime, server_default=func.now())