Skip to content

popomore/sa-openapi

Repository files navigation

sa-openapi

神策分析(SensorsData Analytics)OpenAPI Python SDK 和 CLI 工具

Python Version License

特性

  • 🚀 完整的 API 覆盖:支持 Dashboard、Channel、Dataset、EventMeta、PropertyMeta、Model、SmartAlarm 七大服务,共 38 个 API 端点
  • 🔒 类型安全:基于 Pydantic v2 的完整类型提示和运行时验证
  • 同步/异步支持:同时提供同步和异步客户端
  • 🛠️ 强大的 CLI:命令行工具支持表格、JSON、CSV 多种输出格式
  • 📦 现代化设计:基于 aiohttp 和 Pydantic v2 构建
  • 🧪 高测试覆盖率:完整的模型测试覆盖

安装

作为 CLI 工具安装(推荐)

使用 uv 全局安装(推荐):

uv tool install sa-openapi

安装后可直接使用 sa-openapi 命令。

作为 Python 包安装

pip install sa-openapi
#
uv add sa-openapi

开发环境安装

git clone https://github.com/popomore/sa-openapi.git
cd sa-openapi
uv sync

快速开始

Python SDK

基础用法

from sa_openapi import SensorsAnalyticsClient

# 初始化客户端
client = SensorsAnalyticsClient(
    base_url="https://your-instance.sensorsdata.cn",
    api_key="sk-xxx",
    project="default",
)

# Dashboard 服务
navigations = client.dashboard.list_navigation(type="PRIVATE")
bookmarks = client.dashboard.list_bookmark(navigation_id=123)

# Channel 服务
links = client.channel.list_link(channel_id=456)

# Dataset 服务
result = client.dataset.sql_query(
    dataset_id=1,
    sql="SELECT * FROM events LIMIT 10"
)

# Model 服务 - 漏斗分析
funnel_result = client.model.funnel_report(
    measures=[...],
    filter={...},
    by_fields=[...]
)

异步客户端

from sa_openapi import AsyncSensorsAnalyticsClient

async def main():
    async with AsyncSensorsAnalyticsClient(
        base_url="https://your-instance.sensorsdata.cn",
        api_key="sk-xxx",
        project="default",
    ) as client:
        navigations = await client.dashboard.list_navigation(type="PRIVATE")
        print(navigations)

import asyncio
asyncio.run(main())

错误处理

from sa_openapi.exceptions import (
    SensorsAnalyticsError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
)

try:
    result = client.dataset.sql_query(dataset_id=999, sql="SELECT *")
except NotFoundError as e:
    print(f"资源未找到: {e.message}")
except ValidationError as e:
    print(f"参数验证失败: {e.message}")
except AuthenticationError as e:
    print(f"认证失败: {e.message}")
except SensorsAnalyticsError as e:
    print(f"API 错误: {e.code} - {e.message}")

CLI 工具

CLI 入口命令为 sa-openapi;在仓库开发环境里可以使用 uv run sa-openapi

首次使用先初始化配置:

sa-openapi config init
sa-openapi config show

常用命令示例:

# 查看帮助
sa-openapi --help

# Dashboard
sa-openapi dashboard list --type PRIVATE
sa-openapi dashboard bookmarks --navigation-id 123
sa-openapi dashboard bookmark-data 456 --start-date 2026-03-01 --end-date 2026-03-07

# Channel
sa-openapi channel list
sa-openapi channel get-link 789

# Dataset
sa-openapi dataset list
sa-openapi dataset sql-query --dataset-id 1 --sql "SELECT * FROM events LIMIT 10" --format csv

# Model
sa-openapi model funnel-report --json '{"measures":[{"event":"view_product","aggregator":"COUNT"}]}'
sa-openapi model sql --sql "SELECT * FROM events LIMIT 10" --format json

运行时可以通过全局选项临时覆盖配置:

sa-openapi --profile production --debug dashboard list
sa-openapi --base-url https://override.example.com --api-key sk-override dataset list

支持的环境变量:

export SA_BASE_URL="https://your-instance.sensorsdata.cn"
export SA_API_KEY="sk-xxx"
export SA_PROJECT="default"

详细 CLI 文档见 docs/CLI.md,其中包含:

  • 全部命令组与子命令
  • 参数与输出格式说明
  • profile 配置方式
  • model --json 参数示例
  • 常见问题与排错建议

API 参考

Dashboard 服务

方法 描述
list_navigation(type) 获取概览分组列表
get_navigation(navigation_id) 获取指定概览分组
list_bookmark(navigation_id) 获取概览书签列表
get_bookmark(bookmark_id) 获取指定概览书签
get_bookmark_data(bookmark_id, params) 获取概览书签数据
export_bookmark(bookmark_id, format) 导出概览书签

Channel 服务

方法 描述
list_channel() 获取渠道列表
list_link(channel_id) 获取渠道链接列表
get_link(link_id) 获取指定渠道链接
get_link_data(link_id, params) 获取渠道链接数据
export_link(link_id, format) 导出渠道链接数据

Dataset 服务

方法 描述
get_dataset_detail(dataset_id) 获取数据集详情
list_datasets(params) 获取数据集列表
list_dataset_groups() 获取数据集分组列表
sql_query(sql, query_parameters) 执行 SQL 查询
model_query(params) 模型查询(维度/度量)
refresh_dataset(dataset_id) 触发数据集刷新
get_sync_task_detail(sync_task_id) 查询同步任务状态

EventMeta 服务

方法 描述
list_events_all() 获取所有事件定义
list_event_tags() 获取事件标签列表

PropertyMeta 服务

方法 描述
list_all_event_properties() 获取所有事件属性
list_event_properties(events) 获取指定事件的属性
list_all_user_properties() 获取所有用户属性
list_user_groups() 获取用户分群列表
list_user_tags_with_dir() 获取用户标签树
get_property_values(table_type, property_name) 获取属性候选值

SmartAlarm 服务

方法 描述
get_alarm_config(config_id) 获取报警配置详情
list_alarms(params) 查询报警列表

Model 服务

方法 描述
segmentation_report(**params) 用户分群分析报告
funnel_report(**params) 漏斗分析报告
retention_report(**params) 留存分析报告
interval_report(**params) 间隔分析报告
addiction_report(**params) 黏性分析报告
user_property_report(**params) 用户属性分析报告
attribution_report(**params) 归因分析报告
ltv_report(**params) LTV 分析报告
session_report(**params) Session 分析报告
sql_query(sql, limit) 自定义 SQL 查询

架构设计

认证机制

所有 API 请求需要两个 HTTP 头:

  • api-key: 全局唯一的访问密钥
  • sensorsdata-project: 项目名称

这些在客户端初始化时配置,通过 httpx event hook 自动注入到每个请求中。

响应处理

所有 API 响应都包装在 HttpApiResult<T> 结构中:

{
  "code": "SUCCESS",
  "message": "操作成功",
  "request_id": "abc123",
  "data": { ... },
  "error_info": null
}

SDK 会自动解包 data 字段,当 code 不是 SUCCESS 时抛出相应异常。

双 Base Path

  • Dashboard/Channel/Dataset → /api/v3/analytics/v1
  • Model → /api/v3/analytics/v2

分页处理

对于支持分页的 API,SDK 提供迭代器自动处理翻页:

# 自动处理分页
for item in client.dataset.list_dataset_iter():
    print(item)

开发指南

环境设置

# 克隆仓库
git clone https://github.com/popomore/sa-openapi.git
cd sa-openapi

# 安装开发依赖(推荐使用 uv)
uv sync

# 或使用 pip
pip install -e ".[dev]"

运行测试

uv run pytest
# 带覆盖率报告
uv run pytest --cov=sa_openapi --cov-report=html

代码质量检查

# 全部检查(提交前运行)
uv run ruff check src/ tests/ && uv run ruff format --check src/ tests/ && uv run mypy src/ tests/ && uv run pytest

项目结构

sa-openapi/
├── docs/                          # OpenAPI 规范文件
├── src/sa_openapi/
│   ├── __init__.py                # 公开 API
│   ├── py.typed                   # PEP 561 类型标记
│   ├── client.py                  # 同步客户端
│   ├── async_client.py            # 异步客户端
│   ├── _base_client.py            # 共享基础逻辑
│   ├── _auth.py                   # 认证处理
│   ├── _config.py                 # 配置管理
│   ├── _transport.py              # HTTP 传输层
│   ├── _response.py               # 响应解包
│   ├── _pagination.py             # 分页处理
│   ├── _exceptions.py             # 异常定义
│   ├── models/                    # 数据模型
│   │   ├── common.py
│   │   ├── dashboard.py
│   │   ├── channel.py
│   │   ├── dataset.py
│   │   ├── event_meta.py
│   │   ├── property_meta.py
│   │   ├── smart_alarm.py
│   │   └── model.py
│   ├── services/                  # API 服务实现
│   │   ├── dashboard.py
│   │   ├── channel.py
│   │   ├── dataset.py
│   │   ├── event_meta.py
│   │   ├── property_meta.py
│   │   ├── smart_alarm.py
│   │   └── model.py
│   └── cli/                       # CLI 工具
│       ├── main.py
│       ├── config.py
│       ├── output.py
│       ├── dashboard.py
│       ├── channel.py
│       ├── dataset.py
│       ├── event_meta.py
│       ├── property_meta.py
│       ├── smart_alarm.py
│       └── model.py
├── tests/                         # 测试文件
├── pyproject.toml                 # 项目配置
└── README.md

贡献指南

欢迎贡献!请遵循以下步骤:

  1. Fork 本仓库
  2. 创建特性分支 (git checkout -b feature/amazing-feature)
  3. 提交更改 (git commit -m 'Add some amazing feature')
  4. 推送到分支 (git push origin feature/amazing-feature)
  5. 开启 Pull Request

代码规范

  • 使用 ruff 进行代码格式化和 linting
  • 使用 mypy 进行严格类型检查
  • 所有公开 API 必须有类型注解和文档字符串
  • 测试覆盖率不低于 90%

许可证

MIT License - 详见 LICENSE 文件

致谢

基于神策分析 OpenAPI 3.0.1 规范构建。

变更日志

0.1.0 (2024-03-02)

  • 🎉 初始版本发布
  • ✅ 支持 Dashboard、Channel、Dataset、Model 四大服务
  • ✅ 提供同步和异步客户端
  • ✅ 完整的 CLI 工具
  • ✅ 完整的类型提示支持

About

神策开放平台 sdk

Topics

Resources

License

Contributing

Stars

Watchers

Forks

Packages

 
 
 

Contributors

Languages