import anyio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"
        await anyio.sleep(0)


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

///

## Attendre des messages et envoyer des messages { #await-for-messages-and-send-messages }

Dans votre route WebSocket, vous pouvez `await` des messages et envoyer des messages.

{* ../../docs_src/websockets_/tutorial001_py310.py hl[48:52] *}

Vous pouvez recevoir et envoyer des données binaires, texte et JSON.

        self.context_name = context_name

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        async with AsyncExitStack() as stack:
            scope[self.context_name] = stack

        configHelper.setLoadCallback(() -> {
            latch.countDown();
            return 1;
        });

        configHelper.update();

        assertTrue("Load method should be called within 5 seconds", latch.await(5, TimeUnit.SECONDS));
        assertEquals(1, configHelper.getLoadCallCount());
    }

    @Test
    public void test_update_multipleCallsIncrementLoadCount() throws InterruptedException {

                        },
                    }
                }
            },
            "required": True,
        },
    },
)
async def create_item(request: Request):
    raw_body = await request.body()
    data = magic_data_reader(raw_body)

### Файлы и async { #files-and-async }

В большинстве случаев «файлоподобные» объекты по умолчанию не совместимы с async и await.

Например, у них нет `await file.read()` или `async for chunk in file`.

И во многих случаях чтение таких объектов будет блокирующей операцией (которая может заблокировать цикл событий), потому что данные читаются с диска или из сети.

            AtomicInteger completedOps = new AtomicInteger(0);

            for (int t = 0; t < threadCount; t++) {
                executor.submit(() -> {
                    try {
                        startLatch.await(); // All threads start simultaneously
                        for (int i = 0; i < operationsPerThread; i++) {
                            byte[] localData = new byte[128];

{* ../../docs_src/async_tests/app_a_py310/test_main.py hl[7] *}

/// tip | 팁

`TestClient`를 사용할 때처럼 단순히 `def`가 아니라, 이제 테스트 함수가 `async def`라는 점에 주목하세요.

///

그 다음 앱으로 `AsyncClient`를 만들고, `await`를 사용해 비동기 요청을 보낼 수 있습니다.

{* ../../docs_src/async_tests/app_a_py310/test_main.py hl[9:12] *}

이는 다음과 동등합니다:

```Python
response = client.get('/')
```

`TestClient`로 요청을 보내기 위해 사용하던 코드입니다.

        "requestBody": {
            "content": {"application/x-yaml": {"schema": Item.model_json_schema()}},
            "required": True,
        },
    },
)
async def create_item(request: Request):
    raw_body = await request.body()
    try:
        data = yaml.safe_load(raw_body)
    except yaml.YAMLError:
        raise HTTPException(status_code=422, detail="Invalid YAML")
    try:
        item = Item.model_validate(data)

        with:
          github-token: ${{secrets.GITHUB_TOKEN}}
          script: |
            const script = require('./.github/workflows/create_issue.js')