> **用于发送 Kakao 通知消息、品牌消息与短信的官方 FastAPI 扩展包**

`sendgo-fastapi` 把 [`sendgo-python`](https://github.com/send-go/python) 核心封装为 FastAPI **依赖项**，配置通过 pydantic-settings 模型读取。

要求 FastAPI 0.110+ 与 Python 3.10+。

---

## 安装

```bash
pip install sendgo-fastapi
```

核心 `sendgo-python` 会作为依赖一并安装。

---

## 配置

`SendgoSettings` 读取以下环境变量，因此代码里无需做任何初始化：

```env
SENDGO_ACCESS_KEY=your_access_key
SENDGO_SECRET_KEY=your_secret_key
SENDGO_KAKAO_SENDER_KEY=your_kakao_sender_key
SENDGO_SMS_SENDER_KEY=your_sms_sender_key
SENDGO_API_VERSION=v2
```

`access_key` 与 `secret_key` 没有默认值，因此缺失时 pydantic 会在**启动时**抛出校验错误 —— 配置错误的部署会立刻失败，而不是等到第一次发送。

---

## 快速上手

```python
from fastapi import Depends, FastAPI
from sendgo_fastapi import SendgoDep

app = FastAPI()

@app.post("/notify")
async def notify(phone: str, order_no: str, sendgo = Depends(SendgoDep)):
    return sendgo.alimtalk.send(
        template_code="ORDER_CONFIRM_001",
        contacts=[{"contact": phone, "var1": order_no}],
    )
```

该依赖项产出的就是核心客户端，因此各渠道都可用：

| 属性 | 渠道 |
|------|------|
| `sendgo.alimtalk` | Kakao 通知消息 |
| `sendgo.friendtalk` | Kakao 好友消息 |
| `sendgo.brand_message` | Kakao 品牌消息（仅 v2） |
| `sendgo.sms` | SMS / LMS / MMS |

### Annotated 写法

用 `Annotated` 可以把依赖声明并入类型，签名更整洁：

```python
from typing import Annotated
from fastapi import Depends
from sendgo import Sendgo
from sendgo_fastapi import SendgoDep

SendgoClient = Annotated[Sendgo, Depends(SendgoDep)]

@app.post("/notify")
async def notify(phone: str, order_no: str, sendgo: SendgoClient):
    return sendgo.alimtalk.send(
        template_code="ORDER_CONFIRM_001",
        contacts=[{"contact": phone, "var1": order_no}],
    )
```

---

## 品牌消息

> **好友消息已于 2025-12-31 停止服务。** 自 2026-01-01 起，Kakao 会将好友消息请求
> 自动改以品牌消息（自由形式）发出。`friendtalk` 仍可调用，且自由文本类型
> `FT`/`FI`/`FW` 发往单个接收者的路径目前仍只有这一条 —— 基于模板的富文本类型
> （`FL`/`FC`/`FM`/`FP`/`FA`）、非好友定向（`N`/`I`）与群发（`F`）请使用品牌消息。
品牌消息是好友消息的后继渠道：可触达**非频道好友**的接收者（`targeting="N"`），也可向**已同意接收的全部频道好友群发**（`targeting="F"`）。消息类型与好友消息一一对应（`FT`→`BT`、`FI`→`BI`、`FW`→`BW`、`FL`→`BL`、`FC`→`BC`、`FM`→`BM`、`FP`→`BP`、`FA`→`BA`）—— 传入好友消息的代码，服务端会自动转换。

> 仅 v2 支持。请设置 `SENDGO_API_VERSION=v2`。

```python
@app.post("/campaigns/targeted")
async def targeted(phone: str, sendgo: SendgoClient):
    return sendgo.brand_message.send(
        targeting="M",
        message_type="FL",
        friend_template_uuid="9cd5460b-6458-4edc-9b11-c26d3013c340",
        contacts=[{"contact": phone, "var1": "29,000元"}],
    )

@app.post("/campaigns/broadcast")
async def broadcast(sendgo: SendgoClient):
    # 不传接收者列表 —— 由 Kakao 展开受众
    return sendgo.brand_message.broadcast(
        message_type="FW",
        friend_template_uuid="9cd5460b-6458-4edc-9b11-c26d3013c340",
    )

@app.get("/campaigns")
async def campaigns(sendgo: SendgoClient, count: int = 10):
    # `from` 是 Python 关键字，因此参数名为 `from_`。
    return sendgo.brand_message.campaigns(count=count)

@app.get("/campaigns/{campaign_id}")
async def campaign(campaign_id: str, sendgo: SendgoClient):
    return sendgo.brand_message.campaign(campaign_id)
```

群发在上游为异步处理，因此发送响应只表示已被接受 —— 请轮询 `GET /campaigns/{campaign_id}` 查看进度。

群发可能触达整个好友列表，请为该端点加上依赖式鉴权，不要公开暴露。

`targeting` 的取值含义：

| 值 | 含义 |
|----|------|
| `M` | 频道好友中的指定接收者 |
| `N` | 非频道好友的接收者 |
| `I` | 按标识符指定的接收者 |
| `F` | 已同意接收的全部频道好友（群发） |

---

## 后台任务

核心客户端是**同步**的，因此在 `async def` 端点里直接调用会阻塞事件循环。请把发送移到后台任务：

```python
from fastapi import BackgroundTasks, Depends, FastAPI
from sendgo import Sendgo, SendgoError
from sendgo_fastapi import SendgoDep

def deliver(sendgo: Sendgo, phone: str, order_no: str) -> None:
    try:
        sendgo.alimtalk.send(
            template_code="ORDER_CONFIRM_001",
            contacts=[{"contact": phone, "var1": order_no}],
        )
    except SendgoError as e:
        logger.error("Sendgo %s [%s]: %s", e.status_code, e.error_code, e)

@app.post("/orders", status_code=202)
async def create_order(
    phone: str,
    order_no: str,
    background: BackgroundTasks,
    sendgo = Depends(SendgoDep),
):
    background.add_task(deliver, sendgo, phone, order_no)

    return {"accepted": True}
```

需要在重启后依然可靠的场景，请使用真正的队列（Celery、ARQ、Dramatiq），而不是 `BackgroundTasks` —— 后者只存在于进程内存中，进程结束即丢失。

### 使用线程池

若需要在请求内拿到结果，可用 `run_in_threadpool` 让事件循环保持空闲：

```python
from starlette.concurrency import run_in_threadpool

@app.post("/notify")
async def notify(phone: str, sendgo: SendgoClient):
    return await run_in_threadpool(
        sendgo.alimtalk.send,
        template_code="ORDER_CONFIRM_001",
        contacts=[{"contact": phone}],
    )
```

---

## 短信 / 长短信 / 多媒体短信

```python
@app.post("/sms")
async def sms(phone: str, sendgo: SendgoClient):
    return sendgo.sms.send_sms(
        content="[Sendgo] 验证码：123456（请在 5 分钟内输入）",
        contacts=[{"contact": phone}],
    )

@app.post("/lms")
async def lms(phone: str, sendgo: SendgoClient):
    return sendgo.sms.send_lms(
        content="服务将于 2026-07-25 02:00 ~ 06:00 进行维护。",
        contacts=[{"contact": phone}],
        subject="[重要] 服务维护通知",
    )
```

---

## 测试

覆盖依赖项，这样测试就不会真的发出请求：

```python
from unittest.mock import MagicMock
from fastapi.testclient import TestClient
from sendgo_fastapi import SendgoDep

fake = MagicMock()
app.dependency_overrides[SendgoDep] = lambda: fake

client = TestClient(app)
client.post("/notify", params={"phone": "01012345678", "order_no": "ORD-001"})

fake.alimtalk.send.assert_called_once()

app.dependency_overrides.clear()
```

务必调用 `clear()`（或用 fixture 处理），否则覆盖会泄漏到后续测试。

---

## 错误处理

把 Sendgo 的错误映射为 HTTP 状态码，而不是让它变成 500：

```python
from fastapi import HTTPException
from sendgo import SendgoError

@app.post("/notify")
async def notify(phone: str, sendgo: SendgoClient):
    try:
        return sendgo.alimtalk.send(
            template_code="ORDER_CONFIRM_001",
            contacts=[{"contact": phone}],
        )
    except SendgoError as e:
        if e.error_code in ("INVALID_ACCESS_KEY", "INVALID_SECRET_KEY", "IP_NOT_ALLOWED"):
            # 是我们的配置有误，而非调用方的请求有误。
            logger.critical("Sendgo 配置错误：%s", e.error_code)
            raise HTTPException(status_code=500, detail="消息服务不可用") from e

        if e.error_code == "PAYMENT_REQUIRED":
            raise HTTPException(status_code=402, detail="余额不足") from e

        if e.status_code >= 500:
            raise HTTPException(status_code=502, detail="上游错误") from e

        raise HTTPException(status_code=400, detail=e.error_code) from e
```

请根据 `error_code` 分支，而不要匹配错误消息文本 —— 消息文案可能变更，错误代码才是契约。
`TOKEN_EXPIRED` 与 `TOKEN_MISMATCH` 已在核心 SDK 内部处理：会重新签发令牌并自动重试该请求一次。

---

## 配置项

| 环境变量 | 字段 | 是否必填 | 默认值 | 说明 |
|----------|------|----------|--------|------|
| `SENDGO_ACCESS_KEY` | `access_key` | **必填** | — | Sendgo 访问密钥 |
| `SENDGO_SECRET_KEY` | `secret_key` | **必填** | — | Sendgo 私密密钥 |
| `SENDGO_KAKAO_SENDER_KEY` | `kakao_sender_key` | 选填 | `None` | Kakao 发送者资料密钥 |
| `SENDGO_SMS_SENDER_KEY` | `sms_sender_key` | 选填 | `None` | 短信主叫号码密钥 |
| `SENDGO_API_VERSION` | `api_version` | 选填 | `"v1"` | API 版本（`v1` \| `v2`） |
| `SENDGO_BASE_URL` | `base_url` | 选填 | `"https://sendgo.io"` | API 基础地址 |

---

## 短链接

短链接会缩短消息正文中的链接，并统计该链接是否被真正点击。
短信按字节计费，因此链接更短就意味着正文可以写更多内容。

> 仅 v2 支持。

再次缩短同一个原始 URL 会**直接返回已有的链接**。若希望按活动分别统计反应，
请使用 `forceNew` 生成新的短码。

`deactivate` 不会删除链接，只停止重定向。当已发送消息中的链接必须失效时使用它；
累计统计会保留，访问已停止的链接会返回 `410 Gone`。

```python
@app.post("/shorten")
async def shorten(target_url: str, sendgo: SendgoClient):
    created = sendgo.short_url.create(target_url=target_url)

    return {"shortUrl": created["data"]["shortUrl"]}

@app.get("/shorten/{code}/stats")
async def link_stats(code: str, sendgo: SendgoClient):
    return sendgo.short_url.stats(code)
```

`stats` 返回按日趋势（`daily`）以及按设备（`byDevice`）、来源（`byReferer`）、国家（`byCountry`）的分解。按日趋势读取的是预聚合表，因此点击量再大响应时间也保持稳定。

---

## 包信息

- **包名**：`sendgo-fastapi`（PyPI）
- **仓库**：[send-go/fastapi](https://github.com/send-go/fastapi)
- **注册表**：https://pypi.org/project/sendgo-fastapi/
- **许可**：MIT

### 如何获取 API 密钥

登录 Sendgo 后，在 **API/SDK → API 对接** 菜单中签发访问密钥与私密密钥。
使用通知消息／好友消息需在 **Kakao 渠道** 登记发送者资料以获取 `SENDGO_KAKAO_SENDER_KEY`；使用短信需在 **发送号码** 菜单登记主叫号码。