Repository: ansible/awx
File: awx/main/routing.py (Python)

Code before:
    def __init__(self, *auths):
        self._auths = [a for a in auths]

    async def __call__(self, scope, receive, send):
        """
        Loop through the list of passed in URLRouter's (they may or may not be wrapped by auth).
        We know we have exhausted the list of URLRouter patterns when we get a
        ValueError('No route found for path %s'). When that happens, move onto the next
        URLRouter.
        If the final URLRouter raises an error, re-raise it in the end.

        We know that we found a match when no error is raised, end the loop.
        """
        last_index = len(self._auths) - 1
        for i, auth in enumerate(self._auths):
            try:
                return await auth.__call__(scope, receive, send)
            except ValueError as e:
                if str(e).startswith('No route found for path'):
                    # Only surface the error if on the last URLRouter
                    if i == last_index:
                        raise


websocket_urlpatterns = [
    re_path(r'api/controller/v2/websocket/$', consumers.EventConsumer.as_asgi()),
    re_path(r'api/websocket/$', consumers.EventConsumer.as_asgi()),
    re_path(r'websocket/$', consumers.EventConsumer.as_asgi()),
]
websocket_relay_urlpatterns = [
    re_path(r'websocket/relay/$', consumers.RelayConsumer.as_asgi()),
]


def application_func(cls=AWXProtocolTypeRouter) -> ProtocolTypeRouter:
    return cls(
        {
            'websocket': MultipleURLRouterAdapter(
                URLRouter(websocket_relay_urlpatterns),
                DrfAuthMiddlewareStack(URLRouter(websocket_urlpatterns)),
            )
        }
    )


def __getattr__(name: str) -> ProtocolTypeRouter:
    """
    Defer instantiating application.
    For testing, we just need it to NOT run on import.

    https://peps.python.org/pep-0562/#specification

Review comment (suggestion):
not sure if we want to hardcode `controller` here or reuse the `OPTIONAL_API_URLPATTERN_PREFIX` variable from the settings to build this path

https://github.com/ansible/awx/blob/devel/awx/settings/defaults.py

Code after:
    def __init__(self, *auths):
        self._auths = [a for a in auths]

    async def __call__(self, scope, receive, send):
        """
        Loop through the list of passed in URLRouter's (they may or may not be wrapped by auth).
        We know we have exhausted the list of URLRouter patterns when we get a
        ValueError('No route found for path %s'). When that happens, move onto the next
        URLRouter.
        If the final URLRouter raises an error, re-raise it in the end.

        We know that we found a match when no error is raised, end the loop.
        """
        last_index = len(self._auths) - 1
        for i, auth in enumerate(self._auths):
            try:
                return await auth.__call__(scope, receive, send)
            except ValueError as e:
                if str(e).startswith('No route found for path'):
                    # Only surface the error if on the last URLRouter
                    if i == last_index:
                        raise


websocket_urlpatterns = [
    re_path(r'api/websocket/$', consumers.EventConsumer.as_asgi()),
    re_path(r'websocket/$', consumers.EventConsumer.as_asgi()),
]

if settings.OPTIONAL_API_URLPATTERN_PREFIX:
    websocket_urlpatterns.append(re_path(r'api/{}/v2/websocket/$'.format(settings.OPTIONAL_API_URLPATTERN_PREFIX), consumers.EventConsumer.as_asgi()))

websocket_relay_urlpatterns = [
    re_path(r'websocket/relay/$', consumers.RelayConsumer.as_asgi()),
]


def application_func(cls=AWXProtocolTypeRouter) -> ProtocolTypeRouter:
    return cls(
        {
            'websocket': MultipleURLRouterAdapter(
                URLRouter(websocket_relay_urlpatterns),
                DrfAuthMiddlewareStack(URLRouter(websocket_urlpatterns)),
            )
        }
    )


def __getattr__(name: str) -> ProtocolTypeRouter:
    """
    Defer instantiating application.