#🔒 async @contectlib.contextmanager

74 messages · Page 1 of 1 (latest)

low pecan
#

Re: https://discord.com/channels/267624335836053506/1231477734069374996

I am trying to convert this synchronous code to asynchronous and i am half way through thanks to @twin peak but i am unsure how to proceed with the tricky part.

import contextlib
from urllib.request import urlopen
from aiofiles import os, open

async def info(inp):
    async with openstream(inp) as stream:
        return await probe(stream)

@contextlib.asynccontextmanager
async def openstream(inp):
    if hasattr(inp, 'read'):
        yield inp

    elif inp.startswith('http'):
        # `tricky part`
        with contextlib.closing(urlopen(inp)) as f:
            yield f
        # `tricky part`

    else:
        pass

i am looking to discuss this as i a a novice in asyncio.

winter ravineBOT
#

@low pecan

Python help channel opened

Remember to:

  • Ask your Python question, not if you can ask or if there's an expert who can help.
  • Show a code sample as text (rather than a screenshot) and the error message, if you've got one.
  • Explain what you expect to happen and what actually happens.

:warning: Do not pip install anything that isn't related to your question, especially if asked to over DMs.

Closes after a period of inactivity, or when you send !close.
low pecan
#

i am confused about the contextlib.closing part

#

if inp is a url starting with http the code will try to open it and return the html as urllib.request.urlopen > HttpResponse

twin peak
#

urllib is not an async library though.

low pecan
#

yes i was thinking of useing aiohttp instead

#

would this work ?

twin peak
#

yes, aiohttp is async

low pecan
#

i am not sure how to rewrite this line with contextlib.closing(urlopen(inp)) as f:

#

and replaceurlopen with aiohttp.opon

twin peak
#

aiohttp.open is not a thing I think?

low pecan
# 
async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            # Ensure that the response status is successful
            response.raise_for_status()```
twin peak
#

using aiohttp is a bit of a pain, because you usually need two nestid async with statements.

async with aiohttp.ClientSession() as sess:
    async with sess.get(inp) as resp:
        yield resp
#

yea, using resp.raise_for_status() before the yield might make sense.

low pecan
#

we need to somehow do all this inside contextlib.closing

twin peak
#

no, you don't

#

the async with blocks close the session and response automatically when exiting

#

that's what the context managers are there for

low pecan
#

could you suggest the code changes ?

twin peak
#

I already did

#

though you probably have the issue, that different things return different kind of "streams"

low pecan
#

yes that is the problem i am trying to solve

#

but this line is confusing me. with contextlib.closing(urlopen(inp)) as f:

twin peak
#

not sure there is a good way to achieve that, except for maybe writing a custom wrapper for every type, that doesn't fit a specific protocol (duck type).

low pecan
#

so if it is a url we close the asynccontext manager

twin peak
#

contextlib.closing returns a context manager (not async), that closes whatever is passed in, when it exits the with-block.

#

(at least I believe it's not async.. might be wrong though)

#

not that it matters, as it's not needed if the thing you open already supports with or async with

#

(ah yea, there would be contextlib.aclosing for an async version - but again. not needed for aiohttp)

low pecan
twin peak
low pecan
#

now the problem is how to replace the urllib.urlopen with aiohttp

twin peak
low pecan
#

added a new function

async def async_urlopen(url):
    async with aiohttp.ClientSession() as sess:
        async with sess.get(url) as resp:
            yield resp

which i am using this way,

async with contextlib.aclosing(await async_urlopen(inp)) as f:
    yield f
twin peak
#

I'd also change this part:

    if hasattr(inp, 'read'):
        yield inp
``` the `inp.read` should be a coroutine function as well, for this whole thing to be async. Currently you only check if inp.read exists (which it does for regular open() calls - but those aren't async).
```py
import inspect
``` ```py
    if hasattr(inp, 'read'):
        if not inspect.iscoroutinefunction(inp.read):
            raise TypeError('read method is not a coroutine function.')
        yield inp
twin peak
spark hamlet
#

^

low pecan
#

okay. let me remove that

twin peak
#

just put those three lines I showed you there instead of the old with block.

low pecan
# 
        # tricky part
        # async with contextlib.aclosing(await async_urlopen(inp)) as f:
        #     yield f
        
        async with aiohttp.ClientSession() as sess:
            async with sess.get(inp) as f:
                yield f
        # tricky part```
twin peak
#

yea, that's pretty much all you need. f (the response), should have an async f.read() method as well (if I remember aiohttp correctly)

low pecan
#

so i have rewritten the openstream this way

# 
async def openstream(inp):
    if hasattr(inp, 'read'):
        if not inspect.iscoroutinefunction(inp.read):
            raise TypeError('read method is not a coroutine function.')
        yield inp
    elif os.path.isfile(inp):
        async with open(inp, mode='rb') as f:
            yield f
    elif inp.startswith('http'):
        async with aiohttp.ClientSession() as sess:
            async with sess.get(inp) as f:
                yield f

    elif isinstance(input, str) and input.startswith('data:'):
        parts = input.split(';', 2)
        if len(parts) == 2 and parts[1].startswith('base64,'):
            yield io.BytesIO(base64.b64decode(parts[1][7:]))
    else:
        pass
low pecan
twin peak
#

where does open(...) come from in that second if?

low pecan
#

from aiofiles import os, open

twin peak
#

I'd probably not import it like that, as it makes it confusing what os and open actually is in that case

low pecan
#

i have changed them to

# 
elif aiofiles.os.path.isfile(inp):
    async with aiofiles.open(inp, mode='rb') as f:
        yield f
twin peak
#

yea, and not sure about that isfile.. if it's from aiofiles, is it not an async function? shouldn't it be awaited?

low pecan
#

i think its from the regular std os library

twin peak
#

doesn't seem to be the case. I looked at the source.

#

in doubt: don't import os from aiofiles, but import os directly. not sure it's quite worth it to use async there

low pecan
#

yes. i have made the changes

# 
elif os.path.isfile(inp):
    async with aiofiles_open(inp, mode='rb') as f:
        yield f
#

from aiofiles import open as aiofiles_open

#

let me see that happens when i run it

twin peak
#

aopen is probably good enough 😄 - most people understand that.

low pecan
#

okay. i am new to async so still learning the naming conventions

twin peak
#

it's just that renaming it to aiofiles_open has the same letter count as aiofiles.open ... so... 😄

low pecan
#

yea. i noticed that

#

now i have to convert this last function to async and then i am done.

twin peak
#

io.BytesIO has a read() method, however, it's not async.. so it's incompatible with the other f that are yielded

low pecan
low pecan
twin peak
#

not sure it got that.

#

but the probe only uses stream.read (and you moslty only need to add an await in front of it)

#

so you could probably "monkey-patch" the io.BytesIO object

low pecan
twin peak
# 
import io
from functools import partial
import types

async def aread(self, read, n) -> bytes:
    return read(n)
    
def patch(f: io.BytesIO):
    f.read = partial(types.MethodType(aread, f), f.read)

async def main():
    f = io.BytesIO(b'abc')
    patch(f)

    print(await f.read(1))
    print(await f.read(1))
    print(await f.read(1))
``` I think this works
low pecan
# 
b'b'
b'c'

Process finished with exit code 0```
winter ravineBOT
#
Python help channel closed

This help channel has been closed and it's no longer possible to send messages here. If your question wasn't answered, feel free to create a new post in #1035199133436354600. To maximize your chances of getting a response, check out this guide on asking good questions.