🐛 Fix hanging shell on exit

The shell was hanging on exit, after at least one ORM query was
performed. The cleanup() task never triggered. It now triggers reliably.
This commit is contained in:
Brian Wiborg 2025-09-28 14:49:53 +02:00
parent 82c39540a9
commit b15ce0b044
No known key found for this signature in database

View file

@ -10,9 +10,6 @@ from ohmyapi.core import scaffolding, runtime
from pathlib import Path
app = typer.Typer(help="OhMyAPI — Django-flavored FastAPI scaffolding with tightly integrated TortoiseORM.")
banner = """OhMyAPI Shell | Project: {project_name}
Find your loaded project singleton via identifier: `p`
"""
@app.command()
@ -40,52 +37,50 @@ def serve(root: str = ".", host="127.0.0.1", port=8000):
@app.command()
def shell(root: str = "."):
"""
Launch an interactive IPython shell with the project and apps loaded.
"""
project_path = Path(root).resolve()
project = runtime.Project(project_path)
# Ensure the ORM is shutdown
async def close_project():
banner = f"""
OhMyAPI Project Shell: {getattr(project.settings, 'PROJECT_NAME', 'MyProject')}
Find your loaded project singleton via identifier: `p`; i.e.: `p.apps`
"""
async def init_and_cleanup():
try:
await project.init_orm()
return True
except Exception as e:
print(f"Failed to initialize ORM: {e}")
return False
async def cleanup():
try:
await project.close_orm()
print("Tortoise ORM closed successfully.")
except Exception as e:
print(f"Error closing ORM: {e}")
def cleanup():
loop = None
try:
loop = asyncio.get_running_loop()
except RuntimeError:
pass
if loop and loop.is_running():
asyncio.create_task(close_project())
else:
asyncio.run(close_project())
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(init_and_cleanup())
# Ensure the ORM is initialized
asyncio.run(project.init_orm())
# Prepare shell vars that are to be directly available
shell_vars = {"p": project}
try:
from IPython import start_ipython
shell_vars = {
"p": project,
}
from traitlets.config.loader import Config
c = Config()
c.TerminalIPythonApp.display_banner = True
c.TerminalInteractiveShell.banner2 = banner.format(**{
"project_name": f"{f'{project.settings.PROJECT_NAME} ' if getattr(project.settings, 'PROJECT_NAME', '') else ''}[{Path(project_path).resolve()}]",
})
atexit.register(cleanup)
c.TerminalInteractiveShell.banner2 = banner
start_ipython(argv=[], user_ns=shell_vars, config=c)
except ImportError:
typer.echo("IPython is not installed. Falling back to built-in Python shell.")
import code
atexit.register(cleanup)
code.interact(local={"p": project})
code.interact(local=shell_vars, banner=banner)
finally:
loop.run_until_complete(cleanup())
@app.command()