コンテンツにスキップ

ライフスパン

Starletteアプリケーションは、アプリケーションの起動前またはアプリケーションの終了時に実行する必要のあるコードを処理するライフサイクルハンドラーを登録できます。

contextlibをインポート

Starlette. applicationからStarletteをインポート


@contextlib. asynccontextmanagerについて
async def lifespan(app):
    asyncとsome_async_Resource():
        print("起動時に実行!")
        利回り
        print("シャットダウン時に実行!")


ルート=[
    ...
]

app=Starlette(ルート=ルート、寿命=寿命)

Starletteは、ライフサイクルが実行されるまで、着信リクエストの処理を開始しません。

すべての接続が閉じられ、進行中のバックグラウンドタスクが完了すると、ランタイムは分解されます。

anyio.create_task_group() を使用して非同期タスクを管理することを検討してください。

寿命の状態

寿命には state の概念があります。これは、寿命と要求の間でオブジェクトを共有するために使用できる辞書です。

contextlibをインポート
from type ing import AsyncIterator, TypedDict

インポートhttpx
Starlette. applicationからStarletteをインポート
インポートリクエスト
starletteからPlainTextResponseをインポート
from starlette. routingインポートルート


class State(TypedDict):
    http_client: httpx.AsyncClient


@contextlib. asynccontextmanagerについて
async def lifespan(app: Starlette)->AsyncIterator[State]:
    async with httpx. AsyncClient()as client:
        利回り {"http_client": client}


async def home page(request: Request)->PlainTextResponse:
    クライアント=request. state.http_client
    response=await client. get("https://www.example.com")
    return PlainTextResponse(response. text)


アプリ=Starlette(
    lifespan=寿命,
    Route=[Route("/",ホームページ)]
)

リクエストで受信した state は、ライフハンドラーで受信した状態の浅いコピーです。

テストの実行寿命

呼び出しの寿命を確保するために、TestClient をコンテキストマネージャーとして使用する必要があります。

例:アプリのインポート
starlette. testclientからTestClientをインポート


def test_home page():
    TestClient(アプリ)をクライアントとして使用する:
        ブロックに入るとアプリケーションの寿命が呼び出されます。
        response=client. get("/")
        レスポンスをアサートします。ステータスコード==200

    そして、寿命のティアダウンはブロックを出るときに実行されます。