tieguanyin 1.0.1

TieGuanYin - A lightweight async web framework based on aiohttp.

TieGuanYin (铁观音) 🍵

TieGuanYin 是一个基于 aiohttp 的轻量级异步 Web 框架。

🧭 框架定位

FastAPI	TieGuanYin
基于 Starlette 的成熟异步 Web 框架	基于 aiohttp 的轻量级异步实现
基于 Python logging + ASGI server 日志体系	内置日志系统（异步队列 / 轮转 / 压缩 / 溯源）
依赖注入机制（适合复杂工程）	单一上下文对象模式（session）
工程规范优先（自动文档 + 强类型约束）	灵活优先（动态类型）
功能体系完整	核心能力精简
适合团队与中大型系统	适合个人项目与轻量服务

✨ 特性

1. 简单

@app.get("/ping")
async def ping(session):
    return {"ok": True}

2. 安全

# ❌ 错误示例：路径注入尝试
username = '../../../../etc/passwd'
path_file = f"./fs/avatar/{username}"
await session.send_file(path_file, "./fs/avatar")
# ✗ "访问被拒绝: 文件不在安全范围内" （发送abort响应）

3. 高性能

• 测试场景：
  • JSON → 函数 → JSON
  • 并发：50
  • 持续：5 秒
• 结论：
  • TieGaunYin 的 QPS 约为 FastAPI 的 2.5 倍、Flask 的 8 倍。
  • TieGaunYin 比原版 aiohttp 更快，因为用了性能更好的 JSON 库 orjson。

Framework	QPS	测试代码
aiohttp	~7200	code
FastApi	~3100	code
Flask	~900	code
TieGaunYin	~7600	code

详细测试代码见 benchmark。

🚀 快速开始

1. 安装

pip install tieguanyin

2. 最小示例

import tieguanyin
import asyncio

app = tieguanyin.Server(debug=True)

@app.get("/ping")
async def ping(session):
    return {"status": "ok"}

if __name__ == "__main__":
    asyncio.run(app.start("127.0.0.1", 8080))

或者使用🧪 完整示例，体验文件上传/下载、Cookie管理、流式响应功能。

📖 核心概念

1. Session

session 是唯一上下文对象，用于获取请求参数、返回响应结果。

请求

• session.method: 获取请求方法（GET、POST）
• session.path: 获取请求路径（/ping）
• session.url_origin: 获取请求来源地址（http://127.0.0.1:8080）
• session.ip: 获取请求方IP
• session.query_params: 获取GET参数（?a=1&b=2）
• session.path_params: 获取路径参数（/blog/{id}）
• await session.post_params(): 获取POST参数
• session.post_type: 获取POST参数类型：json / multipart / form
• await session.json(): 获取POST JSON参数
• session.files.keys(): 获取POST附件字段列表
• session.files["k"]['filename']: 获取附件文件名
• session.files["k"]['path_file']: 获取附件路径
• session.save_file(safe_folder,field='k',path_file="xxx",allowed_suffixes={".jpg"}): 另存附件到指定目录

响应

• return dict → JSON
• await session.write(data:bytes) → 流式
• await session.go_to(url:str, status:int) → 重定向
• await session.abort(status:int, message:str) → 错误
• await session.send_file(path_file, safe_folder, download=True) → 发送文件（下载）
• await session.send_file(path_file, safe_folder, download=False) → 发送文件（预览）

Cookie 管理

• session.set_cookie("k", "v"): 设置Cookie
• session.get_cookie("k"): 获取Cookie
• session.delete_cookie("k"): 删除Cookie

2. 异步日志系统

特点：

• 同步调用，异步落盘，不阻塞。
• 自动轮转切片（东8时区每天凌晨0点自动切片）
• 自动 gzip 压缩
• 自动标注“源文件:行号”

用法：

app.logger.info("message")
session.logger.info("message")

输出： 时间 | 文件:行号 | 日志级别 | 消息

20231024-153045|main.py:42|[INFO ]| message

3. 内置安全机制

📌 请求体内存安全控制

• 创建Server对象时可通过max_size_in_MB指定请求体大小限制，默认 10 MB。
• session.post_params 内置应用层防护能力：
  • max_delimiters: 拦截[与{超过 1_000_000 次的JSON。
  • max_field_size: 拦截单个字段超过 640 KB的表单。
  • max_field_number：拦截字段个数超过 100_000 个的JSON与表单。
  • 当任一限制触发时，请求将被立即中止，并返回 abort 响应，防止进入后续业务处理阶段。

📦 文件安全暂存机制

• 附件流式写入暂存目录，避免内存占用，会话结束自动清理
  • 暂存目录默认为主函数文件目录下的 ./tmp 目录
  • 暂存附件在落盘时自动重命名（时间序列号），不保留原始文件后缀
  • 文件默认只读权限

🚫 文件访问安全控制

• 发送文件必须设定安全路径
• 安全路径之外的文件会被拒绝发送，并自动改为 abort 响应

suc, real_path = await session.send_file(
    safe_folder = safe_folder,
    path_file = f"{safe_folder}/{username}.jpg"
)
# suc: 是否成功发送文件
# real_path: 如果成功，则为文件实际路径；如果失败，则为失败原因。

🧷 文件上传安全校验

• 保存附件（就是把收到的附件移出暂存目录）必须设定安全路径，可选安全后缀名。
• 默认的安全后缀名是：{'.jpg', '.jpeg', '.png', '.gif', '.webp', '.pdf', '.txt', '.doc', '.docx', '.xls', '.xlsx', '.zip'}
• 保存到安全路径之外、原始文件后缀或另存文件后缀在安全后缀名之外，会拒绝保存。

suc, real_path = await session.save_file(
    safe_folder = safe_folder,
    path_file = f"{safe_folder}/{username}.jpg"
    allowed_suffixes={'.jpg', '.jpeg', '.png'}
)
# suc: 是否成功保存附件
# real_path: 如果成功，则为附件实际路径；如果失败，则为失败原因。

🧪 完整示例

""" main.py """
import tieguanyin
import asyncio
import os
path_this = os.path.dirname(os.path.abspath(__file__))

app = tieguanyin.Server(
    debug=True,                   # 开启调试
    cors=True, cors_origin="*",   # 允许跨域
    max_size_in_MB=200            # 最大请求体积，默认10MB，此处调整为200MB
)

# 演示1：普通API（返回JSON数据）
@app.get("/ping")
async def api_health(session):
    return {"status": "ok"}

# 演示2：静态文件API（读取URL参数，返回文件数据，记录日志）
@app.get("/{filename:.*}")
async def static(session):
    safe_folder = f"{path_this}/www"
    filename = session.path_params.get("filename") or 'index.html'
    suc, real_path = await session.send_file(f"{safe_folder}/{filename}", safe_folder, download=False)
    app.logger.info(f"{session.ip}|{'✓' if suc else '✗'}|GET|{filename}")

# 演示3：文件上传API（保存用户上传的文件到服务器）
@app.post("/fs/{filename:.*}")
async def fs_set(session):
    safe_folder = f"{path_this}/fs"
    filename = session.path_params.get("filename",'')
    params = await session.post_params() # 先读取post_params()才能保存文件
    suc, real_path = await session.file_save(
        safe_folder,
        #field="file", # 如果前端传入多个文件字段，此处可指定文件字段。默认为第一个文件字段。
        path_file=f"{safe_folder}/{filename}",
        allowed_suffixes={'.jpg', '.jpeg', '.png', '.gif', '.webp', '.pdf', '.txt', '.doc', '.docx', '.xls', '.xlsx', '.zip'}
    )
    app.logger.info(f"{session.ip}|{'✓' if suc else '✗'}|FS_SET|{filename}")
    if suc:
        return {"suc": True, "data": "file saved.", "url": f"{session.url_origin}/fs/{filename}"}
    return {"suc":False, "data": real_path}

@app.get("/fs/{filename:.*}")
async def fs_get(session):
    safe_folder = f"{path_this}/fs"
    filename = session.path_params.get("filename",'')
    suc, real_path = await session.send_file(f"{safe_folder}/{filename}", safe_folder, download=False)
    app.logger.info(f"{session.ip}|{'✓' if suc else '✗'}|FS_GET|{filename}")

# 演示4: 读写用户的cookie数据
@app.get("/login")
async def cookie_set(session):
    username = session.query_params.get("username")
    if not username:
        return {"suc": False, "data": f"Visit '{session.url_origin}/login?username=xxx' to login."}
    session.set_cookie("username", username, max_age=3600)
    return {"suc": True, "data": f"Logged in as {username}"}

@app.get("/whoami")
async def cookie_get(session):
    username = session.get_cookie("username", "guest")
    return {"suc": True, "data": username}

@app.get("/logout")
async def cookie_del(session):
    session.delete_cookie("username")
    return {"suc": True, "data": "Logged out"}

# 演示5: 流式响应
@app.get("/stream")
async def chat(session):
    await session.write(b'how ')
    await asyncio.sleep(1)
    await session.write(b'are ')
    await asyncio.sleep(1)
    await session.write(b'you?')

# 演示6: 回显全部GET参数
@app.get("/debug/{path:.*}")
async def params(session):
    path_params = session.path_params
    query_params = session.query_params
    query_params_str = "&".join(f"{k}={v}" for k,v in query_params.items())
    app.logger.info(f"{session.ip}|✓|DEBUG_GET|path_params={path_params}|query_params={query_params_str}")
    return {"suc": True, "data": {"path": path_params, "query": query_params_str}}

# 演示7: 回显全部POST参数
@app.post("/debug/{path:.*}")
async def params(session):
    path_params = session.path_params
    query_params = session.query_params
    post_params = await session.post_params()
    files = session.files
    query_params_str = "&".join(f"{k}={v}" for k,v in query_params.items())
    if session.post_type == 'json':
        post_params_str = repr(post_params)
    else:
        post_params_str = "&".join(f"{k}={v}" for k,v in post_params.items())
    files_str = "&".join(f"{k}={v['filename']}({os.path.getsize(v['path_file'])} Bytes)" for k,v in files.items())
    app.logger.info(f"{session.ip}|✓|DEBUG_POST|path: {path_params}|query: {query_params_str}|post: {post_params_str}|files: {files_str}")
    return {"suc": True, "data": {"path": path_params, "query": query_params_str, "post": post_params_str, "files": files_str}}


if __name__ == "__main__":
    asyncio.run(app.start(host="127.0.0.1", port=8080))