Add async client transport cleanup

[jonathan343]

Overview

This PR adds deterministic resource cleanup to generated clients and HTTP transports. It also preserves the configured transport across operation-level config copies so clients reuse one session and connection pool.

Current State

Each operation deep copies the client config, including its transport. This can create temporary HTTP clients, sessions, and connection pools that are discarded without cleanup. Applications may see repeated Unclosed client session and Unclosed connector warnings during shutdown.

New Pattern

Generated clients and HTTP transports now support close() and async context managers. Operation config copies retain the client-owned transport instead of copying it.

async with AsyncExampleClient(config) as client:
    response = await client.some_operation(input)

Clients can also be closed manually:

client = AsyncExampleClient(config)
try:
    response = await client.some_operation(input)
finally:
    await client.close()

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[Alan4506]

Thanks @jonathan343! LGTM overall. Just have a non-blocking question:

AIOHTTPClient.close() and AWSCRTHTTPClient.close() release different things. The aiohttp one closes the ClientSession, while the CRT one closes and clears the pooled connections. I understand why: aiohttp only exposes session.close(), and a closed session cannot be reopened; CRT has no session equivalent, and connections are the only thing it can release.

But the problem is: the close() on a generated client now behaves differently per transport. For example, I verified the divergence with the following script:

import asyncio

from smithy_aws_core.identity import EnvironmentCredentialsResolver
from smithy_http.aio.aiohttp import AIOHTTPClient
from smithy_http.aio.crt import AWSCRTHTTPClient

from aws_sdk_apigateway.client import AsyncAPIGatewayClient
from aws_sdk_apigateway.config import Config
from aws_sdk_apigateway.models import GetAccountInput

REGION = "us-east-1"


async def check(label, transport):
    client = AsyncAPIGatewayClient(
        config=Config(
            endpoint_uri=f"https://apigateway.{REGION}.amazonaws.com",
            region=REGION,
            transport=transport,
            aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
        )
    )
    await client.get_account(GetAccountInput())
    print(f"{label}: call before close -> ok")
    await client.close()
    try:
        await client.get_account(GetAccountInput())
        print(f"{label}: call after close  -> ok (no error raised)")
    except Exception as e:
        print(f"{label}: call after close  -> {type(e).__name__}: {str(e)[:60]}")


async def main():
    await check("aiohttp", AIOHTTPClient())
    await check("crt    ", AWSCRTHTTPClient())


asyncio.run(main())

Output:

aiohttp: call before close -> ok
aiohttp: call after close  -> SmithyError: Session is closed
crt    : call before close -> ok
crt    : call after close  -> ok (no error raised)

Do you think this divergence is acceptable? Do we need to document the divergence with some comments?

[arandito]

Overall this PR looks good to me. Left a couple minor comments related to how we preserve config fields when we deepcopy.

[arandito]

[Non-blocking, general comment]
I'm noticing some inconsistency in how we are preserving certain config fields. Our retries strategies each define their own __deepcopy__ override to return the same instance. This works fine for the retry strategies we maintain but if customers provide their own custom implementations, they will need to always implement their own override.

This issue also arises for the config.aws_credentials_identity_resolver field. We need to preserve the same instance to actually reuse cached credentials. I like your approach of modifying the memo dict in the deepcopy() call. I will probably add a memo entry for the identity resolver field in my upcoming PR that integrates the chain into the client.

As a follow up, we should see if its possible to do all of this work inside the config object's own __deepcopy__. This should centralize the logic and ensure any retry strategy, transport, or identity resolver is preserved, plus any future fields.

We can wait to do this until we deprecate the existing Config object in favor of the new AsyncAwsConfig.

[arandito]

If we already codegen this:

config = deepcopy(
    self._config,
    {id(self._config.transport): self._config.transport},
)

Do we need to change the override to return itself?

[arandito]

Same comment at the AIOHTTPClient.__deepcopy__