Skip to content

Commit 90cd211

Browse files
wasin-waeosriwasin-waeosri
committed
created async RDP code, added learn async example
1 parent d1903be commit 90cd211

4 files changed

Lines changed: 175 additions & 37 deletions

File tree

‎README.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,3 +76,9 @@ For demo purposes, scripts print token/output payloads and endpoint responses.
7676
## License
7777

7878
Apache 2.0. See [LICENSE.md](LICENSE.md).
79+
80+
## Reference
81+
82+
- https://realpython.com/async-io-python/
83+
- https://www.twilio.com/en-us/blog/asynchronous-http-requests-in-python-with-httpx-and-asyncio
84+

‎src/async_learn.py‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
import asyncio
2+
import random
3+
import time
4+
5+
async def main():
6+
user_ids = [1, 2, 3]
7+
start = time.perf_counter()
8+
await asyncio.gather(
9+
*(get_user_with_posts(user_id) for user_id in user_ids), return_exceptions=True
10+
)
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)"
34+
)
35+
for post in posts:
36+
print(f" - {post}")
37+
38+
if __name__ == "__main__":
39+
random.seed(42)
40+
asyncio.run(main())

‎src/example_async.py‎

Lines changed: 103 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,108 @@
1-
2-
import httpx
31
import asyncio
2+
import json
3+
import os
4+
import time
5+
import httpx
6+
from dotenv import load_dotenv
7+
8+
AUTH_TOKEN_URL = "/auth/oauth2/v1/token"
9+
CHAIN_URL = "/data/pricing/chains/v1/"
10+
11+
RICS = ["EFX=", "0#.SPX", "0#.FTSE"] # Fetched concurrently
12+
13+
async def post_authentication(machine_id, password, app_key, url, client):
14+
"""Authenticate to RDP and return the token response as JSON."""
15+
# Build the OAuth 2.0 Password Grant request payload.
16+
# Sent as application/x-www-form-urlencoded (httpx encodes a dict automatically).
17+
payload = {
18+
"username": machine_id, # RDP Machine-ID
19+
"password": password, # RDP Password
20+
"grant_type": "password", # OAuth 2.0 grant type
21+
"scope": "trapi", # Target API scope
22+
"takeExclusiveSignOnControl": "true", # Revoke other active sessions
23+
"client_id": app_key # RDP AppKey (acts as client_id)
24+
}
25+
26+
headers = {
27+
"Content-Type": "application/x-www-form-urlencoded"
28+
}
29+
30+
# Send authentication request to the OAuth token endpoint.
31+
# `data=payload` sends a form body required by this endpoint.
32+
response_auth = await client.post(url, data=payload, headers=headers)
33+
response_auth.raise_for_status() # Raise for 4xx/5xx API failures.
34+
return response_auth.json()
435

5-
async def fetch(url):
6-
async with httpx.AsyncClient(verify=False, timeout=10.0) as client:
7-
r = await client.get(url)
8-
return r.text
36+
async def get_chain(ric, token, url, client):
37+
"""Fetch chain data for a single RIC symbol using an access token."""
38+
headers = {
39+
"Authorization": f"Bearer {token}",
40+
"Content-Type": "application/json",
41+
}
42+
# Query string parameters sent with the GET request.
43+
parameters = {
44+
"universe": ric
45+
}
46+
47+
# Request chain data from the pricing chains endpoint.
48+
response_chain = await client.get(url, params=parameters, headers=headers)
49+
response_chain.raise_for_status()
50+
return response_chain.json()
951

1052
async def main():
11-
urls = ["https://example.com", "https://python.org"]
12-
results = await asyncio.gather(*(fetch(u) for u in urls))
13-
print(results)
53+
"""Main entry point for the async example."""
54+
load_dotenv()
55+
56+
# Read credentials and base URL from environment variables.
57+
machine_id = os.getenv("MACHINEID_RDP")
58+
password = os.getenv("PASSWORD_RDP")
59+
app_key = os.getenv("APPKEY_RDP")
60+
base_url = os.getenv("RDP_BASE_URL")
61+
62+
# Reuse one connection pool across all requests.
63+
async with httpx.AsyncClient(
64+
verify=False,
65+
base_url=base_url,
66+
timeout=10.0,
67+
follow_redirects=True,
68+
) as client:
69+
# --- Authentication (must complete before any data requests) ---
70+
try:
71+
token_data = await post_authentication(machine_id, password, app_key, AUTH_TOKEN_URL, client)
72+
print("Authentication successful. Access token obtained.")
73+
except httpx.HTTPStatusError as e:
74+
print(f"HTTP error during authentication: {e.response.status_code} - {e.response.text}")
75+
return
76+
except httpx.HTTPError as e:
77+
print(f"Network error during authentication: {e}")
78+
return
79+
except Exception as e:
80+
print(f"Unexpected error during authentication: {e}")
81+
return
82+
83+
access_token = token_data.get("access_token")
84+
85+
# --- Fetch all RICs concurrently ---
86+
# tasks = [get_chain(ric, access_token, CHAIN_URL, client) for ric in RICS]
87+
# results = await asyncio.gather(*tasks, return_exceptions=True)
88+
89+
# for ric, result in zip(RICS, results):
90+
# if isinstance(result, httpx.HTTPStatusError):
91+
# print(f"HTTP error fetching '{ric}': {result.response.status_code} - {result.response.text}")
92+
# elif isinstance(result, Exception):
93+
# print(f"Error fetching '{ric}': {result}")
94+
# else:
95+
# print(f"Chain data for '{ric}': {result}")
96+
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+
14103

15-
asyncio.run(main())
104+
if __name__ == "__main__":
105+
start = time.perf_counter()
106+
asyncio.run(main())
107+
elapsed = time.perf_counter() - start
108+
print(f"{__file__} executed in {elapsed:0.2f} seconds.")

‎src/example_client.py‎

Lines changed: 26 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
import json
22
import os
33
import time
4-
from typing import Any, Sequence
54

65
import httpx
76
from dotenv import load_dotenv
@@ -45,7 +44,6 @@ def post_authentication(machine_id, password, app_key, url, client):
4544

4645
# Send authentication request to the OAuth token endpoint.
4746
# `data=payload` sends a form body required by this endpoint.
48-
# `verify=False` skips SSL verification (for local/dev only). This is not recommended for production use. I use it to avoid LSEG beloved ZScaler.
4947
response = client.post(url, data=payload)
5048
response.raise_for_status() # Raise for 4xx/5xx API failures.
5149
return response.json()
@@ -59,7 +57,6 @@ def get_chain(ric, token, url, client):
5957
"universe": ric
6058
}
6159
# Request chain data from the pricing chains endpoint.
62-
# `verify=False` skips SSL verification (for local/dev only). This is not recommended for production use. I use it to avoid LSEG beloved ZScaler.
6360
response = client.get(url, params=parameters, headers=headers)
6461
response.raise_for_status()
6562
return response.json()
@@ -75,7 +72,7 @@ def post_historical_event(rics, token, url, client):
7572
}
7673

7774
# `json=payload` serializes and sends JSON in the request body.
78-
# `verify=False` skips SSL verification (for local/dev only). This is not recommended for production use. I use it to avoid LSEG beloved ZScaler.
75+
7976
response = client.post(url, json=payload, headers=headers)
8077
response.raise_for_status()
8178
return response.json()
@@ -91,7 +88,7 @@ def post_auth_refresh(app_key, refresh_token, url, client):
9188
headers = {
9289
"Content-Type": "application/x-www-form-urlencoded"
9390
}
94-
# `verify=False` skips SSL verification (for local/dev only). This is not recommended for production use. I use it to avoid LSEG beloved ZScaler.
91+
9592
response = client.post(url, data=payload, headers=headers)
9693
response.raise_for_status()
9794
return response.json()
@@ -105,7 +102,6 @@ def post_auth_revoke(token, app_key, url, client):
105102

106103
payload = f"token={token}"
107104
auth = httpx.BasicAuth(username=app_key, password="")
108-
# `verify=False` skips SSL verification (for local/dev only). This is not recommended for production use. I use it to avoid LSEG beloved ZScaler.
109105
response = client.post(url, data=payload, headers=headers, auth=auth)
110106
response.raise_for_status()
111107

@@ -144,26 +140,26 @@ def main() -> None:
144140
print(f"Chain data for {ric}:", json.dumps(chain_data, indent=2))
145141

146142
# Example multi-RIC request for historical events endpoint.
147-
rics = ["LSEG.L", "VOD.L", "BP.L"]
148-
print(f"Posting historical event data... for RICs: {rics}")
149-
historical_event_data = post_historical_event(rics, access_token, HISTORICAL_EVENT_URL, client)
150-
print("Historical event data retrieved successfully!")
151-
print(f"Historical event data for {rics}:", json.dumps(historical_event_data))
152-
153-
refresh_token = token_data.get("refresh_token")
154-
if refresh_token:
155-
time.sleep(5) # Sleep for 5 seconds before refreshing token (for demo purposes)
156-
print("Refreshing access token...")
157-
refreshed_token_data = post_auth_refresh(app_key, refresh_token, AUTH_TOKEN_URL, client)
158-
print("Token refreshed successfully!")
159-
print("New Access Token:", json.dumps(refreshed_token_data["access_token"], indent=2))
160-
else:
161-
print("No refresh token available. Cannot refresh access token.")
162-
163-
time.sleep(5) # Sleep for 5 seconds before revoking token (for demo purposes)
164-
print("Revoking access token...")
165-
post_auth_revoke(access_token, app_key, AUTH_REVOKE_URL, client)
166-
print("Access token revoked successfully.")
143+
# rics = ["LSEG.L", "VOD.L", "BP.L"]
144+
# print(f"Posting historical event data... for RICs: {rics}")
145+
# historical_event_data = post_historical_event(rics, access_token, HISTORICAL_EVENT_URL, client)
146+
# print("Historical event data retrieved successfully!")
147+
# print(f"Historical event data for {rics}:", json.dumps(historical_event_data))
148+
149+
# refresh_token = token_data.get("refresh_token")
150+
# if refresh_token:
151+
# time.sleep(5) # Sleep for 5 seconds before refreshing token (for demo purposes)
152+
# print("Refreshing access token...")
153+
# refreshed_token_data = post_auth_refresh(app_key, refresh_token, AUTH_TOKEN_URL, client)
154+
# print("Token refreshed successfully!")
155+
# print("New Access Token:", json.dumps(refreshed_token_data["access_token"], indent=2))
156+
# else:
157+
# print("No refresh token available. Cannot refresh access token.")
158+
159+
# time.sleep(5) # Sleep for 5 seconds before revoking token (for demo purposes)
160+
# print("Revoking access token...")
161+
# post_auth_revoke(access_token, app_key, AUTH_REVOKE_URL, client)
162+
# print("Access token revoked successfully.")
167163

168164
else:
169165
print("Failed to receive access token. Exiting...")
@@ -181,4 +177,7 @@ def main() -> None:
181177

182178

183179
if __name__ == "__main__":
184-
main()
180+
start = time.perf_counter()
181+
main()
182+
elapsed = time.perf_counter() - start
183+
print(f"{__file__} executed in {elapsed:0.2f} seconds.")

0 commit comments

Comments
 (0)