Launch IPython shell synchronously. IPython manages its own event loop for autoawait, so this must be called from a synchronous context to avoid nested event loop errors. Args: namespace: The namespace dict to make available in the shell
(namespace: dict[str, Any])
| 75 | |
| 76 | |
| 77 | def _launch_ipython_shell(namespace: dict[str, Any]) -> None: |
| 78 | """Launch IPython shell synchronously. |
| 79 | |
| 80 | IPython manages its own event loop for autoawait, so this must be called |
| 81 | from a synchronous context to avoid nested event loop errors. |
| 82 | |
| 83 | Args: |
| 84 | namespace: The namespace dict to make available in the shell |
| 85 | """ |
| 86 | # Apply nest_asyncio to allow IPython to run its own event loop |
| 87 | # This is needed because we're already inside an async context |
| 88 | import nest_asyncio |
| 89 | |
| 90 | nest_asyncio.apply() |
| 91 | |
| 92 | with contextlib.suppress(EOFError, ValueError): |
| 93 | # Configure IPython for async/await support |
| 94 | from IPython.terminal.embed import InteractiveShellEmbed |
| 95 | |
| 96 | model_names = [ |
| 97 | k for k in namespace.keys() if k not in ("Tortoise", "tortoise", "connections", "apps") |
| 98 | ] |
| 99 | models_info = ( |
| 100 | f"Available models: {', '.join(model_names)}" if model_names else "No models loaded" |
| 101 | ) |
| 102 | |
| 103 | banner = ( |
| 104 | "Tortoise ORM Shell (IPython with async support)\n" |
| 105 | f"{models_info}\n" |
| 106 | "Use 'await' directly for async operations (e.g., 'await YourModel.all()').\n" |
| 107 | ) |
| 108 | |
| 109 | # Create IPython shell with async autoawait enabled |
| 110 | ipshell = InteractiveShellEmbed( |
| 111 | user_ns=namespace, |
| 112 | banner1=banner, |
| 113 | ) |
| 114 | # Enable autoawait for top-level await |
| 115 | ipshell.autoawait = True |
| 116 | ipshell() |
| 117 | |
| 118 | |
| 119 | async def _launch_ptpython_shell(namespace: dict[str, Any]) -> None: |