Created
July 5, 2025 11:09
-
-
Save cumulus13/fa0502a874e721f3543bb99de83ee831 to your computer and use it in GitHub Desktop.
Replace requests.Session with requests.Session + Progressbar (rich)
This file contains hidden or 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
class ProgressSession(requests.Session): | |
def request(self, method, url, *args, max_try=3, retry_delay=1, **kwargs): | |
attempt = 0 | |
last_exception = None | |
with Progress( | |
SpinnerColumn(), | |
TextColumn("[progress.description]{task.description}"), | |
console=console, | |
transient=True | |
) as progress: | |
task = progress.add_task(f"[#FFFF00]Connecting ...[/]", total=None) | |
while attempt < max_try: | |
attempt += 1 | |
try: | |
progress.update(task, description=f"[#FFFF00]Attempt {attempt}/{max_try}: {method.upper()} {url}[/]") | |
response = super().request(method, url, *args, **kwargs) | |
response.raise_for_status() | |
return response | |
except RequestException as e: | |
last_exception = e | |
progress.update(task, description=f"[red]Attempt {attempt} failed[/]") | |
if attempt < max_try: | |
time.sleep(retry_delay) | |
else: | |
break # stop retrying | |
# All attempts failed | |
progress.update(task, description=f"[red]All {max_try} attempts failed[/]") | |
console.print( | |
Panel.fit( | |
Text.from_markup(f"[white on red]ERROR:[/] {last_exception}"), | |
title="Request Failed", | |
border_style="red" | |
) | |
) | |
raise last_exception # raise after retries |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment