# gek-knowledge-python

**Repository Path**: gaoerkai/gek-knowledge-python

## Basic Information

- **Project Name**: gek-knowledge-python
- **Description**: No description available
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-11-19
- **Last Updated**: 2025-11-25

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 统一返回值与流式响应使用说明

- 全局已启用统一返回值中间件：对非流式且 `application/json` 的响应，自动包裹为 `{"code":0,"message":"success","data":...}`。
- 对流式响应（含 `text/event-stream`、任意 `more_body=True` 的分块传输、非 `application/json` 内容类型）中间件不拦截，保持原样直通。

## 非流式 JSON 返回（推荐：baseResponse）
- 使用统一响应模型：`response_model=baseResponse[YourDataModel]`，并返回 `UnifiedResponse.success(data)`。

```python
from fastapi import APIRouter
from pydantic import BaseModel
from app.core.response import UnifiedResponse, baseResponse

router = APIRouter()

class Hello(BaseModel):
    hello: str

@router.get("/example-non-stream", response_model=baseResponse[Hello])
def example_non_stream():
    return UnifiedResponse.success({"hello": "world"})
```

## SSE（Server-Sent Events）
- 使用 `UnifiedResponse.sse_stream(async_generator, wrap_response=True)` 返回事件流；当 `wrap_response=True` 时每条事件按 `{code,msg,data}` 包装。

```python
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from app.core.response import UnifiedResponse

router = APIRouter()

@router.get("/example-sse")
async def example_sse():
    async def gen():
        yield {"event":"start"}
        yield {"event":"step1"}
        yield {"event":"done"}
    return await UnifiedResponse.sse_stream(gen())
```

## HTTP 流式响应
- 使用 `UnifiedResponse.stream(generator, wrap_response=True, chunk_wrapper=None)` 返回普通流；当 `wrap_response=True` 时每块输出一行 `{code,msg,data}`。

```python
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from app.core.response import UnifiedResponse

router = APIRouter()

@router.get("/example-stream")
async def example_stream():
    async def gen():
        for i in range(2):
            yield f"chunk-{i}"
    return await UnifiedResponse.stream(gen())
```

- NDJSON 流推荐 `application/x-ndjson`，可通过 `wrap_response=False` 并自定义 `chunk_wrapper`：

```python
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from app.core.response import UnifiedResponse
import json

router = APIRouter()

@router.get("/example-ndjson")
async def example_ndjson():
    async def gen():
        yield {"i": 1}
        yield {"i": 2}
    def wrap(chunk):
        return json.dumps(chunk, ensure_ascii=False) + "\n"
    return await UnifiedResponse.stream(gen(), wrap_response=False, chunk_wrapper=wrap)
```

## 异常与错误返回
- 统一异常处理已注册：`HTTPException`、校验异常、未处理异常将返回形如 `{"code":<status>, "message":<detail>}` 的 JSON。
- 中间件检测到已含 `code` 与 `message` 的结构时不会二次包裹。

## 注意事项
- 非流式 JSON 建议通过 `baseResponse[T]` 明确 OpenAPI；返回时用 `UnifiedResponse.success(data)`，与中间件兼容。
- 流式场景请使用 `StreamingResponse` 并选择合适的 `media_type`（如 `text/event-stream`、`text/plain`、`application/x-ndjson`）。
- OpenAPI `response_model` 描述的是业务数据结构，实际响应外层存在统一包裹；如需文档同步可另行扩展。