Aiokafkaconsumer über Fastapi aufzurufen "sollte in einer asynchronisierten Funktion erstellt werden oder direkt" Fehler ⇐ Python

[Guest]

Ich habe eine FastAPI -Anwendung, die sich mit dem Topie von Kafka unter Verwendung asynchronous code abonniert (d. H. Async /

Code: Select all

await

). Ich muss einen Unit-Test für meine Anwendung erstellen.

Code: Select all

def create_consumer() -> AIOKafkaConsumer:
"""Create AIOKafkaConsumer.

Returns:
AIOKafkaConsumer: The created AIOKafkaConsumer instance.
"""
return AIOKafkaConsumer(
settings.kafka_consumer_topic,
bootstrap_servers=f"{settings.kafka_consumer_host}:{settings.kafka_consumer_port}"
)

app = FastAPI()
consumer = create_consumer()

@app.on_event("startup")
async def startup_event():
"""Startup event for FastAPI application."""

log.info("Starting up...")
await consumer.start()
asyncio.create_task(consume())

async def consume(db: Session = next(get_db())):
"""Consume and print messages from Kafka."""
while True:
async for msg in consumer:
...

Ich verwende den TestClient von Fastapi :
import pytest

from fastapi.testclient import TestClient
from app.main import get_task_files
#from app.main import app

client = TestClient(app) # ERR: AIOKafkaConsumer The object should be created within an async function or provide loop directly.
< /code>
Ich erhalte jedoch den folgenden Fehler bei AIOKAFKACONSUMER < /code> Instatiation: < /p>

Das Objekt sollte erstellt werden In einer asynchronen Funktion oder direkt angeben. Es sieht so aus, als müsste ich Kafka-Funktionalität verspotten.