app = FastAPI()


@app.get("/items/")
def read_items(q: str | None = Param(default=None)):  # type: ignore
    return {"q": q}


client = TestClient(app)


def test_default_param_query_none():
    response = client.get("/items/")
    assert response.status_code == 200, response.text
    assert response.json() == {"q": None}


def test_default_param_query():
    response = client.get("/items/?q=foo")

app = FastAPI()


@app.post("/form/python-list")
def post_form_param_list(items: list = Form()):
    return items


@app.post("/form/python-set")
def post_form_param_set(items: set = Form()):
    return items


@app.post("/form/python-tuple")
def post_form_param_tuple(items: tuple = Form()):
    return items


client = TestClient(app)


def test_python_list_param_as_form():
    response = client.post(

async def stream_story_no_annotation():
    for line in message.splitlines():
        yield line


@app.get("/story/stream-no-async-no-annotation", response_class=StreamingResponse)
def stream_story_no_async_no_annotation():
    for line in message.splitlines():
        yield line


@app.get("/story/stream-bytes", response_class=StreamingResponse)
async def stream_story_bytes() -> AsyncIterable[bytes]:

    ],
)
def test_allow_inf_nan_param_default(value: str, code: int):
    response = client.post(f"/?z={value}")
    assert response.status_code == code, response.text


@pytest.mark.parametrize(
    "value,code",
    [
        ("-1", 200),
        ("inf", 422),
        ("-inf", 422),
        ("nan", 422),
        ("0", 200),
        ("342", 200),
    ],
)

app = FastAPI()

security = HTTPBase(scheme="Other")


@app.get("/users/me")
def read_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
    return {"scheme": credentials.scheme, "credentials": credentials.credentials}


client = TestClient(app)


def test_security_http_base():
    response = client.get("/users/me", headers={"Authorization": "Other foobar"})

    description="OAuth2PasswordBearer security scheme",
    auto_error=False,
)


@app.get("/items/")
async def read_items(token: str | None = Security(oauth2_scheme)):
    if token is None:
        return {"msg": "Create an account first"}
    return {"token": token}


client = TestClient(app)


def test_no_token():
    response = client.get("/items")
    assert response.status_code == 200, response.text

    @Inject
    abstract ExecOperations getExecOps()

    @TaskAction
    def transform() {
        logging.captureStandardOutput(LogLevel.INFO)
        logging.captureStandardError(LogLevel.INFO)

        def destDir = destinationDirectory.get().asFile
        fs.delete {
            delete(destDir, temporaryDir)
        }

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel


def test_pydanticv2():
    from pydantic import field_serializer

    class ModelWithDatetimeField(BaseModel):
        dt_field: datetime

        @field_serializer("dt_field")
        def serialize_datetime(self, dt_field: datetime):
            return dt_field.replace(microsecond=0, tzinfo=timezone.utc).isoformat()

### その他のユーティリティ関数 { #other-utility-functions }

あなたが直接呼び出すユーティリティ関数は、通常の `def` でも `async def` でも構いません。FastAPI はその呼び出し方に影響を与えません。

これは、FastAPI があなたの代わりに呼び出す関数 (すなわち *path operation 関数* と依存関係) とは対照的です。

ユーティリティ関数が `def` の通常関数であれば、(あなたのコードに書いたとおりに) 直接呼び出され、スレッドプールでは実行されません。関数が `async def` で作られている場合は、その関数を呼ぶときに `await` すべきです。

# Streaming

## Stream JSON Lines

To stream JSON Lines, declare the return type and use `yield` to return the data.

```python
@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        yield item
```

## Server-Sent Events (SSE)

To stream Server-Sent Events, use `response_class=EventSourceResponse` and `yield` items from the endpoint.