Ontology Operations
Wait for Ontology Upload
The wait_for_ontology_upload function polls the status of an ontology upload until it is completed or fails.
Synchronous
wait_for_ontology_upload
def wait_for_ontology_upload(
ontology_id: str,
polling_interval: float = 0.5,
timeout: int = 3600,
) -> Ontology:Waits for an ontology upload to complete.
Arguments
| Name | Type | Description | Default |
|---|---|---|---|
ontology_id | str | The ID of the ontology to wait for. | - |
polling_interval | float | The interval in seconds between status checks. | 0.5 |
timeout | int | The maximum time in seconds to wait. | 3600 |
Returns
| Type | Description |
|---|---|
Ontology | The completed Ontology object with its final status. |
Example
import perseus_client
# Assume you have just uploaded an ontology and have its ID
ontology_id = "your_uploaded_ontology_id"
try:
print("Waiting for ontology upload to complete...")
ontology = perseus_client.wait_for_ontology_upload(ontology_id)
print(f"Ontology processing finished with status: {ontology.status.value}")
except Exception as e:
print(f"An error occurred: {e}")Asynchronous
wait_for_ontology_upload_async
async def wait_for_ontology_upload_async(
ontology_id: str,
polling_interval: float = 0.5,
timeout: int = 3600,
) -> Ontology:Asynchronously waits for an ontology upload to complete.
Arguments
| Name | Type | Description | Default |
|---|---|---|---|
ontology_id | str | The ID of the ontology to wait for. | - |
polling_interval | float | The interval in seconds between status checks. | 0.5 |
timeout | int | The maximum time in seconds to wait. | 3600 |
Returns
| Type | Description |
|---|---|
Ontology | The completed Ontology object with its final status. |
Example
import asyncio
import perseus_client
# Assume you have just uploaded an ontology and have its ID
ontology_id = "your_uploaded_ontology_id"
async def main():
try:
print("Waiting for ontology upload to complete...")
ontology = await perseus_client.wait_for_ontology_upload_async(ontology_id)
print(f"Ontology processing finished with status: {ontology.status.value}")
except Exception as e:
print(f"An error occurred: {e}")
finally:
await perseus_client.close_async()
if __name__ == "__main__":
asyncio.run(main())