Unable to reload inside an event loop using 'create_task'

[Corfucinas]

Kindly analyze the following code snippet and clarify if this is a potential problem or the code has some flaws, currently it is not possible to load inside a running event loop

# built-in
import asyncio

# external
import uvicorn
from fastapi import FastAPI


def main():
    event_loop = asyncio.new_event_loop()
    app_fastapi = FastAPI()
    server = uvicorn.Server(
        config=uvicorn.Config(
            app_fastapi, # also passing the module as main:app_fastapi and moving this to global does not work
            reload=True,
            loop="uvloop",
        ),
    )

    event_loop.create_task(server.serve())
    event_loop.run_forever()


if __name__ == "__main__":
    main()

The following is another snipper of code that behaves the same way as above

# built-in
import asyncio

# external
import uvicorn
from fastapi import FastAPI


async def main():
    app_fastapi = FastAPI()
    server = uvicorn.Server(
        config=uvicorn.Config(
            app_fastapi,   # also passing the module as main:app_fastapi and moving this to global does not work
            reload=True,
            loop="uvloop",
        ),
    )

    api = asyncio.create_task(server.serve())

    await asyncio.wait([api])


if __name__ == "__main__":
    asyncio.run(main())

Once running, reload will not work upon file changes.

[Kludex]

That's correct. The reload only works when using uvicorn.run.

uvicorn/uvicorn/main.py

Lines 453 to 574 in 23b9f05

	def run(
	app: typing.Union["ASGIApplication", typing.Callable, str],
	*,
	host: str = "127.0.0.1",
	port: int = 8000,
	uds: typing.Optional[str] = None,
	fd: typing.Optional[int] = None,
	loop: LoopSetupType = "auto",
	http: typing.Union[typing.Type[asyncio.Protocol], HTTPProtocolType] = "auto",
	ws: typing.Union[typing.Type[asyncio.Protocol], WSProtocolType] = "auto",
	ws_max_size: int = 16777216,
	ws_ping_interval: typing.Optional[float] = 20.0,
	ws_ping_timeout: typing.Optional[float] = 20.0,
	ws_per_message_deflate: bool = True,
	lifespan: LifespanType = "auto",
	interface: InterfaceType = "auto",
	reload: bool = False,
	reload_dirs: typing.Optional[typing.Union[typing.List[str], str]] = None,
	reload_includes: typing.Optional[typing.Union[typing.List[str], str]] = None,
	reload_excludes: typing.Optional[typing.Union[typing.List[str], str]] = None,
	reload_delay: float = 0.25,
	workers: typing.Optional[int] = None,
	env_file: typing.Optional[typing.Union[str, os.PathLike]] = None,
	log_config: typing.Optional[
	typing.Union[typing.Dict[str, typing.Any], str]
	] = LOGGING_CONFIG,
	log_level: typing.Optional[typing.Union[str, int]] = None,
	access_log: bool = True,
	proxy_headers: bool = True,
	server_header: bool = True,
	date_header: bool = True,
	forwarded_allow_ips: typing.Optional[typing.Union[typing.List[str], str]] = None,
	root_path: str = "",
	limit_concurrency: typing.Optional[int] = None,
	backlog: int = 2048,
	limit_max_requests: typing.Optional[int] = None,
	timeout_keep_alive: int = 5,
	ssl_keyfile: typing.Optional[str] = None,
	ssl_certfile: typing.Optional[typing.Union[str, os.PathLike]] = None,
	ssl_keyfile_password: typing.Optional[str] = None,
	ssl_version: int = SSL_PROTOCOL_VERSION,
	ssl_cert_reqs: int = ssl.CERT_NONE,
	ssl_ca_certs: typing.Optional[str] = None,
	ssl_ciphers: str = "TLSv1",
	headers: typing.Optional[typing.List[typing.Tuple[str, str]]] = None,
	use_colors: typing.Optional[bool] = None,
	app_dir: typing.Optional[str] = None,
	factory: bool = False,
	h11_max_incomplete_event_size: int = DEFAULT_MAX_INCOMPLETE_EVENT_SIZE,
	) -> None:
	if app_dir is not None:
	sys.path.insert(0, app_dir)
	config = Config(
	app,
	host=host,
	port=port,
	uds=uds,
	fd=fd,
	loop=loop,
	http=http,
	ws=ws,
	ws_max_size=ws_max_size,
	ws_ping_interval=ws_ping_interval,
	ws_ping_timeout=ws_ping_timeout,
	ws_per_message_deflate=ws_per_message_deflate,
	lifespan=lifespan,
	interface=interface,
	reload=reload,
	reload_dirs=reload_dirs,
	reload_includes=reload_includes,
	reload_excludes=reload_excludes,
	reload_delay=reload_delay,
	workers=workers,
	env_file=env_file,
	log_config=log_config,
	log_level=log_level,
	access_log=access_log,
	proxy_headers=proxy_headers,
	server_header=server_header,
	date_header=date_header,
	forwarded_allow_ips=forwarded_allow_ips,
	root_path=root_path,
	limit_concurrency=limit_concurrency,
	backlog=backlog,
	limit_max_requests=limit_max_requests,
	timeout_keep_alive=timeout_keep_alive,
	ssl_keyfile=ssl_keyfile,
	ssl_certfile=ssl_certfile,
	ssl_keyfile_password=ssl_keyfile_password,
	ssl_version=ssl_version,
	ssl_cert_reqs=ssl_cert_reqs,
	ssl_ca_certs=ssl_ca_certs,
	ssl_ciphers=ssl_ciphers,
	headers=headers,
	use_colors=use_colors,
	factory=factory,
	h11_max_incomplete_event_size=h11_max_incomplete_event_size,
	)
	server = Server(config=config)
	if (config.reload or config.workers > 1) and not isinstance(app, str):
	logger = logging.getLogger("uvicorn.error")
	logger.warning(
	"You must pass the application as an import string to enable 'reload' or "
	"'workers'."
	)
	sys.exit(1)
	if config.should_reload:
	sock = config.bind_socket()
	ChangeReload(config, target=server.run, sockets=[sock]).run()
	elif config.workers > 1:
	sock = config.bind_socket()
	Multiprocess(config, target=server.run, sockets=[sock]).run()
	else:
	server.run()
	if config.uds and os.path.exists(config.uds):
	os.remove(config.uds) # pragma: py-win32
	if not server.started and not config.should_reload and config.workers == 1:
	sys.exit(STARTUP_FAILURE)

[Corfucinas]

Is it possible to add this to the documentation? Do you want me to open an issue? PR?

[zakkg3]

Is there a workaround other than implement my own reloader?
My use case is another async app running (rocketry) so i have to run programatically and inside an asyncio event loop to make both of them work together.
https://rocketry.readthedocs.io/en/stable/cookbook/fastapi.html?highlight=fastapi