Erstellen einer anhaltenden Hintergrundaufgabe in Python ⇐ Python

[Anonymous]

Ich möchte im Hintergrund einige Funktionen ausführen, aber nach Belieben herunterfahren. Ich habe eine Reihe von Lösungen ausprobiert, von denen das offensichtlichste Asyncio Aufgaben ist:

Code: Select all

# Data source
DataSource = Generator[int, Any, None]
def create_data_source() -> DataSource
...

# Data processing - should be able to run in the background until being shut down
async def data_processor(source: DataSource) -> None:
try:
for item in source:
print("Working")

except Exception as e:
print("Exception", e)
raise

finally:
print("Cancelled")

async def main():
# Run the function
data_source = create_data_source()
processing_task = asyncio.create_task(data_processor(data_source))

# Do other stuff
await asyncio.sleep(3)

# Shut the function down
processing_task.cancel()

asyncio.run(main())
< /code>
Das Problem ist - Sobald die Aufgabe beginnt, wird die Schleife "Arbeiten" nie endet. Ich habe versucht, die Schleife async zu machen, ich habe versucht, es mit Asyncio.event 

ein Shutdown -Signal zu senden, und ich habe sogar versucht, die Funktion mit Concurrent.Futures neu zu schreiben, aber nichts funktioniert. Was fehlt mir?
Dies wird aus Gründen von Python 3.11 ausgeführt, aber wenn neuere Versionen die Lösung haben, kann ich möglicherweise aktualisieren.>