add execution option for asyncpg_timeout

Describe the use case

In our company we use PostgreSQL, one of the tables that is quite big with over 1.5B records. We have few queries constructed with SQLAlchemy Core which we use on it and with help of indexes they're usually quite fast (100's of milliseconds at most). Lately we have discovered and issue with PG's query planner that would omit an index and would run a plan that was stalling the connection for multiple days. After running a particular API request many times our connection pool got exhausted and application basically crashed.

Natural solution to prevent this type of issues would be to implement some sort of timeout. Value of the timeout would depend on the expected execution time of given query. There are of course multiple ways to achieve this:

1. Set statement_timeout on current connection in case of PostgreSQL.
2. Wrap AsyncConnection.execute() in asyncio.wait_for().
3. Add option to pass timeout argument to asyncpg driver via for example execution_options argument.

Where 1. and 2. seems to be possible options I think it would be very useful if option 3. would be implemented in SQLAlchemy. Timeout mechanism is already built in in asyncpg driver but access to it would need to be added.

Databases / Backends / Drivers targeted

asyncpg for PostgreSQL

Example Use

try:
    row = await conn.execute(query, execution_options={'timeout': 3})
except TimeoutError:
    ... # handle case when query took to long to execute

Additional context

No response

Start in lib/sqlalchemy/dialects/postgresql/asyncpg.py at the asyncpg driver area linked in the issue, then trace how execution_options reach statement execution. Confirm the asyncpg timeout argument and its exception behavior before deciding how the option should be passed. Done means the requested execution option is supported for asyncpg PostgreSQL queries and its timeout behavior is covered by the relevant tests.