Redirect upon a failure or successful conditional test

[sjehuda]

This is relevant to JS, outdated client software, et cetera.

I attempt to do something of sort with XSLT and ECMA.

https://biglist.com/lists/lists.mulberrytech.com/xsl-list/archives/202601/msg00041.html

This "middleware" is my cirrent attempt.

It could require a minute or so, because it is activated upon every request.

    @middleware
    async def client_check(
        request: Request,
        handler: Callable[[Request], Awaitable[StreamResponse]]
    ) -> StreamResponse:
        """
        Redirect to (X)HTML documents if XSLT or ECMA were not detected."
        """
        #timestamp = datetime.datetime.now().timestamp()
        await asyncio.sleep(5)
        if request.remote in clients:
            client = clients[request.remote]
            print(client)
            #difference = timestamp - client["time"]
            #if int(difference) > 5:
            if client["xslt"] == False and client["ecma"] == False:
                # Redirect to XHTML.
                print("REDIRECT XHTML")
            elif client["xslt"] == False and client["ecma"] == True:
                # Redirect to SaxonJS.
                print("REDIRECT ECMA")
        return await handler(request)

Client requests that are sent to "/support?feature=xslt" and "/support?feature=ecma" would set the values of client["xslt"] and client["ecma"] to True.

I would appreciate your help.

This is the project which involves XSLT and aiohttp.

https://git.xmpp-it.net/sch/Rivista

[Dreamsorcerer]

Sorry, what is the question/problem?

If it's how to do a redirect, see https://docs.aiohttp.org/en/stable/web_quickstart.html#redirects

[sjehuda]

This is not the same. Concern ------- Suppose that an HTTP client navigates to a URI (GET method). The HTTP client has no JS (e.g. it may be a software such as Dillo). That page is relied on JS; otherwise, there is another page without JS. Plan ---- So, the JS page offers a JS text. var xmlHttpRequest; xmlHttpRequest = new XMLHttpRequest(); xmlHttpRequest.open("GET", "/support?feature=ecma", true); xmlHttpRequest.send(); If aiohttp receives the request of "/support?feature=ecma"; then it would do nothing. However, if aiohttp does not receive the request of "/support?feature=ecma", and no further requests are sent, after five seconds, after the last request; then aiohttp would redirect to the page without JS.

[Dreamsorcerer]

I'm still not seeing a question/problem. Great, you have a plan; have you tried it? If so, what went wrong? What part are you actually struggling with?

[sjehuda]

Have you tried it?

Yes.

If so, what went wrong?

This middleware is executed upon every request, so the awaiting directive of five seconds causes to the page load to occur for five seconds times the number of requests; that is, five seconds per request, which be fifty seconds per ten requests.

What part are you actually struggling with?

The unnecessary time span. I want this middle ware to be executed once; or, the least times as possible. Perhaps, I want for this middleware to immediately retract upon each new request; Once there are no more requests, the middleware would not retract; and Then, the process would occur.

[Dreamsorcerer]

This middleware is executed upon every request, so the awaiting directive of five seconds causes to the page load to occur for five seconds times the number of requests; that is, five seconds per request, which be fifty seconds per ten requests.

But, it's async. If you received 10 requests at the same time, it would wait a total of 5 seconds for all 10 requests...

I want this middle ware to be executed once; or, the least times as possible.

To put that another way, you want something to run after 5 seconds of receiving no requests?

In which case, you could try something like this (untested):

class MyMiddleware:
    def __init__(self, delay: float = 5):
        self._task: asyncio.Task[None] | None = None
        self._delay = delay

    async def task(self) -> None:
        await asyncio.sleep(self._delay)
        # Do Thing
        self._task = None

    async def __call__(
        self,
        request: web.Request,
        handler: Callable[[web.Request], Awaitable[web.StreamResponse]]
    ) -> web.StreamResponse:
        if self._task is not None:
            self._task.cancel()
        self._task = asyncio.create_task(self.task())

        return await handler(request)

Or, as you mention redirects, if you just mean to cancel the sleep when a new request comes in, cutting the delay down, then it could look something more like:

class MyMiddleware:
    def __init__(self, delay: float = 5):
        self._task: asyncio.Task[None] | None = None
        self._delay = delay

    async def __call__(
        self,
        request: web.Request,
        handler: Callable[[web.Request], Awaitable[web.StreamResponse]]
    ) -> web.StreamResponse:
        if self._task is not None:
            self._task.cancel()
        self._task = asyncio.create_task(asyncio.sleep(self._delay))

        with contextlib.suppress(asyncio.CancelledError):
            await self._task

        return await handler(request)

[Dreamsorcerer]

Or, cancelling the sleep, but only running some code if the delay was reached:

class MyMiddleware:
    def __init__(self, delay: float = 5):
        self._task: asyncio.Task[None] | None = None
        self._delay = delay

    async def __call__(
        self,
        request: web.Request,
        handler: Callable[[web.Request], Awaitable[web.StreamResponse]]
    ) -> web.StreamResponse:
        if self._task is not None:
            self._task.cancel()
        self._task = asyncio.create_task(asyncio.sleep(self._delay))

        try:
            await self._task
        except asyncio.CancelledError:
            pass  # New request came in, skip to processing request
        else:  # Timeout was reached.
            # Do Thing

        return await handler(request)

[sjehuda]

Sam. Good afternoon. I am trying with the third solution (with directives try/except); and That solution appears to be working. I am testing with Dillo, and Falkon. I send this message to inform you that your solution works. I will further test, and inform you of results. I thank you very much for your time and effort. Please. Check your email.

[sjehuda]

Attempting to route to documents causes to the test to fail.

This is the middleware which I currently attempt to utilize.

class MyMiddleware:

    def __init__(self, delay: float = 5):
        self._task: asyncio.Task[None] | None = None
        self._delay = delay

    async def __call__(
        self,
        request: Request,
        handler: Callable[[Request], Awaitable[StreamResponse]]
    ) -> StreamResponse:
        if self._task is not None:
            self._task.cancel()
            self._task = asyncio.create_task(asyncio.sleep(self._delay))

            try:
                await self._task
            except asyncio.CancelledError:
                # A new request is engaged, skip to processing request
                #del clients[request.remote]
                pass 
        else:
            # Timeout was reached.
            if request.remote in clients:
                client = clients[request.remote]
                print(client)
                if client["xslt"] == False and client["ecma"] == False:
                    # Route to (X)HTML.
                    #response = HttpRoutes.pages(request, xml=False)
                    print("Redirect to (X)HTML")
                elif client["xslt"] == False and client["ecma"] == True:
                    # Route to SaxonJS.
                    #response = HttpRoutes.pages(request, xml=False)
                    print("Redirect to SaxonJS")
                else:
                    # Route to XML.
                    #response = HttpRoutes.pages(request, xml=True)
                    print("Redirect to XML")
                #return response

        return await handler(request)

@middleware
async def my_middleware(request: Request, handler: Callable[[Request], Awaitable[StreamResponse]]) -> StreamResponse:
    middleware_instance = MyMiddleware(delay=5)  # Create an instance of your middleware
    return await middleware_instance.__call__(request, handler)

This is the function HttpRoutes.pages.

    def pages(request: Request, xml=True):
        if "directory" in request.match_info:
            directory = request.match_info["directory"]
        else:
            directory = ""
        if directory not in ("graphics", "images", "scripts"):
            if "identifier" in request.match_info:
                identifier = request.match_info["identifier"]
            else:
                identifier = ""
            if xml:
                filename = "index.atom"
                mimetype = "application/xml" # application/atom+xml
            else:
                filename = "index.html"
                mimetype = "text/html"
            pathname = os.path.join(publish_properties.directory_output,
                                    directory, identifier, filename)

            if os.path.exists(pathname):
                result = open(pathname, mode="r").read()
                response = Response(
                    text=result,
                    content_type=mimetype)
            elif not os.path.exists(pathname):
                #logger.error(f"{function_name}		{e}")
                # NOTE This condition is redundant.
                if directory not in ("graphics", "images", "scripts"):
                    raise HTTPFound(f"{request.raw_path}/")
                else:
                    # NOTE Error 403 might be better appropriate.
                    response = HttpErrors.handle_404(request)
            return response

Another issue, of lesser importance, is that I failed to clear the dictionary of the IP Address from the "clients" dictionary (del clients[request.remote]), after all requests were exhausted.

[Dreamsorcerer]

    middleware_instance = MyMiddleware(delay=5)  # Create an instance of your middleware

You're creating a new instance on every request, so that's clearly not going to work. I think you should be able to just use the class directly:

app.middlewares.append(MyMiddleware(delay=5))

        if self._task is not None:
            self._task.cancel()
            self._task = asyncio.create_task(asyncio.sleep(self._delay))

            try:
                await self._task
            except asyncio.CancelledError:
                # A new request is engaged, skip to processing request
                #del clients[request.remote]
                pass 
        else:

This indentation is wrong; compare to my original suggestion. Task starts as None and the only place you set it to something else is after checking that it's not None; the code clearly will never run.

Another issue, of lesser importance, is that I failed to clear the dictionary of the IP Address from the "clients" dictionary (del clients[request.remote]), after all requests were exhausted.

Either figure out a place to clear them or use some kind of LRU dict to limit the maximum size of the dict.