Skip to content

Commit f21c05e

Browse files
wasin-waeosriwasin-waeosri
committed
tried asyncio.TaskGroup()
1 parent 90cd211 commit f21c05e

2 files changed

Lines changed: 138 additions & 41 deletions

File tree

‎src/async_learn.py‎

Lines changed: 97 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -1,40 +1,104 @@
11
import asyncio
2-
import random
3-
import time
2+
import time
43

5-
async def main():
6-
user_ids = [1, 2, 3]
4+
5+
async def task_function(name, duration, fail=False):
6+
print(f" Task {name}: Starting (will sleep for {duration}s)")
7+
await asyncio.sleep(duration)
8+
if fail:
9+
raise ValueError(f"Task {name} failed!")
10+
print(f" Task {name}: Finished")
11+
return f"Result of {name}"
12+
13+
14+
# ─────────────────────────────────────────────
15+
# asyncio.gather
16+
# ─────────────────────────────────────────────
17+
# • Available since Python 3.4
18+
# • Accepts coroutines or Tasks
19+
# • Returns a list of results in the same order as inputs
20+
# • Error behaviour (default): if one task raises, the exception is
21+
# propagated immediately to the caller, but the OTHER tasks keep
22+
# running in the background (they are NOT cancelled).
23+
# • return_exceptions=True: exceptions are returned as result values
24+
# instead of being raised, so all tasks always run to completion.
25+
# • Good for fire-and-forget fan-out or when you need return_exceptions.
26+
27+
async def demo_gather():
28+
print("\n=== asyncio.gather (no failures) ===")
729
start = time.perf_counter()
8-
await asyncio.gather(
9-
*(get_user_with_posts(user_id) for user_id in user_ids), return_exceptions=True
30+
results = await asyncio.gather(
31+
task_function("A", 3),
32+
task_function("B", 1),
33+
task_function("C", 2),
1034
)
11-
end = time.perf_counter()
12-
print(f"\n==> Total time: {end - start:.2f} seconds")
13-
14-
async def get_user_with_posts(user_id):
15-
user = await fetch_user(user_id)
16-
await fetch_posts(user)
17-
18-
async def fetch_user(user_id):
19-
delay = random.uniform(0.5,2.0)
20-
print(f"User coro: fetching user by {user_id}")
21-
await asyncio.sleep(delay)
22-
user = {"id": user_id, "name": f"User{user_id}"}
23-
print(f"User coro: fetched user with {user_id} (done in {delay:.2f} seconds)")
24-
return user
25-
26-
async def fetch_posts(user):
27-
delay = random.uniform(0.5,2.0)
28-
print(f"Posts coro: retrieving posts for {user['name']}")
29-
await asyncio.sleep(delay)
30-
posts = [f"Post {i} by {user['name']}" for i in range(1, 3)]
31-
print(
32-
f"Post coro: got {len(posts)} posts by {user['name']}"
33-
f" (done in {delay:.1f} seconds)"
35+
print(f" Done in {time.perf_counter() - start:.2f}s | Results: {results}")
36+
37+
print("\n=== asyncio.gather with return_exceptions=True ===")
38+
start = time.perf_counter()
39+
results = await asyncio.gather(
40+
task_function("A", 1),
41+
task_function("B", 0.5, fail=True), # will raise
42+
task_function("C", 1),
43+
return_exceptions=True, # exceptions become values
3444
)
35-
for post in posts:
36-
print(f" - {post}")
45+
print(f" Done in {time.perf_counter() - start:.2f}s | Results: {results}")
46+
47+
48+
# ─────────────────────────────────────────────
49+
# asyncio.TaskGroup (Python 3.11+)
50+
# ─────────────────────────────────────────────
51+
# • Context-manager-based structured concurrency.
52+
# • Tasks are created with tg.create_task() inside the `async with` block.
53+
# • When the block exits all tasks are AWAITED automatically.
54+
# • Error behaviour: if ANY task raises, ALL remaining tasks are
55+
# cancelled immediately, then the exception is re-raised.
56+
# Multiple simultaneous exceptions are collected into an ExceptionGroup.
57+
# • Cleaner resource management; preferred for structured concurrency.
58+
59+
async def demo_task_group():
60+
print("\n=== asyncio.TaskGroup (no failures) ===")
61+
start = time.perf_counter()
62+
results = []
63+
async with asyncio.TaskGroup() as tg:
64+
t1 = tg.create_task(task_function("A", 3))
65+
t2 = tg.create_task(task_function("B", 1))
66+
t3 = tg.create_task(task_function("C", 2))
67+
# All tasks are guaranteed finished here
68+
results = [t1.result(), t2.result(), t3.result()]
69+
print(f" Done in {time.perf_counter() - start:.2f}s | Results: {results}")
70+
71+
print("\n=== asyncio.TaskGroup with a failing task ===")
72+
try:
73+
async with asyncio.TaskGroup() as tg:
74+
tg.create_task(task_function("A", 1))
75+
tg.create_task(task_function("B", 0.1, fail=True)) # raises quickly
76+
tg.create_task(task_function("C", 1)) # will be cancelled
77+
except* ValueError as eg:
78+
print(f" Caught ExceptionGroup errors: {eg}")
79+
80+
81+
# ─────────────────────────────────────────────
82+
# Summary
83+
# ─────────────────────────────────────────────
84+
# | Feature | gather | TaskGroup (3.11+) |
85+
# |--------------------------|-------------------------|-------------------------|
86+
# | API style | function call | async context manager |
87+
# | Task creation | pass coroutines/tasks | tg.create_task() |
88+
# | On first exception | others keep running * | others are cancelled |
89+
# | Multiple exceptions | only first propagated | ExceptionGroup |
90+
# | return_exceptions | yes | no (use try/except) |
91+
# | Structured concurrency | no | yes |
92+
# | Python version | 3.4+ | 3.11+ |
93+
#
94+
# * Unless return_exceptions=False (default) and you shield tasks.
95+
96+
async def main():
97+
await demo_gather()
98+
await demo_task_group()
99+
37100

38101
if __name__ == "__main__":
39-
random.seed(42)
40-
asyncio.run(main())
102+
asyncio.run(main())
103+
104+

‎src/example_async.py‎

Lines changed: 41 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
AUTH_TOKEN_URL = "/auth/oauth2/v1/token"
99
CHAIN_URL = "/data/pricing/chains/v1/"
1010

11-
RICS = ["EFX=", "0#.SPX", "0#.FTSE"] # Fetched concurrently
11+
RICS = ["EFX=", "0#.HSI", "0#.FTSE","0#.SPX"] # Fetched concurrently
1212

1313
async def post_authentication(machine_id, password, app_key, url, client):
1414
"""Authenticate to RDP and return the token response as JSON."""
@@ -33,7 +33,7 @@ async def post_authentication(machine_id, password, app_key, url, client):
3333
response_auth.raise_for_status() # Raise for 4xx/5xx API failures.
3434
return response_auth.json()
3535

36-
async def get_chain(ric, token, url, client):
36+
async def get_chain(ric, token, url, client, semaphore=None):
3737
"""Fetch chain data for a single RIC symbol using an access token."""
3838
headers = {
3939
"Authorization": f"Bearer {token}",
@@ -45,7 +45,12 @@ async def get_chain(ric, token, url, client):
4545
}
4646

4747
# Request chain data from the pricing chains endpoint.
48-
response_chain = await client.get(url, params=parameters, headers=headers)
48+
if semaphore:
49+
async with semaphore:
50+
response_chain = await client.get(url, params=parameters, headers=headers)
51+
else:
52+
response_chain = await client.get(url, params=parameters, headers=headers)
53+
4954
response_chain.raise_for_status()
5055
return response_chain.json()
5156

@@ -94,11 +99,39 @@ async def main():
9499
# else:
95100
# print(f"Chain data for '{ric}': {result}")
96101

97-
ric = "EFX="
98-
print(f"Fetching chain data... for RIC: {ric}")
99-
chain_data = await get_chain(ric, access_token, CHAIN_URL, client)
100-
print("Chain data retrieved successfully!")
101-
print(f"Chain data for {ric}:", json.dumps(chain_data, indent=2))
102+
# --- Fetch all RICs concurrently ---
103+
# TaskGroup cancels all remaining tasks as soon as one raises.
104+
# Exceptions are collected into an ExceptionGroup — use `except*` to handle them.
105+
MAX_CONCURRENT_TASKS = 3
106+
sem = asyncio.Semaphore(MAX_CONCURRENT_TASKS) # Limit concurrent tasks
107+
sem = None # No concurrency limit
108+
try:
109+
async with asyncio.TaskGroup() as tg:
110+
tasks = {ric: tg.create_task(get_chain(ric, access_token, CHAIN_URL, client, sem)) for ric in RICS}
111+
112+
# Reached only if ALL tasks succeeded.
113+
results = {ric: task.result() for ric, task in tasks.items()}
114+
print("Chain data for all RICs:", json.dumps(results, indent=2))
115+
116+
except* httpx.HTTPStatusError as eg:
117+
for exc in eg.exceptions:
118+
print(f"HTTP error fetching chain data: {exc.request.url} -> {exc.response.status_code}: {exc.response.text}")
119+
except* httpx.HTTPError as eg:
120+
for exc in eg.exceptions:
121+
print(f"Network error fetching chain data: {exc}")
122+
123+
124+
# try:
125+
# ric = "EFX="
126+
# ric = "0#.FTSE"
127+
# print(f"Fetching chain data... for RIC: {ric}")
128+
# chain_data = await get_chain(ric, access_token, CHAIN_URL, client)
129+
# print("Chain data retrieved successfully!")
130+
# print(f"Chain data for {ric}:", json.dumps(chain_data, indent=2))
131+
# except httpx.HTTPStatusError as e:
132+
# print(f"HTTP error fetching chain data:{e.request.url}: {e.response.status_code} - {e.response.text}")
133+
# except httpx.HTTPError as e:
134+
# print(f"Network error fetching chain data:{e.request.url}: {e}")
102135

103136

104137
if __name__ == "__main__":

0 commit comments

Comments
 (0)