FastAPI“状态”对象没有属性“实现”

Question

我在使用 FastAPI 测试代码时收到错误“State”对象没有属性“implementation”。

我的来源：

# tests/example_test.py

import pytest

from httpx import AsyncClient

from webserver import app

@pytest.mark.anyio
async def test_get_value():
    async with AsyncClient(app=app) as client:
        response = await client.get('/api/get_value')
    assert response.status_code == 200
    assert response.json() == {
        'value': 0,
    }

# webserver.py

from fastapi import FastAPI
from fastapi import Request

from httpx import AsyncClient

from contextlib import asynccontextmanager

from implementation import Implementation


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncClient(app=app) as client:
        implementation = Implementation()
        yield {'implementation': implementation}
        implementation.shutdown()

app = FastAPI(lifespan=lifespan)

@app.post('/api/increment')
async def api_increment(
    request: Request,
):
    implementation: Implementation = request.state.implementation
    implementation.increment()
    return {}

@app.get('/api/get_value')
async def api_get_value(
    request: Request,
):
    implementation: Implementation = request.state.implementation
    value = implementation.get_value()
    return {
        'value', value,
    }

# implementation.py

class Implementation():

    def __init__(self) -> None:
        self.current_value = 0
        self._initialized = True
        print(f'Implementation starts')

    def shutdown(self) -> None:
        '''
        A function which must be called to cleanup resources before exit
        '''
        self._initialized = False
        print(f'Implementation stops')

    def increment(self) -> None:
        '''
        In reality this would do something like read/write to a file, db etc
        '''
        self.current_value += 1

    def get_value(self) -> None:
        return self.current_value

如何修复？

Answer 1

当您使用

AsyncClient

时，它不会运行应用程序的生命周期功能。 您应该手动运行它（

async with app.router.lifespan_context(app):

）：

@pytest.mark.anyio
async def test_get_value():
    async with AsyncClient(app=app) as client:
        async with app.router.lifespan_context(app):
            response = await client.get('/api/get_value')
    assert response.status_code == 200
    assert response.json() == {
        'value': 0,
    }

您还可以使用

asgi-lifespan

库：https://pypi.org/project/asgi-lifespan/