shell: Handle async log and print
This PR works print and output in order to properly display print and log asynchronously without breaking the shell.
-
stdout_patch
is defined from the app_session- A pure
gevent
one was designed - This handle
write
above the prompt-toolkit application
- A pure
- The logging handler now use
stdout_patch
- The default
print
was reworked to useprint_formatted_text
when it's possible
Supported
Concurrent | Output | Validity |
---|---|---|
gevent | logging | |
gevent | ||
async | logging | |
async |
|
|
thread | logging | |
thread |
Works
import gevent
import logging
logger = logging.getLogger("foobar")
def process_log_gevent():
for _ in range(5):
gevent.sleep(1)
logger.error("log from gevent")
g = gevent.spawn(process_log_gevent)
import gevent
def process_print_gevent():
for _ in range(5):
gevent.sleep(1)
print("print from gevent")
g = gevent.spawn(process_print_gevent)
import asyncio
import logging
logger = logging.getLogger("foobar")
async def process_log_async():
for _ in range(5):
await asyncio.sleep(1)
logger.error("log from asyncio")
asyncio.ensure_future(process_log_async())
import threading
import time
import logging
class Foo(threading.Thread):
def run(self):
for i in range(5):
logging.error("log from thread")
time.sleep(1)
foo = Foo()
foo.start()
Works with local patched
import asyncio
from prompt_toolkit.application import current
app_session = current.get_app_session()
async def process_async():
current._current_app_session.set(app_session)
for _ in range(5):
await asyncio.sleep(1)
print("print from asyncio")
asyncio.ensure_future(process_async())
Not working
import threading
import time
import logging
class Foo(threading.Thread):
def run(self):
for i in range(5):
print("print from thread")
time.sleep(1)
foo = Foo()
foo.start()
import threading
import time
import logging
# app_session = current.get_app_session()
class Foo(threading.Thread):
def __init__(self, *args, **kwargs):
threading.Thread.__init__(self, *args, **kwargs)
self.ctx = contextvars.copy_context()
# current._current_app_session.set(app_session)
def run(self):
self.ctx.run(self._run)
def _run(self):
for i in range(5):
time.sleep(1)
print("print from thread")
foo = Foo()
foo.start()
Edited by Valentin Valls