anystack 0.1.5

面向云原生应用的 Python 资源抽象层，提供统一的异步协议包装数据库、消息队列、对象存储、KV、调度器等基础设施能力

Anystack

项目简介

Anystack 是一个面向云原生应用的 Python 资源抽象层。项目希望通过一组统一的异步协议，包装常见的基础设施能力（数据库、消息队列、对象存储、KV、调度器、实时通道、网关等），降低应用对具体厂商或产品的耦合度，让服务可以在不同环境之间自由迁移。

愿景

• 用统一接口使用各种资源
• 消除业务代码与特定基础设施的强绑定
• 让本地开发、云端部署、边缘环境具备一致的编程体验

核心特性

• 协议优先：src/anystack/protocol 定义面向数据库、KV、队列、调度器等的最小公共接口。
• 可插拔适配器：每类资源在 base.py 中给出抽象基类，再在子模块内提供具体的实施（如 SQLAlchemy、Cloudflare D1、Supabase 等）。
• 异步优先：所有适配器都以 async 接口暴露，方便在 FastAPI、Starlette 等异步框架中直接使用。
• 类型友好：广泛使用 Pydantic、类型提示和 Protocol，帮助 IDE 和静态检查理解接口契约。

项目结构

src/anystack/
    protocol/        # 统一协议定义
    db/              # 数据库适配器（SQLAlchemy、Cloudflare D1 等）
    storage/         # 基于 OpenDAL 的对象存储实现
    kv/              # KV 抽象与 Redis、Cloudflare KV、PostgreSQL 等实现
    queue/           # 消息队列（AsyncIO、Cloudflare Queue、PGMQ、DB Queue）
    scheduler/       # 调度器抽象与 Supabase 调度实现
    realtime/        # Realtime 通道（Supabase 等）
    gateway/         # Starlette 网关封装
    auth/            # JWT、OAuth 等鉴权能力
    migration/       # 数据库迁移工具

快速开始

环境要求

• Python 3.12+
• 异步运行时（asyncio）
• 具体适配器所需的第三方服务/凭证（如 Cloudflare、Supabase、PostgreSQL 等）

安装

本地安装推荐使用 uv：

uv pip install anystack

也可以使用标准 pip：

pip install anystack

最小示例

以下示例演示如何同时使用 SQLAlchemy 数据库适配器和 PostgreSQL PGMQ 队列：

import asyncio

from sqlalchemy.ext.asyncio import create_async_engine

from anystack.db.sqlalchemy import SQLAlchemyDB, Config as SAConfig
from anystack.queue.pgmq_queue import PGMQueue, Config as QueueConfig

async def main() -> None:
    engine = create_async_engine("postgresql+asyncpg://user:pass@localhost:5432/app")
    db = SQLAlchemyDB(SAConfig(engine=engine))

    queue = PGMQueue(QueueConfig(db=db, queue_name="jobs"))

    await queue.put({"task": "demo", "payload": {"foo": "bar"}})
    job = await queue.get()
    print(job)

    await db.close()

if __name__ == "__main__":
    asyncio.run(main())

要将应用暴露为 ASGI 服务，可以结合 Starlette 路由：

from starlette.responses import JSONResponse

from anystack.gateway.base import PathRoute, ListRouter
from anystack.gateway.starlette_impl import StarletteGateway

async def health(request):
    return JSONResponse({"status": "ok"})

router = ListRouter([
    PathRoute("/health", health, methods=["GET"]),
])

gateway = StarletteGateway(router=router)
app = gateway  # 直接作为 ASGI app 运行

资源适配器一览

数据库

适配器	描述	主要依赖
anystack.db.sqlalchemy.SQLAlchemyDB	通用 AsyncEngine 封装，支持事务、连接复用	sqlalchemy[asyncio]
anystack.db.d1.D1DB	通过 HTTP 与 Cloudflare D1 交互，模拟事务批量提交	httpx, Cloudflare API token

对象存储

适配器	描述	主要依赖
anystack.storage.opendal.OpenDALStorage	基于 OpenDAL 的统一对象存储访问，支持多种后端（S3、Supabase Storage、OSS 等）	opendal

KV 存储

适配器	描述	主要依赖
anystack.kv.redis.RedisKV	Redis JSON KV，支持键前缀、连接池	redis[asyncio]
anystack.kv.cloudflare_kv.CloudflareKV	调用 Cloudflare KV HTTP API	httpx, Cloudflare API token
anystack.kv.memory.MemoryKV	进程内内存实现，适合本地测试	无
anystack.kv.postgresql.PostgreSQLKV	使用关系型数据库持久化 KV	sqlalchemy, asyncpg

消息队列

适配器	描述	主要依赖
anystack.queue.asyncio_queue.AsyncioQueue	基于 asyncio.PriorityQueue 的轻量实现	内置
anystack.queue.cloudflare_queue.CloudflareQueue	Cloudflare Queues API 封装	httpx, Cloudflare API token
anystack.queue.pgmq_queue.PGMQueue	PostgreSQL PGMQ 扩展适配，支持优先级和 metrics	PostgreSQL, sqlalchemy, asyncpg
anystack.queue.db_queue.DBQueue	纯 SQL 的通用队列，支持优先级和自动建表	任意符合 DB 协议的适配器

调度器

适配器	描述	主要依赖
anystack.scheduler.single.supabase.SupabaseScheduler	依托 Supabase/PostgreSQL 的任务调度，支持定时、延迟、Cron	supabase, croniter
anystack.scheduler.single.worker.SingleWorker	单实例 worker，负责轮询执行调度任务	同上

实时系统

适配器	描述	主要依赖
anystack.realtime.supabase.SupabaseRealtime	封装 Supabase Realtime，支持 broadcast、db changes、presence	supabase

网关与应用容器

适配器	描述	主要依赖
anystack.gateway.base.PathRoute / ListRouter	纯 Python 路由和中间件装配	starlette (用于类型)
anystack.gateway.starlette_impl.StarletteGateway	将协议路由映射到 Starlette/ASGI	starlette, uvicorn(运行时)

鉴权

适配器	描述	主要依赖
anystack.auth.jwt_provider.JWTProvider	无状态 JWT 认证，支持自定义用户校验器	authlib
anystack.auth.oauth.github.GitHubOAuthProvider	GitHub OAuth 登录适配器，可与 JWTProvider 组合	authlib, GitHub OAuth App
anystack.auth.oauth.logto.LogtoOAuthProvider	Logto OIDC 登录适配器，提供通用用户信息映射	authlib, Logto Cloud 或自托管实例
anystack.auth.oauth.provider.OAuthAuthProvider	直接复用第三方 OAuth Access Token，不再额外颁发 JWT	authlib

数据库迁移

适配器	描述	主要依赖
anystack.migration.db.DBMigration	基于 SQL 文件的数据库迁移工具，支持版本管理和自动追踪	任意符合 DB 协议的适配器

示例：使用数据库迁移工具管理表结构变更：

from anystack.db.sqlalchemy import SQLAlchemyDB, Config as SAConfig
from anystack.migration.db import DBMigration
from sqlalchemy.ext.asyncio import create_async_engine

async def run_migrations():
    engine = create_async_engine("sqlite+aiosqlite:///app.db")
    db = SQLAlchemyDB(SAConfig(engine=engine))
    
    # 从目录加载 SQL 迁移文件，或直接传入 MigrationData 列表
    migration = DBMigration(db=db, migration="./migrations", table_name="migrations")
    
    # 执行所有待执行的迁移
    await migration.up()
    
    await db.close()

迁移工具会自动创建 migrations 表来追踪已执行的版本，只执行未执行过的迁移文件。SQL 文件按文件名排序执行。

---

示例：使用 GitHub OAuth 获取用户信息并颁发内部 JWT：

from starlette.responses import RedirectResponse, JSONResponse

from anystack.auth.jwt_provider import JWTConfig, JWTProvider
from anystack.auth.oauth.github import GitHubOAuthConfig, GitHubOAuthProvider
from anystack.auth.oauth.provider import OAuthAuthProvider

github = GitHubOAuthProvider(
    GitHubOAuthConfig(
        client_id="<client-id>",
        client_secret="<client-secret>",
        redirect_uri="https://example.com/auth/github/callback",
    )
)

jwt_provider = JWTProvider(JWTConfig(secret_key="change-me"))
oauth_provider = OAuthAuthProvider(github)

async def github_login(request):
    url, state = await github.get_authorize_url()
    request.session["github_state"] = state
    return RedirectResponse(url)

async def github_callback(request):
    code = request.query_params["code"]
    state = request.query_params.get("state")
    if state != request.session.pop("github_state", None):
        return JSONResponse({"detail": "state mismatch"}, status_code=400)

    result = await github.authenticate(code=code, state=state)

    if request.query_params.get("mode") == "oauth":
        token = oauth_provider.to_auth_token(result)
        user = result.user
        return JSONResponse({
            "mode": "oauth",
            "user": user.extra | {"id": user.id, "email": user.email},
            "github_token": {
                "access_token": token.access_token,
                "token_type": token.token_type,
            },
        })

    internal_token = await jwt_provider.login({"email": result.user.email, "user_id": result.user.id})
    return JSONResponse({
        "mode": "jwt",
        "user": result.user.extra | {"id": result.user.id, "email": result.user.email},
        "github_token": result.token.raw,
        "internal_token": internal_token.access_token,
    })

设计理念

1. Protocol 层 定义服务与资源的能力边界（如 DB, Queue, Storage）。
2. Base 层 在每个子模块内实现可复用的抽象基类，封装公共逻辑。
3. Adapter 层 将具体厂商或产品接入，保持最小依赖范围和一致行为。

这种分层让业务代码只依赖协议接口，替换具体实现时不需要改动逻辑；同时方便在测试环境注入内存实现或桩对象。

开发与测试

• 克隆仓库后，使用 uv pip install -e . 或 pip install -e . 安装依赖。
• 建议为需要的第三方服务准备本地或云端实例，例如 PostgreSQL、Redis、Supabase、Cloudflare 等。
• 目前仓库尚未附带统一的自动化测试；编写新适配器时，建议使用 pytest 并为协议契约增补测试用例。
• 完成后请运行自测脚本或示例代码，确保接口与协议保持兼容。

路线图

• 数据库适配器（SQLAlchemy、Cloudflare D1）
• 存储系统（OpenDAL）
• 认证系统（JWT Provider、GitHub OAuth）
• 消息队列（AsyncIO、Cloudflare Queue、PGMQ、DB Queue）
• 调度系统（Supabase 单实例调度器）
• 实时系统（Supabase Realtime）
• 路由系统 / API 网关（Starlette Gateway）
• 数据库迁移工具（DBMigration）
• 更多 OAuth 提供商（Google、Discord 等）
• 更多实时系统适配器
• 边缘函数 / 边缘执行环境
• 丰富网关中间件（限流、CORS、日志等）
• 完善测试覆盖

欢迎在 issues 或 PR 中讨论新的适配器需求与设计建议。

贡献

欢迎贡献代码、报告问题或提出建议！在提交 PR 之前，请确保：

1. 代码符合项目的类型提示和异步编程风格
2. 新增适配器遵循协议层定义
3. 添加必要的示例代码和文档说明
4. 保持依赖最小化，可选依赖应在文档中明确说明

许可证

本项目采用 MIT 许可证。详见 LICENSE 文件。