fly-common 0.1.3

轻量级 Python 公共工具库：JWT 认证、密码加密、单例模式、Redis 分布式锁与缓存

fly-common

轻量级 Python 公共工具库：JWT 认证、密码加密、单例模式、Redis 分布式锁与缓存、阿里云短信

📖 介绍

fly-common 是一个功能完善的 Python 公共库，主要提供以下核心功能：

• JWT Token 管理：支持对称加密（HS256）和非对称加密（RS256）的 JWT Token 生成、验证和刷新
• 密码加密：提供 MD5、PBKDF2 等多种加密方式
• 工具函数：提供单例模式、统一响应模型等实用工具
• Redis 分布式锁：面向生产场景的分布式锁，支持 fencing token、watchdog 自动续期、租约校验
• Redis 缓存：基于逻辑过期策略的缓存组件，支持异步刷新、防穿透、防雪崩
• 阿里云短信：封装阿里云短信服务，支持模板发送、自动重试、单例模式

✨ 功能特性

安全模块 (safety)

JWT Token

• 支持对称加密（HS256）和非对称加密（RS256）
• 提供 Access Token 和 Refresh Token 的生成
• Token 验证与刷新机制
• 支持自定义过期时间、签发者（issuer）和受众（audience）
• 单例模式实现，确保全局唯一实例

密码加密

• MD5 加密
• PBKDF2 加密（推荐用于密码存储）
• 密码验证功能
• 随机字符串生成
• 数字验证码生成

工具模块 (tools)

• 线程安全的单例模式装饰器
• 统一响应模型（Response）

服务模块 (service)

阿里云短信 (AliyunSMS)

• 支持模板短信发送
• 自动重试机制（指数退避 + 抖动）
• 单例模式，全局唯一实例
• 统一 Response 响应格式
• 详细的错误码映射和友好提示
• 10秒逻辑超时控制
• 业务限流直接返回，避免无效重试

Redis 分布式锁模块 (redis.lock)

• 基于 Lua 脚本的原子操作
• 支持 fencing token 防止旧写覆盖新写
• Watchdog 自动续期机制
• 租约校验与失效检测
• 支持阻塞/非阻塞模式
• 支持 namespace 隔离
• 提供 @synchronized 装饰器
• 支持 guarded_execute() 和 guarded_iterator() 受保护执行
• 生成结构化 SQL 辅助（execute_update() / execute_delete()）

Redis 缓存模块 (redis.cache)

• 基于逻辑过期策略（stale-while-revalidate）
• 异步刷新机制，避免缓存雪崩
• 空值缓存防止缓存穿透
• 支持热点数据缓存（hot_cached 装饰器）
• TTL 随机抖动防止集体过期
• 支持自定义序列化器
• 支持事件监控回调
• 提供预设配置（hot() / cold() / realtime()）

📦 安装

环境要求

• Python >= 3.9

安装方式

使用 uv（推荐）

# 安装核心功能
uv pip install fly-common

# 安装包含 Redis 支持的完整版本
uv pip install "fly-common[redis]"

# 安装包含阿里云短信支持的版本
uv pip install "fly-common[sms]"

# 安装完整功能（Redis + 短信）
uv pip install "fly-common[redis,sms]"

使用 pip

# 安装核心功能
pip install fly-common

# 安装包含 Redis 支持的完整版本
pip install "fly-common[redis]"

# 安装包含阿里云短信支持的版本
pip install "fly-common[sms]"

# 安装完整功能（Redis + 短信）
pip install "fly-common[redis,sms]"

依赖说明

核心依赖（自动安装）：

• python-jose - JWT Token 处理
• cryptography - 加密算法支持
• passlib - 密码加密
• pydantic - 数据验证

可选依赖：

• redis - Redis 分布式锁和缓存功能
• alibabacloud-dysmsapi20170525 - 阿里云短信服务功能

🚀 使用说明

JWT Token 使用示例

对称加密（HS256）

from fly_common.safety.jwt_token import JWTSymmetry

# 初始化（单例模式）
jwt_sym = JWTSymmetry(
    secret_key="your-secret-key",
    access_token_expire_seconds=86400,  # 24小时
    refresh_token_expire_seconds=604800,  # 7天
    issuer="your-app",
    audience="your-users"
)

# 生成 Token
result = jwt_sym.create_at_rf_token(payload={"sub": "user123"})
if result.ok:
    access_token = result.data["access_token"]
    refresh_token = result.data["refresh_token"]

# 验证 Token
verify_result = jwt_sym.verify_token(access_token)
if verify_result.ok:
    print("Token 有效:", verify_result.data)

# 刷新 Token
refresh_result = jwt_sym.refresh_token(access_token, refresh_token)
if refresh_result.ok:
    new_tokens = refresh_result.data

非对称加密（RS256）

from fly_common.safety.jwt_token import JWTAsymmetry

# 初始化（单例模式）
jwt_asym = JWTAsymmetry(
    private_key="your-private-key",
    public_key="your-public-key",
    algorithm="RS256"
)

# 使用方式与对称加密相同
result = jwt_asym.create_at_rf_token(payload={"sub": "user123"})

密码加密使用示例

from fly_common.safety.md5 import md5, en_password, check_password, code_number

# MD5 加密
hashed = md5("your-string")

# 密码加密（PBKDF2）
password_hash = en_password("user-password")

# 验证密码
is_valid = check_password("user-password", password_hash)

# 生成验证码
code = code_number(6)  # 生成6位数字验证码

单例模式使用示例

from fly_common.tools.single import Singleton

@Singleton
class MyService:
    def __init__(self):
        self.data = []

# 获取实例
service1 = MyService()
service2 = MyService()

# service1 和 service2 是同一个实例
assert service1 is service2

阿里云短信使用示例

基本使用

from fly_common.service.aliyun_sms import AliyunSMS

# 初始化（单例模式，相同参数返回同一实例）
sms = AliyunSMS(
    access_key_id="your-access-key-id",
    access_key_secret="your-access-key-secret",
    sign_name="你的签名",
)

# 发送短信
result = sms.send_sms(
    phone="13800138000",
    template_code="SMS_123456789",
    params={"code": "123456", "expire": "5"},
)

if result.ok:
    print(f"发送成功: {result.data}")
    # result.data 包含 request_id 和 biz_id
else:
    print(f"发送失败: {result.msg}")
    # result.data 包含 error_code 和 error_message

自定义签名发送

from fly_common.service.aliyun_sms import AliyunSMS

sms = AliyunSMS(
    access_key_id="your-access-key-id",
    access_key_secret="your-access-key-secret",
    sign_name="默认签名",
)

# 使用自定义签名（覆盖默认签名）
result = sms.send_sms(
    phone="13800138000",
    template_code="SMS_123456789",
    params={"code": "123456"},
    sign_name="自定义签名",
)

错误处理

from fly_common.service.aliyun_sms import AliyunSMS

sms = AliyunSMS(
    access_key_id="your-access-key-id",
    access_key_secret="your-access-key-secret",
    sign_name="你的签名",
)

result = sms.send_sms(
    phone="13800138000",
    template_code="SMS_123456789",
    params={"code": "123456"},
)

if not result.ok:
    error_code = result.data.get("error_code")
    
    # 常见错误码处理
    if error_code == "isv.BUSINESS_LIMIT_CONTROL":
        print("发送频率过高，请稍后再试")
    elif error_code == "isv.MOBILE_NUMBER_ILLEGAL":
        print("手机号格式错误")
    elif error_code == "isv.AMOUNT_NOT_ENOUGH":
        print("账户余额不足")
    elif error_code == "TOTAL_TIMEOUT":
        print("请求超时")
    elif error_code == "MAX_RETRIES_EXCEEDED":
        print("重试次数耗尽")
    else:
        print(f"未知错误: {result.msg}")

单例模式特性

from fly_common.service.aliyun_sms import AliyunSMS

# 相同参数返回同一实例
sms1 = AliyunSMS(
    access_key_id="key-id",
    access_key_secret="key-secret",
    sign_name="签名",
)

sms2 = AliyunSMS(
    access_key_id="key-id",
    access_key_secret="key-secret",
    sign_name="签名",
)

# sms1 和 sms2 是同一个实例
assert sms1 is sms2

结合验证码使用

from fly_common.safety.md5 import code_number
from fly_common.service.aliyun_sms import AliyunSMS

sms = AliyunSMS(
    access_key_id="your-access-key-id",
    access_key_secret="your-access-key-secret",
    sign_name="你的签名",
)

# 生成6位验证码
code = code_number(6)

# 发送验证码短信
result = sms.send_sms(
    phone="13800138000",
    template_code="SMS_123456789",  # 验证码模板
    params={"code": code, "expire": "5"},
)

if result.ok:
    # 将 code 存入 Redis 或数据库，设置5分钟过期
    print(f"验证码已发送: {code}")

Redis 分布式锁使用示例

基本使用

from redis import Redis
from fly_common.redis.lock import LockPreset, RedisLockSDK

redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
sdk = RedisLockSDK(redis_client, LockPreset.strict())

# 使用 with 语句获取锁
with sdk.lock("order:1001") as lock:
    # 执行业务逻辑
    print("处理订单")

受保护执行

from redis import Redis
from fly_common.redis.lock import LockPreset, RedisLockSDK

redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
sdk = RedisLockSDK(redis_client, LockPreset.strict())

def settle_order(order_id: str) -> None:
    print(f"结算订单: {order_id}")

with sdk.lock("order:1001") as lock:
    # 执行前后自动校验租约
    lock.guarded_execute(settle_order, "1001")

装饰器方式

from redis import Redis
from fly_common.redis.lock import LockPreset, RedisLockSDK
from fly_common.redis.lock.sdk import synchronized

redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
sdk = RedisLockSDK(redis_client, LockPreset.strict())

@synchronized(sdk, key_builder=lambda order_id: f"order:{order_id}")
def settle_order(order_id: str) -> None:
    print(f"结算订单: {order_id}")

settle_order("1001")

多业务域隔离

from redis import Redis
from fly_common.redis.lock import LockPreset, RedisLockSDK

redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
base_sdk = RedisLockSDK(redis_client, LockPreset.strict())

# 使用不同的 namespace 隔离业务域
order_sdk = base_sdk.with_namespace("order-service")
inventory_sdk = base_sdk.with_namespace("inventory-service")

with order_sdk.lock("order:1001") as lock:
    print("订单业务")

with inventory_sdk.lock("sku:1001") as lock:
    print("库存业务")

Redis 缓存使用示例

基本使用

from redis import Redis
from fly_common.redis.cache import RedisCache

redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
cache = RedisCache(redis_client)

# 缓存未命中时自动调用 db_func 查询数据库
def query_user(user_id):
    # 模拟数据库查询
    return {"id": user_id, "name": f"User {user_id}"}

data = cache.get("user:123", lambda: query_user(123))

使用预设配置

from redis import Redis
from fly_common.redis.cache import RedisCache, CachePreset

redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)

# 热点数据预设（长 TTL）
cache = RedisCache(redis_client, config=CachePreset.hot())

# 冷数据预设（短 TTL）
cache = RedisCache(redis_client, config=CachePreset.cold())

# 实时数据预设（极短 TTL）
cache = RedisCache(redis_client, config=CachePreset.realtime())

装饰器方式

from redis import Redis
from fly_common.redis.cache import RedisCache, cached, hot_cached

redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
cache = RedisCache(redis_client)

# 普通缓存装饰器
@cached(cache, lambda user_id: f"user:{user_id}", ttl=300)
def get_user(user_id):
    return db.query_user(user_id)

# 热点数据装饰器
@hot_cached(cache, lambda product_id: f"product:{product_id}")
def get_product(product_id):
    return db.query_product(product_id)

自定义配置

from redis import Redis
from fly_common.redis.cache import RedisCache, CacheConfig

config = CacheConfig(
    namespace="myapp",
    default_ttl=300,
    default_null_ttl=30,
    default_jitter=60,
)
redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
cache = RedisCache(redis_client, config=config)

监控事件

from redis import Redis
from fly_common.redis.cache import RedisCache, CacheConfig

def metrics_callback(event, payload):
    print(f"Event: {event}, Key: {payload.get('key')}")

config = CacheConfig(event_callback=metrics_callback)
redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=False)
cache = RedisCache(redis_client, config=config)

📁 项目结构

src/fly_common/
├── safety/                    # 安全模块
│   ├── jwt_token.py           # JWT Token 管理
│   └── md5.py                 # 密码加密工具
├── tools/                     # 工具模块
│   ├── single.py              # 单例模式
│   └── response.py            # 统一响应模型
├── service/                   # 服务模块
│   └── aliyun_sms.py          # 阿里云短信服务
└── redis/                     # Redis 模块
    ├── lock/                  # 分布式锁
    │   ├── core/              # 核心实现
    │   │   ├── lock.py        # 锁实现
    │   │   ├── config.py      # 配置与预设
    │   │   ├── watchdog.py    # 自动续期
    │   │   ├── fencing.py     # Fencing Token
    │   │   ├── events.py      # 事件系统
    │   │   └── errors.py      # 异常定义
    │   ├── sdk/               # 业务入口
    │   │   ├── manager.py     # SDK 管理器
    │   │   └── decorators.py  # 装饰器
    │   ├── examples/          # 使用示例
    │   └── docs/              # 详细文档
    └── cache/                 # 缓存组件
        ├── core/              # 核心实现
        │   ├── cache.py       # 缓存实现
        │   ├── config.py      # 配置与预设
        │   ├── serializer.py  # 序列化器
        │   └── errors.py      # 异常定义
        ├── sdk/               # 装饰器
        ├── examples/          # 使用示例
        └── docs/              # 详细文档

📚 详细文档

• Redis 分布式锁文档
• Redis 缓存文档

🤝 参与贡献

欢迎提交 Issue 和 Pull Request！

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

📄 许可证

本项目采用 MIT 许可证