# sa_token_python

**Repository Path**: fujc2dev/sa_token_python

## Basic Information

- **Project Name**: sa_token_python
- **Description**: Java 版 Sa-Token (https://sa-token.cc/)功能对等的 Python 版本，核心在于两点：架构设计上对标其“StpUtil 门面 + SaStorage 存储抽象”，技术选型上利用 Python 生态对应 Java 生态。
- **Primary Language**: Unknown
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-07-23
- **Last Updated**: 2026-09-20

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Sa-Token Python 版

> 一个高度模块化、工程化、功能对齐 Java 官方版本的 Python 权限认证框架。
> 采用分层架构、依赖注入、配置驱动，权限数据通过 `StpInterface` 回调业务数据源（零业务数据存储）。
> **核心包零 Web 依赖**，FastAPI / Flask 集成与 Redis / SQLite 存储以**独立扩展包**形式按需安装。

## 特性

- **完整对齐 Java 版**：登录、权限/角色校验、Session、踢人下线、账号封禁、注解式鉴权
- **StpInterface 回调模式**：业务方实现接口从自己的 DB 查权限/角色，框架零业务数据存储（对应 Java 版 `StpInterface`）
- **多账号体系**：通过 `StpLogic` 支持创建多个独立账号体系（如 `StpUtil` / `StpAdminUtil`）
- **路由拦截 DSL**：`SaRouter.match("/**").not_match("/login").check(...)` 链式配置
- **全局侦听器**：登录/登出/踢人/封禁等事件钩子，便于业务侧扩展
- **存储可插拔**：内置内存 `SaTokenDaoDefaultImpl`；生产环境安装 `satoken-redis` / `satoken-sqlite`，或自行实现 `SaTokenDao` 注入
- **Web 集成按需安装**：`satoken-web` 提供 FastAPI 中间件 + Depends、Flask before_request + g
- **类型安全**：完整类型注解 + 枚举类型，mypy 严格模式通过

## 包结构

核心包 `satoken` 只负责认证授权，Web 集成与持久化存储全部外置到仓库根目录 `extensions/` 下的独立包：

| 发行包名（pip install 用） | 导入包名（import 用） | 源码目录 | 作用 |
|:---------------------|:-----------------|:-----------------------|:------------------------------------------|
| `satoken`           | `satoken`        | `satoken/`             | 核心鉴权（零 Web 依赖、仅 `pydantic` 依赖）   |
| `satoken-web`       | `satoken_web`    | `extensions/web/`      | FastAPI / Flask 集成（中间件、Depends、`g`） |
| `satoken-redis`     | `satoken_redis`  | `extensions/redis/`    | Redis 持久层（多进程 / 多机部署）            |
| `satoken-sqlite`    | `satoken_sqlite` | `extensions/sqlite/`   | SQLite 持久层（单机持久化、开发测试）          |

> 三个扩展包各有独立的 `pyproject.toml`，**单独发版**；因位于顶层 `extensions/` 目录，
> 也不会随核心包 wheel 一起打包（根 `pyproject.toml` 的 `include = ["satoken*"]` 已天然排除）。

## 快速开始

### 安装

```bash
# 核心包（仅鉴权能力，零 Web 依赖）
pip install satoken

# FastAPI 集成
pip install "satoken-web[fastapi]"

# Flask 集成
pip install "satoken-web[flask]"

# 两个 Web 框架都要
pip install "satoken-web[all]"

# 可选：生产级持久化存储
pip install satoken-redis     # Redis
pip install satoken-sqlite    # SQLite
```

> 框架本身的开发者请看 [本地开发](#本地开发) 章节，扩展包需从源码目录可编辑安装。

### 快速体验（无需 Web 框架）

```bash
python examples/quickstart.py
```

该 demo 在控制台演示完整认证授权流程：实现 StpInterface → 登录 → 鉴权 → Session → 踢人 → 封禁。

### FastAPI 示例

```python
from fastapi import FastAPI, Depends
from satoken import StpUtil, StpInterface, SaRouter, SaTokenSettings
from satoken.decorators.auth import sa_check_login, sa_check_permission
from satoken_web.fastapi import SaTokenMiddleware, get_current_login_id


# 1. 业务方实现 StpInterface（从业务 DB 查权限/角色）
class AppStpInterface(StpInterface):
    def get_permission_list(self, login_id, login_type):
        return db.query("SELECT perm FROM user_perm WHERE uid=?", login_id)

    def get_role_list(self, login_id, login_type):
        return db.query("SELECT role FROM user_role WHERE uid=?", login_id)


# 2. 初始化配置 + 注册 StpInterface
StpUtil.configure(
    SaTokenSettings(),
    stp_interface=AppStpInterface(),
)

# 3. 创建应用并注册中间件
app = FastAPI()
app.add_middleware(SaTokenMiddleware, exclude_paths=["/login", "/public"])


@app.post("/login")
def login(username: str, password: str):
    if username == "admin" and password == "123":
        # 登录即可，权限由 StpInterface 实时回调
        token = StpUtil.login(login_id=10001, is_remember=True)
        return {"token": token}
    return {"msg": "账号或密码错误"}


@app.delete("/user/{user_id}")
@sa_check_login()
@sa_check_permission("user:delete")  # 自动回调 AppStpInterface
def delete_user(user_id: int):
    return {"success": True}
```

### Flask 示例

```python
from flask import Flask, request, g, jsonify
from satoken import StpUtil, StpInterface, SaTokenSettings
from satoken_web.flask import SaTokenExtension

class AppStpInterface(StpInterface):
    def get_permission_list(self, login_id, login_type):
        return db.query("SELECT perm FROM user_perm WHERE uid=?", login_id)
    def get_role_list(self, login_id, login_type):
        return db.query("SELECT role FROM user_role WHERE uid=?", login_id)

app = Flask(__name__)
StpUtil.configure(
    SaTokenSettings(),
    stp_interface=AppStpInterface(),
)
ext = SaTokenExtension(app, exclude_paths=["/login"])
```

## 核心 API 速查

| 业务场景        | 一句代码                                                                     |
|:------------|:-------------------------------------------------------------------------|
| 用户登录        | `token = StpUtil.login(user_id)`                                         |
| 登出          | `StpUtil.logout()`                                                       |
| 校验登录        | `StpUtil.check_login()`                                                  |
| 获取当前用户 ID   | `uid = StpUtil.get_login_id()`                                           |
| 校验权限        | `StpUtil.check_permission("order:delete")`（回调 StpInterface）               |
| 校验角色        | `StpUtil.check_role("admin")`（回调 StpInterface）                            |
| 踢人下线        | `StpUtil.kickout(user_id)`                                               |
| 封禁用户        | `StpUtil.disable(user_id, 86400)`                                        |
| Session 操作 | `StpUtil.set_session_value("k", "v")` / `StpUtil.get_session_value("k")` |
| 注册权限数据源     | `StpUtil.configure(settings, stp_interface=MyImpl())`                    |

> ⚠️ 权限/角色数据**不再**通过 `set_permission` / `set_role` 写入框架，改为实现 `StpInterface` 回调业务数据源。

## 枚举类型

框架所有配置级常量均以枚举集中定义在 `satoken.enums`，业务侧可一次性导入：

```python
from satoken import LoginType, TokenStyle, CookieSameSite, CheckMode
```

| 枚举               | 用途                 | 成员                                                                   |
|:-----------------|:-------------------|:---------------------------------------------------------------------|
| `LoginType`      | 账号体系类型（键空间隔离依据）    | `USER` / `ADMIN`（可继承扩展）                                              |
| `TokenStyle`     | Token 生成风格         | `UUID` / `RANDOM_32` / `SIMPLE_UUID`                                 |
| `CookieSameSite` | Cookie SameSite 策略 | `STRICT` / `LAX` / `NONE`                                            |
| `CheckMode`      | 多权限/角色校验模式         | `AND` / `OR`                                                         |
| `ReplacedLoginExitMode` | 顶替下线策略（`is_concurrent=False` 时生效） | `KICKOUT` / `INTERCEPT` |

> **注意**：枚举统一采用 `str, Enum` 模式（兼容 Python 3.10+）。在 f-string 拼接存储键等场景需显式 `.value` 取值，否则会输出
`LoginType.USER` 而非 `user`。

### 自定义账号体系

业务方通过继承 `LoginType` 扩展自定义账号类型：

```python
from satoken.enums import LoginType

class AppLoginType(LoginType):
    MEMBER = "member"
    API = "api"
```

## 多账号体系

```python
from satoken import StpUtil, StpAdminUtil, SaTokenSettings

_stp_interface = MyStpInterface()  # 业务方实现
StpUtil.configure(
    SaTokenSettings(),
    stp_interface=_stp_interface,
)         # login_type=LoginType.USER
StpAdminUtil.configure(
    SaTokenSettings(),
    stp_interface=_stp_interface,
)    # login_type=LoginType.ADMIN

StpUtil.login(10001)              # 普通用户登录
StpAdminUtil.login("root-01")     # 管理员登录

# 两套体系键空间隔离，互不影响
```

## StpInterface：权限/角色数据源（核心）

对应 Java 版 `cn.dev33.satoken.stp.StpInterface`。业务方实现本接口，框架在鉴权时实时回调获取数据：

```python
from satoken import StpInterface

class MyStpInterface(StpInterface):
    def get_permission_list(self, login_id, login_type):
        # login_type 可用于多账号体系区分查询逻辑
        # 实际项目从业务 DB 查询
        return {"user:read", "user:write"}

    def get_role_list(self, login_id, login_type):
        return {"admin"}

StpUtil.configure(
    SaTokenSettings(),
    stp_interface=MyStpInterface(),
)
```

**设计要点**：

- 框架**不存储**权限/角色数据，避免与业务数据源冗余
- 鉴权时实时回调，业务 DB 变更立即生效
- `login_type` 参数支持多账号体系区分查询逻辑

## 全局侦听器

```python
from satoken import SaTokenListener, StpUtil

class MyListener(SaTokenListener):
    def on_login(self, login_type, login_id, token):
        print(f"[{login_type}] 用户 {login_id} 登录，token={token}")

    def on_kickout(self, login_type, login_id, token):
        print(f"[{login_type}] 用户 {login_id} 被踢下线")

StpUtil.configure(listeners=...)  # 见文档详细说明
```

## 配置

所有配置通过 `SaTokenSettings`（pydantic-settings）管理，支持 `.env` 环境变量加载：

```python
from satoken import StpUtil, SaTokenSettings

# 方式一：使用默认配置（自动从环境变量加载）
StpUtil.configure()

# 方式二：手动传入配置
settings = SaTokenSettings(token_name="MyAppToken")
StpUtil.configure(settings, stp_interface=MyStpInterface())
```

完整环境变量见 [.env.example](.env.example)。

## 存储：从内存到持久化

`StpUtil.configure(storage=...)` 是唯一的存储注入入口，**不需要工厂或注册表**：

```python
StpUtil.configure(settings, storage=<SaTokenDao 实例>, stp_interface=...)
```

不传 `storage` 时使用内置内存实现 `SaTokenDaoDefaultImpl`（仅适用开发 / 单机进程内）。

### Redis 存储（satoken-redis）

```bash
pip install satoken-redis
```

```python
import redis
from satoken import StpUtil, SaTokenSettings
from satoken_redis import SaTokenDaoRedis

# 客户端由调用方构造与持有；decode_responses=True 为强制要求，否则构造时抛 ValueError
client = redis.from_url("redis://localhost:6379/0", decode_responses=True)

StpUtil.configure(
    SaTokenSettings(),
    storage=SaTokenDaoRedis(client),
    stp_interface=MyStpInterface(),
)
```

适用于多 worker / 多机部署，登录态全局共享：

```bash
gunicorn -w 4 app:app  # 4 个 worker 共享同一 Redis
```

### SQLite 存储（satoken-sqlite）

```bash
pip install satoken-sqlite
```

```python
import os
from sqlalchemy import create_engine
from satoken import StpUtil, SaTokenSettings
from satoken_sqlite import SaTokenDaoSQLite

# 推荐：基于绝对路径，避免工作目录漂移
db_path = os.path.join(os.path.dirname(__file__), "data", "satoken.db")
os.makedirs(os.path.dirname(db_path), exist_ok=True)

# 由调用方创建 Engine（跨线程需 check_same_thread=False），注入 DAO
engine = create_engine(
    f"sqlite:///{db_path}",
    connect_args={"check_same_thread": False},
)

StpUtil.configure(
    SaTokenSettings(),
    storage=SaTokenDaoSQLite(engine),
    stp_interface=MyStpInterface(),
)
```

> `SaTokenDaoSQLite` 只持有传入的 `Engine`，不创建、不关闭它。连接池与线程安全配置（如
> `check_same_thread`）由调用方在创建 `Engine` 时负责；表结构在构造时通过 `init_schema()`
> 幂等创建。

### 自定义其它存储（MongoDB / PostgreSQL 等）

实现 `satoken.storage.SaTokenDao` 抽象基类即可，覆盖三类数据结构与过期语义：

| 分组          | 需实现的方法                                                       |
|:------------|:-------------------------------------------------------------|
| string      | `get` / `set` / `update` / `delete`                          |
| 过期时间        | `get_timeout` / `expire` / `has_key`                         |
| set         | `set_data_set` / `get_data_set` / `in_set`                   |
| hash        | `hset` / `hget` / `hdel` / `hget_all`                        |
| 键扫描         | `search_data(prefix)`                                        |

超时约定：`-1` 永久、`-2` 键不存在、`>=0` 剩余秒数。
完整参考实现见 [extensions/redis/satoken_redis/dao.py](extensions/web_storager_redis/satoken_redis/dao.py)
与 [extensions/sqlite/satoken_sqlite/dao.py](extensions/web_storager_sqlite/satoken_sqlite/dao.py)。

## 异常处理（satoken-web）

FastAPI 中间件优先委托应用层 `@app.exception_handler(SaTokenException)`，未注册时使用默认兜底
`default_exception_handler`；Flask 同理。二者均可通过构造参数传入自定义 handler：

```python
from satoken_web.fastapi import SaTokenMiddleware
from satoken_web.flask import SaTokenExtension

app.add_middleware(SaTokenMiddleware, exclude_paths=["/login"], exception_handler=my_handler)
ext = SaTokenExtension(app, exclude_paths=["/login"], exception_handler=my_handler)
```

## 本地开发

扩展包不在 `satoken/` 包内，而是各自独立的 Python 项目，需**单独可编辑安装**才能被 import：

```bash
# 1. 核心包
pip install -e .

# 2. Web 集成（跑 tests/test_web 必装）
pip install -e "extensions/web[all]"

# 3. 存储扩展（按需）
pip install -e "extensions/redis"
pip install -e "extensions/sqlite"

# 4. 开发工具（保持核心包依赖精简，未写入 pyproject）
pip install pytest>=7.4 pytest-cov>=4.1 httpx>=0.24 black>=23.0 isort>=5.12 flake8>=6.0 mypy>=1.5 Flake8-pyproject>=1.2.3
```

> 更多开发工作流（跑测试、格式化、提交前检查）见 [examples/DEVELOPMENT.md](examples/DEVELOPMENT.md)
> （注意：该文档中的 `pip install -e ".[fastapi]"` 已随本次拆分失效，请改用上面的可扩展包安装方式）。

打包发布：

```bash
# 核心包
python -m build
# 扩展包（进入各自目录）
cd extensions/web && python -m build
```

## 静态检查

```bash
# 代码格式化与静态检查（行宽统一 88，工具配置统一在 pyproject.toml）
black satoken examples --check
isort satoken examples --check-only
flake8 satoken examples
mypy satoken
```

> 扩展包各自维护 `mypy` 配置（如 `extensions/web/pyproject.toml` 中忽略 `flask.*` / `starlette.*`，
> 因为它们有 Web 依赖）。

## 架构

```
satoken/               # 核心包（pip install satoken）：零 Web 依赖
├── core/          # 核心领域层（StpLogic / LoginService / StpInterface / SaRouter / Listener）
├── storage/       # 持久层抽象与默认实现（SaTokenDao / SaTokenDaoDefaultImpl）
├── facade/        # 门面 API（StpUtil / StpAdminUtil）
├── decorators/    # @SaCheckLogin / @SaCheckPermission / @SaCheckRole
├── utils/         # 请求解析工具
├── enums.py       # 枚举类型
├── settings.py    # SaTokenSettings（pydantic-settings）
└── exceptions.py  # SaTokenException 体系

extensions/            # 独立扩展包（各含 pyproject.toml，独立发版，不进核心 wheel）
├── web/           # 导入名 satoken_web    → pip install satoken-web[fastapi|flask|all]
├── redis/         # 导入名 satoken_redis  → pip install satoken-redis
└── sqlite/        # 导入名 satoken_sqlite → pip install satoken-sqlite
```

依赖方向单向无环：`storage → core → facade`，扩展包只依赖 `satoken`，核心包不反向依赖任何扩展。

## License

MIT