Connexion with running async environments

[alexgeraldo]

Description

Connexion has been an invaluable tool for me as a beginner developer working on cloud projects. However, I've encountered a challenge when trying to integrate it into a specific environment where I already have my own async setup with other apps (like grpc server and aiomysql pool). Unfortunately, I couldn't find a straightforward way to serve the Connexion app within this existing async environment.

Expected behavior

I hope to see an alternative method, perhaps something like serve(), that allows the Connexion app to be served within an already running async environment. This would be similar to using uvicorn.Server.serve() as outlined in the uvicorn official documentation.

Suggested Improvement

I suggest adding a new method to Connexion, such as serve(), which internally utilizes uvicorn's async capabilities. Below is a simple implementation that I've found effective:

class MyAsyncApp(AsyncApp):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    async def serve(self, **kwargs):
        try:
            import uvicorn
        except ImportError:
            raise RuntimeError(
                "uvicorn is not installed. Please install Connexion using the uvicorn extra "
                "(connexion[uvicorn])"
            )
        app = self.middleware
        config = uvicorn.Config(app, **kwargs)
        server = uvicorn.Server(config)
        await server.serve()

# SERVING CONNEXION ALONGSIDE OTHER APPS (EXCERPT FROM MY CODE)
async def connexion_run_task():
    rest_server.add_api("./api/specification.yaml", swagger_ui_options=SwaggerUIOptions(swagger_ui_config={"persistAuthorization": True}))
    await rest_server.serve(host="0.0.0.0", port=env.REST_APP_PORT)

async def main():
    # Create tasks for both Connexion and async gRPC servers
    connexion_task = asyncio.create_task(connexion_run_task())
    grpc_task = asyncio.create_task(grpc_run_task())

    # Wait for both tasks to complete
    await asyncio.gather(connexion_task, grpc_task)

# Run the asyncio event loop
if __name__ == "__main__":
    asyncio.run(main())

This addition would provide Connexion users with the flexibility to seamlessly integrate their API applications into existing async environments, enhancing the overall versatility and usability of the framework.

I'm new to this, so please feel free to provide any guidance or feedback.

[RobbeSneyders]

Hi @alexgeraldo,

You can run your connexion application using uvicorn, so I would expect it to work with serve as well.

If your connexion application is defined as app in main.py, I would expect the following snippet from the documentation you linked to work:

async def main():
    config = uvicorn.Config("main:app", port=5000, log_level="info")
    server = uvicorn.Server(config)
    await server.serve()

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

Let me know if that isn't the case.