# Without aliases


@app.get("/optional-str")
async def read_optional_str(p: Annotated[str | None, Cookie()] = None):
    return {"p": p}


class CookieModelOptionalStr(BaseModel):
    p: str | None = None


@app.get("/model-optional-str")
async def read_model_optional_str(p: Annotated[CookieModelOptionalStr, Cookie()]):
    return {"p": p.p}

`UploadFile` 具有以下 `async` 方法。它们都会在底层调用对应的文件方法（使用内部的 `SpooledTemporaryFile`）。

* `write(data)`：将 `data` (`str` 或 `bytes`) 写入文件。
* `read(size)`：读取文件中 `size` (`int`) 个字节/字符。
* `seek(offset)`：移动到文件中字节位置 `offset` (`int`)。
    * 例如，`await myfile.seek(0)` 会移动到文件开头。
    * 如果你先运行过 `await myfile.read()`，然后需要再次读取内容时，这尤其有用。
* `close()`：关闭文件。

由于这些方法都是 `async` 方法，你需要对它们使用 await。

例如，在 `async` *路径操作函数* 内，你可以这样获取内容：

@app.get(
    "/a",
    response_class=JsonApiResponse,
    responses={500: {"description": "Error", "model": JsonApiError}},
)
async def a():
    pass  # pragma: no cover


@app.get("/b", responses={500: {"description": "Error", "model": Error}})
async def b():
    pass  # pragma: no cover


client = TestClient(app)


def test_openapi_schema():
    response = client.get("/openapi.json")

BrokenSessionDep = Annotated[Session, Depends(broken_dep_session)]

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, session: SessionDep):
    await websocket.accept()
    for item in session:
        await websocket.send_text(f"{item}")


@app.websocket("/ws-broken")
async def websocket_endpoint_broken(websocket: WebSocket, session: BrokenSessionDep):
    await websocket.accept()

送り返す各 JSON アイテムが `Item`（Pydantic モデル）型で、関数が async の場合、戻り値の型を `AsyncIterable[Item]` と宣言できます:

{* ../../docs_src/stream_json_lines/tutorial001_py310.py ln[1:24] hl[9:11,22] *}

戻り値の型を宣言すると、FastAPI はそれを使ってデータを**検証**し、OpenAPI に**ドキュメント化**し、**フィルター**し、Pydantic で**シリアライズ**します。

/// tip | 豆知識

Pydantic は **Rust** 側でシリアライズを行うため、戻り値の型を宣言しない場合に比べて大幅に高い**パフォーマンス**が得られます。

///

        return HTTPException(
            status_code=HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers=self.make_authenticate_headers(),
        )

    async def __call__(self, request: Request) -> HTTPAuthorizationCredentials | None:
        authorization = request.headers.get("Authorization")
        scheme, credentials = get_authorization_scheme_param(authorization)

// TODO(b/154564140): Move this to its own header. This requires splitting
// c_api_experimental.h
struct TFE_ContextOptions {
  TF_SessionOptions session_options;
  // true if async execution is enabled.
  bool async = false;
  TFE_ContextDevicePlacementPolicy device_placement_policy{
      TFE_DEVICE_PLACEMENT_SILENT};
  // If true, use TFRT backend
  bool use_tfrt = false;

                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


@app.get("/")
async def get():
    return HTMLResponse(html)


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()

@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: str = Header()):
    if x_token != fake_secret_token:
        raise HTTPException(status_code=400, detail="Invalid X-Token header")
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]


@app.post("/items/")
async def create_item(item: Item, x_token: str = Header()) -> Item:

/// tip | 提示

由于 Pydantic 会在**Rust** 端序列化它，相比未声明返回类型，你将获得更高的**性能**。

///

### 非 async 的*路径操作函数* { #non-async-path-operation-functions }

你也可以使用常规的 `def` 函数（没有 `async`），并以同样的方式使用 `yield`。

FastAPI 会确保其正确运行，从而不阻塞事件循环。

由于此时函数不是 async，正确的返回类型应为 `Iterable[Item]`：

{* ../../docs_src/server_sent_events/tutorial001_py310.py ln[28:31] hl[29] *}

### 无返回类型 { #no-return-type }