ACIL FM

Nama	Ukuran	Aksi
__pycache__	-	Open
base_events.py	73.62 MB	View DL Edit
base_futures.py	1.96 MB	View DL Edit
base_subprocess.py	8.66 MB	View DL Edit
base_tasks.py	2.58 MB	View DL Edit
constants.py	1.29 MB	View DL Edit
coroutines.py	3.32 MB	View DL Edit
events.py	27.97 MB	View DL Edit
exceptions.py	1.71 MB	View DL Edit
format_helpers.py	2.35 MB	View DL Edit
futures.py	13.88 MB	View DL Edit
locks.py	18.57 MB	View DL Edit
log.py	124 B	View DL Edit
mixins.py	481 B	View DL Edit
proactor_events.py	32.48 MB	View DL Edit
protocols.py	6.79 MB	View DL Edit
queues.py	7.79 MB	View DL Edit
runners.py	6.68 MB	View DL Edit
selector_events.py	44.34 MB	View DL Edit
sslproto.py	31 MB	View DL Edit
staggered.py	5.85 MB	View DL Edit
streams.py	26.86 MB	View DL Edit
subprocess.py	7.5 MB	View DL Edit
taskgroups.py	8.27 MB	View DL Edit
tasks.py	33.63 MB	View DL Edit
threads.py	790 B	View DL Edit
timeouts.py	5.2 MB	View DL Edit
transports.py	10.47 MB	View DL Edit
trsock.py	2.42 MB	View DL Edit
unix_events.py	50.7 MB	View DL Edit
windows_events.py	33.88 MB	View DL Edit
windows_utils.py	4.94 MB	View DL Edit
__init__.py	1.16 MB	View DL Edit
__main__.py	3.3 MB	View DL Edit

Edit file: /opt/alt/python311/lib64/python3.11/asyncio/__main__.py

import ast
import asyncio
import code
import concurrent.futures
import inspect
import sys
import threading
import types
import warnings

from . import futures

class AsyncIOInteractiveConsole(code.InteractiveConsole):

def __init__(self, locals, loop):
        super().__init__(locals)
        self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT

self.loop = loop

def runcode(self, code):
        future = concurrent.futures.Future()

def callback():
            global repl_future
            global repl_future_interrupted

repl_future = None
            repl_future_interrupted = False

func = types.FunctionType(code, self.locals)
            try:
                coro = func()
            except SystemExit:
                raise
            except KeyboardInterrupt as ex:
                repl_future_interrupted = True
                future.set_exception(ex)
                return
            except BaseException as ex:
                future.set_exception(ex)
                return

if not inspect.iscoroutine(coro):
                future.set_result(coro)
                return

try:
                repl_future = self.loop.create_task(coro)
                futures._chain_future(repl_future, future)
            except BaseException as exc:
                future.set_exception(exc)

loop.call_soon_threadsafe(callback)

try:
            return future.result()
        except SystemExit:
            raise
        except BaseException:
            if repl_future_interrupted:
                self.write("\nKeyboardInterrupt\n")
            else:
                self.showtraceback()

class REPLThread(threading.Thread):

def run(self):
        try:
            banner = (
                f'asyncio REPL {sys.version} on {sys.platform}\n'
                f'Use "await" directly instead of "asyncio.run()".\n'
                f'Type "help", "copyright", "credits" or "license" '
                f'for more information.\n'
                f'{getattr(sys, "ps1", ">>> ")}import asyncio'
            )

console.interact(
                banner=banner,
                exitmsg='exiting asyncio REPL...')
        finally:
            warnings.filterwarnings(
                'ignore',
                message=r'^coroutine .* was never awaited$',
                category=RuntimeWarning)

loop.call_soon_threadsafe(loop.stop)

if __name__ == '__main__':
    sys.audit("cpython.run_stdin")

loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

repl_locals = {'asyncio': asyncio}
    for key in {'__name__', '__package__',
                '__loader__', '__spec__',
                '__builtins__', '__file__'}:
        repl_locals[key] = locals()[key]

console = AsyncIOInteractiveConsole(repl_locals, loop)

repl_future = None
    repl_future_interrupted = False

try:
        import readline  # NoQA
    except ImportError:
        pass

repl_thread = REPLThread()
    repl_thread.daemon = True
    repl_thread.start()

while True:
        try:
            loop.run_forever()
        except KeyboardInterrupt:
            if repl_future and not repl_future.done():
                repl_future.cancel()
                repl_future_interrupted = True
            continue
        else:
            break

Batal