Metadata-Version: 2.4
Name: tieguanyin
Version: 1.0.1
Summary: TieGuanYin - A lightweight async web framework based on aiohttp.
Project-URL: Documentation, https://github.com/cqian-cs/tieguanyin#readme
Project-URL: Issues, https://github.com/cqian-cs/tieguanyin/issues
Project-URL: Source, https://github.com/cqian-cs/tieguanyin
Author-email: cqian-cs <cqian.cs@qq.com>
License-Expression: MIT
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: aiohttp
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP :: Dynamic Content
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Requires-Python: >=3.10
Requires-Dist: aiofiles>=24.0
Requires-Dist: aiohttp>=3.9
Requires-Dist: orjson>=3.9
Description-Content-Type: text/markdown

# TieGuanYin (铁观音) 🍵

**TieGuanYin** 是一个基于 [aiohttp](https://docs.aiohttp.org/en/stable/) 的轻量级异步 Web 框架。

## 🧭 框架定位

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



## ✨ 特性

### 1. 简单

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

### 2. 安全

```python
# ❌ 错误示例：路径注入尝试
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](https://github.com/ijl/orjson)。

| Framework      | QPS       |测试代码  |
| -------------- | --------- |--------- |
| aiohttp        | ~7200     |[code](benchmark/test_aiohttp.py)    |
| FastApi        | ~3100     |[code](benchmark/test_fastapi.py)    |
| Flask          | ~900      |[code](benchmark/test_flask.py)      |
| **TieGaunYin** | **~7600** |[code](benchmark/test_tieguanyin.py) |


详细测试代码见 [benchmark](benchmark/test.py)。


## 🚀 快速开始

### 1. 安装

```bash
pip install tieguanyin
```


### 2. 最小示例

```python
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 压缩
* 自动标注“源文件:行号”


**用法：**
```python
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 响应
```python
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'}`
* 保存到安全路径之外、原始文件后缀或另存文件后缀在安全后缀名之外，会拒绝保存。

```python
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: 如果成功，则为附件实际路径；如果失败，则为失败原因。
```

## 🧪 完整示例

```python
""" 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))
```
