Created
February 11, 2024 15:04
-
-
Save leobeeson/19ecfdc9a195a84f91790609f282f182 to your computer and use it in GitHub Desktop.
Async nested event loop for line-by-line execution of async requests.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
import nest_asyncio | |
# Apply nest_asyncio to enable nested event loop usage: | |
nest_asyncio.apply() | |
# Class used for the example: | |
extractor = EntityExtractor() | |
# Wrap the awaited line in an async function: | |
async def extract_entities(document_type, filepath, content): | |
return await extractor.extract_annotated_entities(document_type, filepath, content) | |
# Get the current event loop and run the coroutine in environments where an event loop is already running: | |
loop = asyncio.get_event_loop() | |
# Schedule the coroutine to run and wait for the result | |
document_type: str = "type_x", | |
filepath: str = "somewhere/there/data.csv | |
content: str = "some natural language text..." | |
extractions = loop.run_until_complete(extract_entities(document_type=document_type, filepath=filepath, content=content)) | |
# `extractions` inherits Pydantic `BaseModel`, so: | |
print(extractions.model_dump_json(indent=2)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment