Design pipeline ownership in multi-level Storage client

[lmazuel]

I was reviewing the Storage AIO PR, and found something that concerns me a little:

We have several client:

• BlobServiceClient
• BlobClient
• ContainerClient

From a BlobServiceClient, you can receive a BlobClient using “get_blob_client”. “get_blob_client” will create a BlobClient by sharing the same “pipeline”.

This means this code:

with BlobServiceClient(….) as bsc:
    with bsc.get_bloc_client() as bc:
        # pass

Will call twice the aenter and aexit on the transport. That already could be a problem, but if you do code still like this, that starts to become really unhealthy for you:

with BlobServiceClient(….) as bsc:
    with bsc.get_bloc_client() as bc:
        # pass
    bsc.get_account_information()  # transport closed!

Discussing with @johanste , we don't want reference count to enable scenario like "I create BlobServiceClient, from which I create a ContainerClient, from that I create the BlobServiceClient and it stills work".

Current proposal would be if you create a local client from an upper client, the upper client owns the pipeline (or the transport at least) and you need to close them in the correct order.

[lmazuel]

Looking at azure core, the fix would be to pass "session_owner=False" to transport of created client, with the same session from the main transport.

[mayurid]

@rakshith91 : Please take a look

[lmazuel]

Thinking about this a bit further, transport is supposed to be a black box, not we're not supposed to know how to instantiate a new transport. Also, if customer provide their own transport, we don't know at all how to create it.

Our problem is that we don't want the inner client to close the outer transport. We can achieve that with a wrapper that does not close:

class AsyncTransportWrapper(AsyncHttpTransport):

    def __init__(self, async_transport):
        self._transport = async_transport

    async def send(self, request, **kwargs):
        return await self._transport.send(request, **kwargs)

    async def open(self):
        pass

    async def close(self):
        pass

    async def __aexit__(self, *args):
        pass

[lmazuel]

Example of unittest:

transport = RequestsTransport()
with BlobServiceClient(transport=transport) as bsc:
    assert transport.session is not None
    with bsc.get_blob_client() as bc:
        assert transport.session is not None
    assert transport.session is not None  # Right now it's None

[kristapratico]

merged