给 Flask 网店后端接入 Sentry 错误监控:未处理的异常自动上报,带上版本号和环境;发送前过滤 Cookie、Authorization 头和异常信息里的手机号;用标签记录支付方式,方便按渠道筛选错误。
先看清成品
本机用 sentry-sdk 2.72.0 跑了 3 个测试,全部通过:用自定义 transport 截获事件,确认异常被上报、手机号被打码、敏感请求头不出现原值、正常请求不产生事件、支付方式标签存在。没有发送到真实 Sentry 项目,控制台中的展示以实际为准。
注意:Sentry 有免费和付费计划,额度与价格以官方价格页为准。错误数据里可能包含用户个人信息,上线前确认采集范围符合你所在地区的个人信息保护法规(如《个人信息保护法》),并在隐私政策中说明。
做完之后你会得到什么?
| 成果 | 说明 |
|---|---|
| 错误上报 | 未处理异常 + 堆栈 + 请求信息 |
| 隐私过滤 | before_send 打码手机号、过滤认证头 |
| 版本与环境 | release、environment |
| 标签 | payment=alipay 等 |
| 离线测试 | 不连 Sentry 也能验证 |
需要准备哪些环境?
| 项目 | 要求 | 说明 |
|---|---|---|
| Sentry 项目 | 平台选 Flask | 在项目设置里拿到 DSN |
| Python | 3.9 以上 | |
| 依赖 | flask==3.1.3、sentry-sdk[flask]==2.72.0 |
requirements.txt |
文件结构
25-sentry-A-src/
├── app.py
├── requirements.txt
└── tests/test_sentry.py
第 1 步:接入代码
app.py:
import os, re
import sentry_sdk
from flask import Flask, jsonify, request
PHONE = re.compile(r"1[3-9]\d{9}")
def scrub(event, hint):
"""发送前去掉敏感信息:请求头里的 Cookie/Authorization,消息里的手机号。"""
headers = event.get("request", {}).get("headers", {})
for k in list(headers):
if k.lower() in ("cookie", "authorization", "x-api-key"):
headers[k] = "[已过滤]"
for exc in event.get("exception", {}).get("values", []):
if exc.get("value"):
exc["value"] = PHONE.sub("1**********", exc["value"])
return event
def init_sentry(**overrides):
options = dict(
dsn=os.environ.get("SENTRY_DSN"), # 未设置时 SDK 不发送任何数据
environment=os.environ.get("APP_ENV", "development"),
release=os.environ.get("APP_RELEASE", "shop@0.1.0"),
# 不传 send_default_pii / data_collection:使用最保守的默认采集(不带 IP、用户信息、Cookie)
traces_sample_rate=float(os.environ.get("SENTRY_TRACES_RATE", "0.1")),
before_send=scrub,
)
options.update(overrides) # 测试时可替换 transport 等参数
sentry_sdk.init(**options)
app = Flask(__name__)
@app.get("/orders/<int:oid>")
def order(oid):
if oid == 0:
raise ValueError(f"订单不存在,联系电话 13812345678")
return jsonify(id=oid, status="paid")
@app.post("/checkout")
def checkout():
data = request.get_json(silent=True) or {}
sentry_sdk.set_tag("payment", data.get("method", "unknown"))
total = sum(i["price"] * i["qty"] for i in data.get("items", [])) # 缺字段会抛 KeyError
return jsonify(total=total)
if __name__ == "__main__":
init_sentry()
app.run(port=5000)
要点:
- DSN 从环境变量读,未设置时 SDK 不发送,本地开发不会误报。
traces_sample_rate=0.1只采样 10% 请求做性能追踪,控制用量。- 不设置
send_default_pii,SDK 使用保守的默认采集(不发送 IP、用户信息、Cookie);before_send再兜底过滤。
第 2 步:离线测试
tests/test_sentry.py:
import os, sys
import sentry_sdk
from sentry_sdk.transport import Transport
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import app as m
class Capture(Transport):
def __init__(self, options=None):
super().__init__(options); self.events = []
def capture_envelope(self, envelope):
for item in envelope.items:
if item.type == "event": self.events.append(item.payload.json)
def setup(monkeypatch):
monkeypatch.setenv("SENTRY_DSN", "https://public@o0.ingest.sentry.io/0") # 假 DSN,事件被 Capture 截获
t = Capture()
m.init_sentry(transport=t, traces_sample_rate=0)
m.app.config["PROPAGATE_EXCEPTIONS"] = False
return t, m.app.test_client()
def test_error_captured_and_scrubbed(monkeypatch):
t, c = setup(monkeypatch)
r = c.get("/orders/0", headers={"Cookie": "session=abc", "Authorization": "Bearer x"})
sentry_sdk.flush()
assert r.status_code == 500 and len(t.events) == 1
ev = t.events[0]
assert ev["exception"]["values"][0]["value"] == "订单不存在,联系电话 1**********"
assert ev["request"]["headers"].get("Cookie") in (None, "[已过滤]")
assert ev["request"]["headers"].get("Authorization") in (None, "[已过滤]")
assert ev["release"] == "shop@0.1.0" and ev["environment"] == "development"
def test_tag_on_checkout_error(monkeypatch):
t, c = setup(monkeypatch)
c.post("/checkout", json={"method": "alipay", "items": [{"price": 10}]})
sentry_sdk.flush()
assert t.events[0]["tags"]["payment"] == "alipay"
assert t.events[0]["exception"]["values"][0]["type"] == "KeyError"
def test_ok_no_event(monkeypatch):
t, c = setup(monkeypatch)
assert c.get("/orders/7").json == {"id": 7, "status": "paid"}
sentry_sdk.flush(); assert t.events == []
pip install -r requirements.txt pytest
python -m pytest -q tests
检查结果:3 passed。
第 3 步:连接真实项目(本文未执行)
export SENTRY_DSN="https://xxx@oNNN.ingest.sentry.io/NNN"
export APP_ENV=staging APP_RELEASE=shop@0.1.0
python app.py
curl -i http://127.0.0.1:5000/orders/0
预期结果:Sentry Issues 中出现 ValueError: 订单不存在,联系电话 1**********,release 为 shop@0.1.0,environment 为 staging。
第 4 步:告警
在 Sentry Alerts → Create Alert → Issues:条件选“新问题首次出现”,筛选 environment 等于 production,通知发到邮箱或团队工具。每次发版改 APP_RELEASE,就能看到哪个版本引入了新错误。
调试示例
第一版 init_sentry(**extra) 直接把 traces_sample_rate 写在 init 里又允许外部传入,测试时报错:
TypeError: sentry_sdk._init_implementation._init() got multiple values for keyword argument 'traces_sample_rate'
改为先组装 options 字典再 options.update(overrides)。
另一个:最初写了 send_default_pii=False,sentry-sdk 2.72.0 在测试中给出警告:
DeprecationWarning: The send_default_pii option is deprecated. Use data_collection instead.
官方文档说明该选项仍完全支持、计划在未来大版本中替换为 data_collection。本文去掉这行,直接使用默认值(效果相同)。注意:一旦传入 data_collection,未显式设置的类别会按更宽松的默认值采集,需要逐项关闭。
出错了怎么办?
| 现象 | 原因 | 解决 |
|---|---|---|
| 控制台没有事件 | DSN 没设置或写错 | echo $SENTRY_DSN 检查 |
| 本地调试时事件丢失 | 进程退出前没发出去 | 脚本结束前 sentry_sdk.flush() |
| 事件里有手机号、身份证号 | 没过滤 | 在 before_send 中补规则,并开启服务端数据清洗 |
| 用量很快用完 | 采样率太高或某个错误刷屏 | 降低 traces_sample_rate,修复高频错误或设置速率限制 |
| 看不到是哪个版本 | 没设置 release | 部署时设 APP_RELEASE |
模板规则
- DSN、环境、版本都从环境变量读。
- 默认不采集个人信息,
before_send再过滤一遍,并用测试锁住。 - 每次发版更新 release。
项目资源
| 用途 | 地址 |
|---|---|
| Flask 集成 | https://docs.sentry.io/platforms/python/integrations/flask/ |
| 过滤敏感数据 | https://docs.sentry.io/platforms/python/data-management/sensitive-data/ |
| Python SDK 采集的数据 | https://docs.sentry.io/platforms/python/data-management/data-collected/ |
| Python SDK 配置项 | https://docs.sentry.io/platforms/python/configuration/options/ |
| Sentry 价格 | https://sentry.io/pricing/ |
| Sentry 服务条款 | https://sentry.io/terms/ |
错误监控接进来只要几行代码,难的是“报什么、不报什么”。把隐私过滤写成函数并用测试验证,以后改代码也不会把用户信息悄悄发出去。