How to Poll an Operation Status Endpoint in Python
Mock a polling endpoint in Python that simulates checking an async operation's status until it completes or times out.
Python code
26 linesimport time
import random
def poll_status(url: str, timeout: float = 5.0) -> dict:
"""Mock a polling endpoint that eventually returns a completed status."""
start = time.time()
while time.time() - start < timeout:
# Simulate delayed response
time.sleep(0.2)
# 80% chance to report still processing, then return complete
if random.random() < 0.8:
continue
return {"url": url, "status": "completed", "attempts": attempt_count(start, timeout)}
return {"url": url, "status": "timeout", "attempts": attempt_count(start, timeout)}
def attempt_count(start: float, timeout: float) -> int:
"""Calculate total poll attempts from start time."""
return int((time.time() - start) / 0.2)
if __name__ == "__main__":
random.seed(42) # deterministic for example output
result = poll_status("https://api.example.com/jobs/123", timeout=2.0)
print(result)
Output
{'url': 'https://api.example.com/jobs/123', 'status': 'completed', 'attempts': 10}
How it works
The poll_status function uses a while loop with a deadline to repeatedly check a simulated endpoint. Each iteration sleeps 0.2 seconds and uses random.random() to decide if the operation is still processing or done. The attempt_count helper calculates the number of polls attempted by dividing the elapsed time by the interval. This pattern mimics real-world polling loops for asynchronous tasks and ensures the function doesn't hang indefinitely by enforcing a timeout. The random.seed(42) in the __main__ block makes the output reproducible for demos.
Common mistakes
- Using `time.sleep(0.2)` inside the loop without a timeout, which can cause infinite loops if the endpoint never completes.
- Not handling the case where the timeout expires, leading to a MissingFieldError when accessing 'status'.
- Assuming `random.random()` always returns a value less than 0.8, which can make the loop run longer than expected.
- Forgetting to convert elapsed time to an integer for attempt count, causing float artifacts in the output.
Variations
- Use `while time.monotonic() < start + timeout` to avoid issues with system clock adjustments.
- Replace the random decision with an actual HTTP call to a real endpoint using `requests.get()` and check the response status code.
Real-world use cases
- Polling a cloud API (e.g., AWS ECS task status) until it transitions from RUNNING to STOPPED.
- Checking the status of a long-running background job in a web app (e.g., video transcoding) until completion.
- Monitoring a CI/CD pipeline build status every few seconds until it passes or fails.
Sponsored
More from API design & gRPC
- Build a Bulk Array POST Mock Server in Python medium
- Build a Mock REST API with PUT and GET in Python medium
- Convert Protobuf to JSON and Dict in Python easy
- Create a Data Helper in Python for gRPC-style APIs easy
- Format data in Python using dataclasses like gRPC messages easy
- Generate an OpenAPI Spec from Mock Routes in Python easy
Keep learning
Related tutorials and quizzes for this topic.