1import concurrent.futures as cf
 2import sys
 3import threading
 4import time
 5
 6
 7def slow_function(x: int | float) -> str:
 8    time.sleep(x)
 9    return f"Operation took {x} seconds"
10
11
12def print_progress(stop_event):
13    # Create a spinning progress indicator
14    spin_chars = ["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"]
15    while not stop_event.is_set():
16        for char in spin_chars:
17            if stop_event.is_set():  # Check if we need to stop the thread
18                return
19            sys.stdout.write(f"\r{char} Waiting...")
20            sys.stdout.flush()
21            time.sleep(0.1)
22
23
24def main():
25    stop_event = threading.Event()  # Create an event to signal thread to exit
26    with cf.ThreadPoolExecutor() as executor:
27        future_spinner = executor.submit(print_progress, stop_event)
28        try:
29            result = slow_function(2)
30        except KeyboardInterrupt as err:
31            raise err
32        finally:
33            stop_event.set()
34            cf.wait([future_spinner])
35            sys.stdout.write("\r\033[K")  # Clear the current line
36
37    return result
38
39
40if __name__ == "__main__":
41    try:
42        print(main())
43    except KeyboardInterrupt:
44        print("Operation cancelled")

 1import concurrent.futures as cf
 2import sys
 3import threading
 4import time
 5
 6
 7def progress_wheel(func):
 8    def print_progress(stop_event):
 9        # Create a spinning progress indicator
10        spin_chars = ["⣾", "⣽", "⣻", "⢿", "⡿", "⣟", "⣯", "⣷"]
11        while not stop_event.is_set():
12            for char in spin_chars:
13                if stop_event.is_set():  # Check if we need to stop the thread
14                    return
15                sys.stdout.write(f"\r{char} Waiting...")
16                sys.stdout.flush()
17                time.sleep(0.1)
18
19    def wrapped_func(*args, **kwargs):
20        stop_event = threading.Event()  # Create an event to signal thread to exit
21        with cf.ThreadPoolExecutor() as executor:
22            future_spinner = executor.submit(print_progress, stop_event)
23            try:
24                result = func(*args, **kwargs)
25            except KeyboardInterrupt as err:
26                raise err
27            finally:
28                stop_event.set()
29                cf.wait([future_spinner])
30                sys.stdout.write("\r\033[K")  # Clear the current line
31
32        return result
33
34    return wrapped_func
35
36
37@progress_wheel
38def slow_function(x: int | float) -> str:
39    time.sleep(x)
40    return f"Operation took {x} seconds"
41
42
43def main():
44    return slow_function(2)
45
46
47if __name__ == "__main__":
48    try:
49        print(main())
50    except KeyboardInterrupt:
51        print("Operation cancelled")