I have a program that now will get a REST API. Instead of rewriting a bunch of code I thought I would just run the program, capture the prints and put that into a message response.
On How to capture stdout output from a Python function call? I saw:
- this solution which is apparently not thread-safe. I tried it with concurrent futures where it worked, but I trust the assessment there that it is not thread safe.
- I also noticed the implementation here, but while it uses threads to be asynchronous, it never explicitly states that it is itself thread-safe.
This is how I tried to test whether it is thread-safe or not, before I read that it is apparently not thread safe. I never had any issues with this (i.e. output always was ['hello world', 'hello world2'], but maybe this is a characteristic of concurrent futures that is not present when using asynchronous functions from REST API modules (FastAPI in my case).
import concurrent.futures
from io import StringIO
import sys
def main():
num_tests = 30
with concurrent.futures.ThreadPoolExecutor() as executor:
futures = [executor.submit(test) for _ in range(num_tests)]
for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
except Exception as e:
print(f"An error occurred: {e}")
class Capturing(list):
def __enter__(self):
self._stdout = sys.stdout
sys.stdout = self._stringio = StringIO()
return self
def __exit__(self, *args):
self.extend(self._stringio.getvalue().splitlines())
del self._stringio # free up some memory
sys.stdout = self._stdout
def test():
with Capturing() as output:
print('hello world')
print('displays on screen')
with Capturing(output) as output: # note the constructor argument
print('hello world2')
print('done')
print('output:', output)
main()
In the best case I'm looking for an idea how to capture stdout asynchronously. In the worst case an explanation why that isn't possible. In that case I will be looking for other solutions.
