pgcr-fastapi 0.1.2
基于fastapi框架封装的，定制化的，更符合中国人习惯的python异步api框架
License: MIT
Requires Python: >=3.7,<4.0

# pgcr_fastapi

## 介绍

基于fastapi框架封装的，定制化的，更符合中国人习惯的python异步api框架

## 快速入门

### 路径参数

```python
from pgcr_fastapi import Api

app = Api()


@app.get("/")
async def root():
    return {"message": "Hello World"}


@app.get("/items/{item_id}")
async def read_item(item_id):
    return {"item_id": item_id}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 查询参数

```python
from pgcr_fastapi import Api

app = Api()

students = [{"name": f"张珊珊{i}"} for i in range(100)]


@app.get("/students/")
async def read_item(skip: int = 0, limit: int = 10):
    return students[skip: skip + limit]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 请求体（json）参数

```python
from pgcr_fastapi import (Api, Optional, BaseModel)

app = Api()


class Student(BaseModel):  # 学生模型
    name: str
    age: int
    gender: Optional[str] = "男"
    height: float
    weight: Optional[float] = None


@app.post("/students/")
async def student(student: Student):
    return f"创建学生：{student}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 查询参数校验

```python
from pgcr_fastapi import *

app = Api()


@app.get("/students/")
async def student(name: Optional[str] = Query(None, max_length=50, min_length=6, regex="^张")):
    return f"查询学生：{name}"


@app.get("/hobbies/")
async def student(hobbies: Optional[List[str]] = Query(["篮球", "羽毛球"])):
    return f"兴趣爱好：{hobbies}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 路径参数校验

```python
from pgcr_fastapi import *

app = Api()


# ... 表示必填参数
# gt： 大于
# ge： 大于等于
# lt: 小于
# le: 小于等于，支持浮点数
@app.get("/students/{id}")
async def student(id: Optional[int] = Path(..., ge=1, le=1000)):
    return f"查询学生：{id}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 多个请求体模型

```python
from pgcr_fastapi import *

app = Api()


class Student(BaseModel):  # 学生模型
    name: str
    age: int
    gender: Optional[str] = "男"
    height: float
    weight: Optional[float] = None


class Book(BaseModel):  # 图书模型
    title: str


@app.put("/test")
async def student(*, item: int = Body(...), student: Student, book: Book):
    return f"单个值：{item}，学生：{student}, 图书：{book}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 请求体字段校验

```python
from pgcr_fastapi import *

app = Api()


class Book(BaseModel):  # 图书模型
    title: str


class Student(BaseModel):  # 学生模型
    name: Optional[str] = Field(
        None, title="姓名", max_length=300, min_length=2
    )
    # 嵌套的模型
    tags: List[str] = []
    books: List[Book] = []


@app.put("/test")
async def student(*, item: int = Body(...), student: Student, book: Book):
    return f"单个值：{item}，学生：{student}, 图书：{book}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 使用嵌套的模型

```python
from pgcr_fastapi import *

app = Api()


class Book(BaseModel):  # 图书模型
    title: str


class Student(BaseModel):  # 学生模型
    name: Optional[str] = Field(
        None, title="姓名", max_length=300, min_length=2
    )
    # 嵌套模型
    tags1: List[str] = []
    tags2: Set[str] = set(["aaa", "bbb", "ccc"])
    tags3: Optional[Book] = None
    tags4: Optional[List[Book]] = []


@app.put("/test")
async def student(*, item: int = Body(...), student: Student, book: Book):
    return f"单个值：{item}，学生：{student}, 图书：{book}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 自定义示例参数

方法1：在Config内置类中添加

```python
from pgcr_fastapi import *

app = Api()


class Student(BaseModel):  # 学生模型
    name: str = None

    # 自定义示例参数
    class Config:
        schema_extra = {
            "example": {
                "name": "张三",
            }
        }


@app.put("/test")
async def student(student: Student):
    return f"学生：{student}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

方法2：在field中添加

```python
from pgcr_fastapi import *

app = Api()


class Student(BaseModel):  # 学生模型
    name: str = Field(..., example="filed的张三")
    description: Optional[str] = Field(None, example="field的自定义描述")
    price: float = Field(..., example=35.4)
    tax: Optional[float] = Field(None, example=3.2)


@app.put("/test")
async def student(student: Student):
    return f"学生：{student}"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 其他参数类型

```python
from pgcr_fastapi import *

app = Api()


@app.put("/items/{item_id}")
async def read_items(
        item_id: UUID,
        start_datetime: Optional[datetime] = Body(None),
        end_datetime: Optional[datetime] = Body(None),
        repeat_at: Optional[time] = Body(None),
        process_after: Optional[timedelta] = Body(None),
):
    start_process = start_datetime + process_after
    duration = end_datetime - start_process
    return {
        "item_id": item_id,
        "start_datetime": start_datetime,
        "end_datetime": end_datetime,
        "repeat_at": repeat_at,
        "process_after": process_after,
        "start_process": start_process,
        "duration": duration,
    }


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 接收cookie参数

```python
from pgcr_fastapi import *

app = Api()


@app.get("/items/")
async def read_items(test: Optional[str] = Cookie(None)):
    return {"test_cookie": test}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 请求体参数

```python
from pgcr_fastapi import *

app = Api()


@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

支持接收列表类型的请求头参数

```shell
@app.get("/items/")
async def read_items(x_token: Optional[List[str]] = Header(None)):
    return {"X-Token values": x_token}
```

### 输入模型和输出模型

```python
from pgcr_fastapi import *

app = Api()


class UserIn(BaseModel):  # 输入模型
    username: str
    password: str
    email: EmailStr
    full_name: Optional[str] = None


class UserOut(BaseModel):  # 输出模型
    username: str
    email: EmailStr
    full_name: Optional[str] = None


@app.post("/user/", response_model=UserOut)
async def create_user(user: UserIn):
    return user


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 响应状态码

```python
from pgcr_fastapi import *

app = Api()


@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
    return {"name": name}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 获取表单参数

```python
from pgcr_fastapi import *

app = Api()


@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username, "password": "******"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 上传文件

```python
from pgcr_fastapi import *

app = Api()


@app.post("/files/")
async def create_file(file: bytes = File(...)):
    return {"file_size": len(file)}


@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 多文件上传

```python
from pgcr_fastapi import *

app = Api()


@app.post("/files/")
async def create_files(files: List[bytes] = File(...)):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}


@app.get("/")
async def main():
    content = """
<body>
<form action="/files/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
<input name="files" type="file" multiple>
<input type="submit">
</form>
</body>
    """
    return HTMLResponse(content=content)


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

UploadFile

- 属性
    - filename
    - content_type
    - file
- 异步方法
    - write(data)
    - read(size)
    - seek(offset)
    - close()

### 抛出异常

```python
from pgcr_fastapi import *

app = Api()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 捕获异常

```python
from pgcr_fastapi import *

app = Api()


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )


@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 自定义接口描述

```python
from pgcr_fastapi import *

app = Api()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []


@app.post("/items/", response_model=Item, tags=["items"], summary="Create an item")
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 依赖注入

### 基本使用

```python
from pgcr_fastapi import *

app = Api()


async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 类作为依赖注入

```python
from pgcr_fastapi import *

app = Api()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")
async def read_items(commons=Depends(CommonQueryParams)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip: commons.skip + commons.limit]
    response.update({"items": items})
    return response


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 请求头依赖

```python
from pgcr_fastapi import *

app = Api()


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 全局依赖

```python
from pgcr_fastapi import *


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key


app = Api(dependencies=[Depends(verify_token), Depends(verify_key)])


@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 安全

### 基本使用

```python
from pgcr_fastapi import *

app = Api()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 获取当前用户

```python
from pgcr_fastapi import *

app = Api()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


def fake_decode_token(token):
    return User(
        username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
    )


async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    return user


@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 使用用户名和密码

```python
from pgcr_fastapi import *

app = Api()
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "fakehashedsecret",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Wonderson",
        "email": "alice@example.com",
        "hashed_password": "fakehashedsecret2",
        "disabled": True,
    },
}


def fake_hash_password(password: str):
    return "fakehashed" + password


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def fake_decode_token(token):
    # This doesn't provide any security at all
    # Check the next version
    user = get_user(fake_users_db, token)
    return user


async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user_dict = fake_users_db.get(form_data.username)
    if not user_dict:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    user = UserInDB(**user_dict)
    hashed_password = fake_hash_password(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    return {"access_token": user.username, "token_type": "bearer"}


@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 使用jwt

```python
from pgcr_fastapi import *

app = Api()
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Optional[str] = None


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    return [{"item_id": "Foo", "owner": current_user.username}]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 中间件

### 基本示例

```python
from pgcr_fastapi import *
import time

app = Api()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response


@app.get("/hello")
async def read_own_items():
    return "hello world"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 跨域中间件

```python
from pgcr_fastapi import *
import time

app = Api()

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/hello")
async def read_own_items():
    return "hello world"


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 路由器

### 基本使用

```python
from pgcr_fastapi import *
import time

app = Api()


@app.get("/hello")
async def read_own_items():
    return "hello world"


router = APIRouter()


@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]


app.include_router(router)

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 后台任务

### 基本使用

```python
from pgcr_fastapi import *

app = Api()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 文档

### 文档描述

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0", )

if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 响应

### 响应orjson

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0", )


@app.get("/items/", response_class=ORJSONResponse)
async def read_items():
    return [{"item_id": "Foo"}]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 响应HTML

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0", )


@app.get("/items/", response_class=HTMLResponse)
async def read_items():
    return """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 定制响应内容

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0", )


def generate_html_response():
    html_content = """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """
    return HTMLResponse(content=html_content, status_code=200)


@app.get("/items/", response_class=HTMLResponse)
async def read_items():
    return generate_html_response()


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 响应XML

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0", )


@app.get("/legacy/")
def get_legacy_data():
    data = """<?xml version="1.0"?>
    <shampoo>
    <Header>
        Apply shampoo here.
    </Header>
    <Body>
        You'll have to use soap here.
    </Body>
    </shampoo>
    """
    return Response(content=data, media_type="application/xml")


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 响应数据流

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0", )


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer())


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

示例2：响应mp4

```python
from pgcr_fastapi import *

some_file_path = "large-video-file.mp4"
app = Api()


@app.get("/")
def main():
    file_like = open(some_file_path, mode="rb")
    return StreamingResponse(file_like, media_type="video/mp4")
```

示例3：响应文件

```python
from pgcr_fastapi import *

some_file_path = "large-video-file.mp4"
app = Api()


@app.get("/")
async def main():
    return FileResponse(some_file_path)
```

### 设置默认响应类

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)


@app.get("/items/")
async def read_items():
    return [{"item_id": "Foo"}]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 自定义状态码响应内容

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)


class Item(BaseModel):
    id: str
    value: str


class Message(BaseModel):
    message: str


@app.get("/items/{item_id}", response_model=Item, responses={404: {"model": Message}})
async def read_item(item_id: str):
    if item_id == "foo":
        return {"id": "foo", "value": "there goes my hero"}
    else:
        return JSONResponse(status_code=404, content={"message": "Item not found"})


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 响应cookie

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)


@app.post("/cookie-and-object/")
def create_cookie(response: Response):
    response.set_cookie(key="fakesession", value="fake-cookie-session-value")
    return {"message": "Come to the dark side, we have cookies"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 自定义响应头

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)


@app.get("/headers-and-object/")
def get_headers(response: Response):
    response.headers["X-Cat-Dog"] = "alone in the world"
    return {"message": "Hello World"}


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 配置

### 基本使用

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str = None
    items_per_user: int = 50


settings = Settings()


@app.get("/info")
async def info():
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email,
        "items_per_user": settings.items_per_user,
    }


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 读取.env文件

```python
from pgcr_fastapi import *

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str = None
    items_per_user: int = 50

    class Config:
        env_file = ".env"


settings = Settings()


@app.get("/info")
async def info():
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email,
        "items_per_user": settings.items_per_user,
    }


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

### 使用缓存和依赖注入

config.py

```python
from pgcr_fastapi import BaseSettings


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str
    items_per_user: int = 50

    class Config:
        env_file = ".env"
```

example.py

```python
from pgcr_fastapi import *
import config

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)


@cache()  # 缓存让配置只创建一次
def get_settings():
    return config.Settings()


@app.get("/info")
async def info(settings: config.Settings = Depends(get_settings)):
    return {
        "app_name": settings.app_name,
        "admin_email": settings.admin_email,
        "items_per_user": settings.items_per_user,
    }


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```

## 生命周期

### 启动和关闭

```python
from pgcr_fastapi import *
import config

app = Api(title="My Super Project",
          description="This is a very fancy project, with auto docs for the API and everything",
          version="2.5.0",
          default_response_class=ORJSONResponse)
items = {}


@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]


@app.on_event("shutdown")
def shutdown_event():
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")


@app.get("/items/")
async def read_items():
    return [{"name": "Foo"}]


if __name__ == '__main__':
    import uvicorn

    uvicorn.run("example:app", reload=True)
```