### Acquire and Setup Connection

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Acquires a connection, optionally runs a setup function, and returns a connection proxy. Handles exceptions during setup by closing the connection.

```python
        self._proxy = proxy = PoolConnectionProxy(self, self._con)

        if self._setup is not None:
            try:
                await self._setup(proxy)
            except (Exception, asyncio.CancelledError) as ex:
                # If a user-defined `setup` function fails, we don't
                # know if the connection is safe for re-use, hence
                # we close it.  A new connection will be created
                # when `acquire` is called again.
                try:
                    # Use `close()` to close the connection gracefully.
                    # An exception in `setup` isn't necessarily caused
                    # by an IO or a protocol error.  close() will
                    # do the necessary cleanup via _release_on_close().
                    await self._con.close()
                finally:
                    raise ex

        self._in_use = self._loop.create_future()

        return proxy
```

--------------------------------

### Pool Initialization with Custom Parameters

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Example of initializing an asyncpg Pool with various configuration parameters, including custom connection setup and reset coroutines.

```Python
return Pool(
        dsn,
        connection_class=connection_class,
        record_class=record_class,
        min_size=min_size,
        max_size=max_size,
        max_queries=max_queries,
        loop=loop,
        connect=connect,
        setup=setup,
        init=init,
        reset=reset,
        max_inactive_connection_lifetime=max_inactive_connection_lifetime,
        **connect_kwargs,
    )
```

--------------------------------

### Install asyncpg via pip

Source: https://magicstack.github.io/asyncpg/current/installation.html

Standard installation command for the asyncpg library.

```bash
$ pip install asyncpg
```

--------------------------------

### Build asyncpg from source

Source: https://magicstack.github.io/asyncpg/current/installation.html

Installs the package in editable mode from the root of the source checkout.

```bash
$ pip install -e .
```

--------------------------------

### Install asyncpg with GSSAPI/SSPI support

Source: https://magicstack.github.io/asyncpg/current/installation.html

Installs asyncpg with additional support for GSSAPI/SSPI authentication.

```bash
$ pip install 'asyncpg[gssauth]'
```

--------------------------------

### Create and Use Connection Pools

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Examples of creating a connection pool and performing operations using different patterns.

```python
    async with asyncpg.create_pool(user='postgres',
                                   command_timeout=60) as pool:
        await pool.fetch('SELECT 1')
```

```python
    async with asyncpg.create_pool(user='postgres',
                                   command_timeout=60) as pool:
        async with pool.acquire() as con:
            await con.execute('''
               CREATE TABLE names (
                  id serial PRIMARY KEY,
                  name VARCHAR (255) NOT NULL)
            ''')
            await con.fetch('SELECT 1')
```

```python
    pool = await asyncpg.create_pool(user='postgres', command_timeout=60)
    con = await pool.acquire()
    try:
        await con.fetch('SELECT 1')
    finally:
        await pool.release(con)
```

--------------------------------

### Setup Inactivity Callback

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Sets up a timer to track connection inactivity. Raises an error if a timer already exists.

```python
    def _setup_inactive_callback(self) -> None:
        if self._inactive_callback is not None:
            raise exceptions.InternalClientError(
                'pool connection inactivity timer already exists')

        if self._max_inactive_time:
            self._inactive_callback = self._pool._loop.call_later(
                self._max_inactive_time, self._deactivate_inactive_connection)
```

--------------------------------

### Execute asyncpg test suite

Source: https://magicstack.github.io/asyncpg/current/installation.html

Runs the project test suite; requires PostgreSQL to be installed.

```bash
$ python setup.py test
```

--------------------------------

### Register a custom type codec

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Example demonstrating how to set an encoder and decoder for a custom data type using set_type_codec.

```python
>>> import asyncpg
>>> import asyncio
>>> import datetime
>>> from dateutil.relativedelta import relativedelta
>>> async def run():
...     con = await asyncpg.connect(user='postgres')
...     def encoder(delta):
...         ndelta = delta.normalized()
...         return (ndelta.years * 12 + ndelta.months,
...                 ndelta.days,
...                 ((ndelta.hours * 3600 +
...                    ndelta.minutes * 60 +
...                    ndelta.seconds) * 1000000 +
...                  ndelta.microseconds))
...     def decoder(tup):
...         return relativedelta(months=tup[0], days=tup[1],
```

--------------------------------

### Get Execution Plan with EXPLAIN

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/prepared_stmt.html

Retrieve the execution plan of a prepared statement using the SQL EXPLAIN command. Set 'analyze' to True to execute the statement and include runtime statistics.

```python
query = 'EXPLAIN (FORMAT JSON, VERBOSE'
        if analyze:
            query += ', ANALYZE) '
        else:
            query += ') '
        query += self._state.query
```

--------------------------------

### Web Service with Asyncpg Connection Pool

Source: https://magicstack.github.io/asyncpg/current/usage.html

This example demonstrates a simple web service using aiohttp and asyncpg to create a connection pool, handle requests, acquire connections, and execute database queries. Ensure the database and user are correctly configured.

```python
import asyncio
import asyncpg
from aiohttp import web


async def handle(request):
    """Handle incoming requests."""
    pool = request.app['pool']
    power = int(request.match_info.get('power', 10))

    # Take a connection from the pool.
    async with pool.acquire() as connection:
        # Open a transaction.
        async with connection.transaction():
            # Run the query passing the request argument.
            result = await connection.fetchval('select 2 ^ $1', power)
            return web.Response(
                text="2 ^ {} is {}".format(power, result))


async def init_db(app):
    """Initialize a connection pool."""
     app['pool'] = await asyncpg.create_pool(database='postgres',
                                             user='postgres')
     yield
     await app['pool'].close()


def init_app():
    """Initialize the application server."""
    app = web.Application()
    # Create a database context
    app.cleanup_ctx.append(init_db)
    # Configure service routes
    app.router.add_route('GET', '/{power:\d+}', handle)
    app.router.add_route('GET', '/', handle)
    return app


app = init_app()
web.run_app(app)

```

--------------------------------

### Build asyncpg with debug mode

Source: https://magicstack.github.io/asyncpg/current/installation.html

Installs the package in editable mode with additional runtime checks enabled via the ASYNCPG_DEBUG environment variable.

```bash
$ env ASYNCPG_DEBUG=1 pip install -e .
```

--------------------------------

### Get Prepared Statement Parameters

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/prepared_stmt.html

Retrieve a description of the parameter types for a prepared statement. This helps in understanding the expected input for the statement.

```python
stmt = await connection.prepare('SELECT ($1::int, $2::text)')
print(stmt.get_parameters())
```

--------------------------------

### Pool Configuration Parameters

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Details on parameters for configuring an asyncpg connection pool, including connection reset and type codec setup.

```APIDOC
## Pool Configuration

This section describes the parameters used when creating an `asyncpg.pool.Pool` instance.

### Parameters

- **dsn** (string) - The database connection string.
- **connection_class** (class) - The connection class to use. Defaults to `asyncpg.connection.Connection`.
- **record_class** (class) - The record class to use. Defaults to `asyncpg.Record`.
- **min_size** (integer) - The minimum number of connections in the pool.
- **max_size** (integer) - The maximum number of connections in the pool.
- **max_queries** (integer) - The maximum number of queries to run on a connection before it is closed and replaced.
- **loop** (asyncio.BaseEventLoop) - The asyncio event loop instance. If `None`, the default event loop will be used.
- **connect** (coroutine) - A coroutine to run after a connection is established. It receives the connection object as an argument.
- **setup** (coroutine) - A coroutine to run after a connection is established and before it is returned to the pool. It receives the connection object as an argument.
- **init** (coroutine) - A coroutine to run after a connection is established. It receives the connection object as an argument.
- **reset** (coroutine) - A coroutine to reset a connection before it is returned to the pool by `Pool.release()`. The function is supposed to reset any changes made to the database session so that the next acquirer gets the connection in a well-defined state. The default implementation calls `Connection.reset()`, which runs `SELECT pg_advisory_unlock_all(); CLOSE ALL; UNLISTEN *; RESET ALL;`.
- **max_inactive_connection_lifetime** (float) - The maximum time in seconds a connection can be inactive before it is closed.
- **connect_kwargs** (dict) - Additional keyword arguments to pass to the connection constructor.

### Type Codec Setup

An example use case would be to setup type codecs with:

- `:meth:`Connection.set_builtin_type_codec()`
- `:meth:`Connection.set_type_codec()`

### Connection Reset Behavior

The default `reset` coroutine performs the following actions:

```sql
SELECT pg_advisory_unlock_all();
CLOSE ALL;
UNLISTEN *;
RESET ALL;
```

The exact reset query is determined by detected server capabilities. A custom `reset` implementation can obtain the default query by calling `:meth:`Connection.get_reset_query()`.

### Version Changes

- **0.10.0**: An `~asyncpg.exceptions.InterfaceError` will be raised on any attempted operation on a released connection.
- **0.13.0**: An `~asyncpg.exceptions.InterfaceError` will be raised on any attempted operation on a prepared statement or a cursor created on a connection that has been released to the pool.
- **0.13.0**: An `~asyncpg.exceptions.InterfaceWarning` will be produced if there are any active listeners present on the connection at the moment of its release to the pool.
- **0.22.0**: Added the `record_class` parameter.
- **0.30.0**: Added the `connect` and `reset` parameters.
```

--------------------------------

### Get Server Version

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Retrieves the version of the connected PostgreSQL server. The returned value is a named tuple similar to sys.version_info.

```python
>>> con.get_server_version()
ServerVersion(major=9, minor=6, micro=1,
              releaselevel='final', serial=0)
```

--------------------------------

### Format IPv6 DSN URI

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Example of a valid connection URI containing an IPv6 address enclosed in square brackets.

```text
postgres://dbuser@[fe80::1ff:fe23:4567:890a%25eth0]/dbname
```

--------------------------------

### Initialize Cursor State

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/cursor.html

Initializes the cursor's internal state by getting a prepared statement from the connection. This method should be called before other cursor operations.

```python
async def _init(self, timeout):
        if self._state is None:
            self._state = await self._connection._get_statement(
                self._query,
                timeout,
                named=True,
                record_class=self._record_class,
            )
            self._state.attach()
        self._check_ready()
        await self._bind(timeout)
        return self
```

--------------------------------

### Get Server Version

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Returns the version of the connected PostgreSQL server as a named tuple. The tuple includes major, minor, and micro version numbers, as well as the release level and serial number, similar to sys.version_info.

```python
return self._server_version
```

--------------------------------

### Prepare and execute a query

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Demonstrates creating a prepared statement using Connection.prepare() and executing it multiple times.

```python
>>> import asyncpg, asyncio
>>> async def run():
...     conn = await asyncpg.connect()
...     stmt = await conn.prepare('''SELECT 2 ^ $1''')
...     print(await stmt.fetchval(10))
...     print(await stmt.fetchval(20))
...
>>> asyncio.run(run())
1024.0
1048576.0
```

--------------------------------

### Connection Initialization

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Entry point for establishing a new database connection.

```python
[docs]
async def connect(dsn=None, *,
                  host=None, port=None,
                  user=None, password=None, passfile=None,
                  service=None,
```

--------------------------------

### Start Transaction or Savepoint

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/transaction.html

Starts a transaction or a savepoint. Handles nested transactions and isolation level checks. Raises InterfaceError if the transaction is already started or if a nested transaction has a different isolation level.

```python
@connresource.guarded
    async def start(self):
        """Enter the transaction or savepoint block."""
        self.__check_state_base('start')
        if self._state is TransactionState.STARTED:
            raise apg_errors.InterfaceError(
                'cannot start; the transaction is already started')

        con = self._connection

        if con._top_xact is None:
            if con._protocol.is_in_transaction():
                raise apg_errors.InterfaceError(
                    'cannot use Connection.transaction() in '
                    'a manually started transaction')
            con._top_xact = self
        else:
            # Nested transaction block
            if self._isolation:
                top_xact_isolation = con._top_xact._isolation
                if top_xact_isolation is None:
                    top_xact_isolation = ISOLATION_LEVELS_BY_VALUE[
                        await self._connection.fetchval(
                            'SHOW transaction_isolation;')]
                if self._isolation != top_xact_isolation:
                    raise apg_errors.InterfaceError(
                        'nested transaction has a different isolation level: '
                        'current {!r} != outer {!r}'.format(
                            self._isolation, top_xact_isolation))
            self._nested = True

        if self._nested:
            self._id = con._get_unique_id('savepoint')
            query = 'SAVEPOINT {};'.format(self._id)
        else:
            query = 'BEGIN'
            if self._isolation == 'read_committed':
                query += ' ISOLATION LEVEL READ COMMITTED'
            elif self._isolation == 'read_uncommitted':
                query += ' ISOLATION LEVEL READ UNCOMMITTED'
            elif self._isolation == 'repeatable_read':
                query += ' ISOLATION LEVEL REPEATABLE READ'
            elif self._isolation == 'serializable':
                query += ' ISOLATION LEVEL SERIALIZABLE'
            if self._readonly:
                query += ' READ ONLY'
            if self._deferrable:
                query += ' DEFERRABLE'
            query += ';'

        try:
            await self._connection.execute(query)
        except BaseException:
```

--------------------------------

### Connect and execute queries with asyncpg

Source: https://magicstack.github.io/asyncpg/current/usage.html

Establishes a database connection and demonstrates basic table creation, insertion, and row fetching using positional arguments.

```python
import asyncio
import asyncpg
import datetime

async def main():
    # Establish a connection to an existing database named "test"
    # as a "postgres" user.
    conn = await asyncpg.connect('postgresql://postgres@localhost/test')
    # Execute a statement to create a new table.
    await conn.execute('''
        CREATE TABLE users(
            id serial PRIMARY KEY,
            name text,
            dob date
        )
    ''')

    # Insert a record into the created table.
    await conn.execute('''
        INSERT INTO users(name, dob) VALUES($1, $2)
    ''', 'Bob', datetime.date(1984, 3, 1))

    # Select a row from the table.
    row = await conn.fetchrow(
        'SELECT * FROM users WHERE name = $1', 'Bob')
    # *row* now contains
    # asyncpg.Record(id=1, name='Bob', dob=datetime.date(1984, 3, 1))

    # Close the connection.
    await conn.close()

asyncio.run(main())
```

--------------------------------

### GET /connection/reset-query

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Retrieves the SQL query string used to reset a connection state before it is returned to the connection pool.

```APIDOC
## GET /connection/reset-query

### Description
Returns the query sent to the PostgreSQL server on connection release. This is used by the pool to ensure a clean state for the next acquirer.

### Method
GET

### Endpoint
get_reset_query()

### Response
#### Success Response (200)
- **query** (string) - The concatenated SQL commands (e.g., RESET ALL, UNLISTEN *, etc.) based on server capabilities.
```

--------------------------------

### Establish a basic database connection

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Connect to a PostgreSQL database and perform a simple query.

```python
>>> import asyncpg
>>> import asyncio
>>> async def run():
...     con = await asyncpg.connect(user='postgres')
...     types = await con.fetch('SELECT * FROM pg_type')
...     print(types)
...
>>> asyncio.run(run())
[<Record typname='bool' typnamespace=11 ... 
```

--------------------------------

### Configure SSL with verify-full

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Programmatically configure an SSL context equivalent to sslmode=verify-full with client certificate authentication.

```python
>>> import asyncpg
>>> import asyncio
>>> import ssl
>>> async def main():
...     # Load CA bundle for server certificate verification,
...     # equivalent to sslrootcert= in DSN.
...     sslctx = ssl.create_default_context(
...         ssl.Purpose.SERVER_AUTH,
...         cafile="path/to/ca_bundle.pem")
...     # If True, equivalent to sslmode=verify-full, if False:
...     # sslmode=verify-ca.
...     sslctx.check_hostname = True
...     # Load client certificate and private key for client
...     # authentication, equivalent to sslcert= and sslkey= in
...     # DSN.
...     sslctx.load_cert_chain(
...         "path/to/client.cert",
...         keyfile="path/to/client.key",
...     )
...     con = await asyncpg.connect(user='postgres', ssl=sslctx)
...     await con.close()
>>> asyncio.run(main())
```

--------------------------------

### Get Prepared Statement Query

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/prepared_stmt.html

Retrieve the SQL query text associated with a prepared statement. This is useful for debugging or logging.

```python
stmt = await connection.prepare('SELECT $1::int')
assert stmt.get_query() == "SELECT $1::int"
```

--------------------------------

### Acquire a connection from the pool

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Demonstrates how to acquire a connection using an async context manager or manual release.

```python
            async with pool.acquire() as con:
                await con.execute(...)
```

```python
            con = await pool.acquire()
            try:
                await con.execute(...)
            finally:
                await pool.release(con)
```

--------------------------------

### Create and use a connection pool

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Demonstrates using a connection pool with an async context manager for automatic resource cleanup.

```python
async with asyncpg.create_pool(user='postgres',
                               command_timeout=60) as pool:
    await pool.fetch('SELECT 1')
```

```python
async with asyncpg.create_pool(user='postgres',
                               command_timeout=60) as pool:
    async with pool.acquire() as con:
        await con.execute('''
           CREATE TABLE names (
              id serial PRIMARY KEY,
              name VARCHAR (255) NOT NULL)
        ''')
        await con.fetch('SELECT 1')
```

```python
pool = await asyncpg.create_pool(user='postgres', command_timeout=60)
con = await pool.acquire()
try:
    await con.fetch('SELECT 1')
finally:
    await pool.release(con)
```

--------------------------------

### Transaction Context Manager Entry

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/transaction.html

Enters the transaction context, marking it as managed and starting the transaction. Raises InterfaceError if already in a block.

```python
    async def __aenter__(self):
        if self._managed:
            raise apg_errors.InterfaceError(
                'cannot enter context: already in an `async with` block')
        self._managed = True
        await self.start()
```

--------------------------------

### Initialize asyncpg Connection

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Initializes a new Connection instance with protocol, transport, loop, configuration, and parameters. Debug mode enables source traceback logging.

```python
def __init__(self, protocol, transport, loop,
                 addr,
                 config: connect_utils._ClientConfiguration,
                 params: connect_utils._ConnectionParameters):
        self._protocol = protocol
        self._transport = transport
        self._loop = loop
        self._top_xact = None
        self._aborted = False
        # Incremented every time the connection is released back to a pool.
        # Used to catch invalid references to connection-related resources
        # post-release (e.g. explicit prepared statements).
        self._pool_release_ctr = 0

        self._addr = addr
        self._config = config
        self._params = params

        self._stmt_cache = _StatementCache(
            loop=loop,
            max_size=config.statement_cache_size,
            on_remove=functools.partial(
                _weak_maybe_gc_stmt, weakref.ref(self)),
            max_lifetime=config.max_cached_statement_lifetime)

        self._stmts_to_close = set()
        self._stmt_cache_enabled = config.statement_cache_size > 0

        self._listeners = {}
        self._log_listeners = set()
        self._cancellations = set()
        self._termination_listeners = set()
        self._query_loggers = set()

        settings = self._protocol.get_settings()
        ver_string = settings.server_version
        self._server_version = \
            serverversion.split_server_version_string(ver_string)

        self._server_caps = _detect_server_capabilities(
            self._server_version, settings)

        if self._server_version < (14, 0):
            self._intro_query = introspection.INTRO_LOOKUP_TYPES_13
        else:
            self._intro_query = introspection.INTRO_LOOKUP_TYPES

        self._reset_query = None
        self._proxy = None

        # Used to serialize operations that might involve anonymous
        # statements.  Specifically, we want to make the following
        # operation atomic:
        #    ("prepare an anonymous statement", "use the statement")
        #
        # Used for `con.fetchval()`, `con.fetch()`, `con.fetchrow()`,
        # `con.execute()`, and `con.executemany()`.
        self._stmt_exclusive_section = _Atomic()

        if loop.get_debug():
            self._source_traceback = _extract_stack()
        else:
            self._source_traceback = None
```

--------------------------------

### asyncpg.connect

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Establishes a connection to a PostgreSQL database and returns a Connection instance.

```APIDOC
## asyncpg.connect

### Description
Establishes a connection to a PostgreSQL database. Returns a Connection instance.

### Parameters
#### Request Body
- **max_cacheable_statement_size** (int) - Optional - The maximum size of a statement that can be cached (15KiB by default). Pass 0 to allow all statements.
- **command_timeout** (float) - Optional - The default timeout for operations on this connection.
- **ssl** (bool/ssl.SSLContext/str) - Optional - SSL configuration. Values: 'disable', 'prefer', 'allow', 'require', 'verify-ca', 'verify-full'.
- **direct_tls** (bool) - Optional - Skip PostgreSQL STARTTLS mode and perform a direct SSL connection.
- **server_settings** (dict) - Optional - An optional dict of server runtime parameters.
- **connection_class** (type) - Optional - Class of the returned connection object. Must be a subclass of Connection.
- **record_class** (type) - Optional - Class to use for records returned by queries. Must be a subclass of Record.
- **target_session_attrs** (SessionAttribute) - Optional - Check that the host has the correct attribute ('any', 'primary', 'standby', 'read-write', 'read-only', 'prefer-standby').
- **krbsrvname** (str) - Optional - Kerberos service name to use when authenticating with GSSAPI.
- **gsslib** (str) - Optional - GSS library to use for GSSAPI/SSPI authentication.

### Response
#### Success Response (200)
- **Connection** (instance) - A Connection instance.
```

--------------------------------

### Manual Transaction Control

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Use this when you need explicit control over starting, committing, or rolling back transactions. Requires manual error handling.

```python
tr = connection.transaction()
await tr.start()
try:
    ...
except:
    await tr.rollback()
    raise
else:
    await tr.commit()
```

--------------------------------

### Acquire a connection from the pool

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Demonstrates two ways to acquire and use a connection from the pool: using an async context manager or manual acquisition and release.

```python
async with pool.acquire() as con:
    await con.execute(...)
```

```python
con = await pool.acquire()
try:
    await con.execute(...)
finally:
    await pool.release(con)
```

--------------------------------

### asyncpg.connect

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Establishes a connection to a PostgreSQL server.

```APIDOC
## asyncpg.connect

### Description
A coroutine to establish a connection to a PostgreSQL server. The connection parameters may be specified either as a connection URI in dsn, or as specific keyword arguments.

### Parameters
#### Keyword Arguments
- **dsn** (string) - Optional - Connection URI in libpq format.
- **host** (string/sequence) - Optional - Database host address or Unix-domain socket path.
- **port** (int/sequence) - Optional - Port number or socket file extension.
- **user** (string) - Optional - Database role name.
- **database** (string) - Optional - Name of the database.
- **password** (string/callable) - Optional - Password for authentication.
- **passfile** (string) - Optional - Path to password file.
- **service** (string) - Optional - Postgres connection service name.
- **servicefile** (string) - Optional - Path to connection service file.
- **loop** (asyncio.AbstractEventLoop) - Optional - Event loop instance.
- **timeout** (float) - Optional - Connection timeout in seconds.
- **statement_cache_size** (int) - Optional - Size of prepared statement LRU cache.
- **max_cached_statement_lifetime** (int) - Optional - Max time in seconds a statement stays in cache.

### Response
- **Connection** (object) - Returns a new Connection object.
```

--------------------------------

### Commit Transaction

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/transaction.html

Asynchronously commits the current transaction. Handles nested transactions by releasing savepoints. Raises an error if the transaction is not in a started state.

```python
async def __commit(self):
        self.__check_state('commit')

        if self._connection._top_xact is self:
            self._connection._top_xact = None

        if self._nested:
            query = 'RELEASE SAVEPOINT {};'.format(self._id)
        else:
            query = 'COMMIT;'

        try:
            await self._connection.execute(query)
        except BaseException:
            self._state = TransactionState.FAILED
            raise
        else:
            self._state = TransactionState.COMMITTED
```

--------------------------------

### Initialize Pool Connections

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Internal method to set up the connection queue and pre-connect the minimum required connections.

```python
    async def _initialize(self):
        self._queue = asyncio.LifoQueue(maxsize=self._maxsize)
        for _ in range(self._maxsize):
            ch = PoolConnectionHolder(
                self,
                max_queries=self._max_queries,
                max_inactive_time=self._max_inactive_connection_lifetime,
                setup=self._setup)

            self._holders.append(ch)
            self._queue.put_nowait(ch)

        if self._minsize:
            # Since we use a LIFO queue, the first items in the queue will be
            # the last ones in `self._holders`.  We want to pre-connect the
            # first few connections in the queue, therefore we want to walk
            # `self._holders` in reverse.

            # Connect the first connection holder in the queue so that
            # any connection issues are visible early.
            first_ch = self._holders[-1]  # type: PoolConnectionHolder
            await first_ch.connect()

            if self._minsize > 1:
                connect_tasks = []
                for i, ch in enumerate(reversed(self._holders[:-1])):
                    # `minsize - 1` because we already have first_ch
                    if i >= self._minsize - 1:
                        break
                    connect_tasks.append(ch.connect())

                await asyncio.gather(*connect_tasks)
```

--------------------------------

### Connection.transaction()

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Creates a transaction or savepoint block. Supports context manager usage for automatic commit/rollback or manual control via start(), commit(), and rollback().

```APIDOC
## Connection.transaction()

### Description
Creates a new Transaction object. When used as an asynchronous context manager, it automatically starts a transaction upon entering the block and commits or rolls back upon exiting.

### Methods
- **start()**: Enters the transaction or savepoint block.
- **commit()**: Exits the block and commits changes.
- **rollback()**: Exits the block and rolls back changes.

### Usage Examples

#### Context Manager
```python
async with connection.transaction():
    await connection.execute("INSERT INTO mytable VALUES(1, 2, 3)")
```

#### Nested Transactions (Savepoints)
```python
async with connection.transaction():
    async with connection.transaction():
        await connection.execute('INSERT INTO mytab (a) VALUES (1)')
        raise Exception # Automatically rolls back the nested transaction
```

#### Manual Control
```python
tr = connection.transaction()
await tr.start()
try:
    await connection.execute('...')
except:
    await tr.rollback()
else:
    await tr.commit()
```
```

--------------------------------

### Create a cursor from a prepared statement

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Shows how to create a cursor from a prepared statement to handle parameterized queries.

```python
async def iterate(con: Connection):
    # Create a prepared statement that will accept one argument
    stmt = await con.prepare('SELECT generate_series(0, $1)')

    async with con.transaction():
        # Postgres requires non-scrollable cursors to be created
        # and used in a transaction.

        # Execute the prepared statement passing `10` as the
        # argument -- that will generate a series or records
        # from 0..10.  Iterate over all of them and print every
        # record.
        async for record in stmt.cursor(10):
            print(record)
```

--------------------------------

### Connect to a PostgreSQL database

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Establishes a connection to a PostgreSQL database and executes a query to fetch records.

```pycon
>>> import asyncpg
>>> import asyncio
>>> async def run():
...     con = await asyncpg.connect(user='postgres')
...     types = await con.fetch('SELECT * FROM pg_type')
...     print(types)
...
>>> asyncio.run(run())
```

--------------------------------

### Get Connection Settings

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Retrieves the connection settings for the current connection. This returns an object of type asyncpg.ConnectionSettings, which encapsulates various parameters related to the connection.

```python
return self._protocol.get_settings()
```

--------------------------------

### Transaction Management

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Methods for managing database transactions, including starting new transactions with specific isolation levels and checking if the connection is currently within a transaction.

```APIDOC
## Transaction Management

### `transaction(isolation='read_committed', readonly=False, deferrable=False)`

Starts a new transaction with the specified isolation level, read-only status, and deferrability.

**Parameters**
- **isolation** (str) - Optional - Transaction isolation mode. Can be one of: `'serializable'`, `'repeatable_read'`, `'read_uncommitted'`, `'read_committed'`. Defaults to `'read_committed'` if not specified.
- **readonly** (bool) - Optional - Specifies whether the transaction is read-only.
- **deferrable** (bool) - Optional - Specifies whether the transaction is deferrable.

### `is_in_transaction()`

Checks if the connection is currently inside a transaction.

**Returns**
- bool: True if inside a transaction, False otherwise.
```

--------------------------------

### Rollback Transaction

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/transaction.html

Asynchronously rolls back the current transaction. Handles nested transactions by rolling back to a savepoint. Raises an error if the transaction is not in a started state.

```python
async def __rollback(self):
        self.__check_state('rollback')

        if self._connection._top_xact is self:
            self._connection._top_xact = None

        if self._nested:
            query = 'ROLLBACK TO {};'.format(self._id)
        else:
            query = 'ROLLBACK;'

        try:
            await self._connection.execute(query)
        except BaseException:
            self._state = TransactionState.FAILED
            raise
        else:
            self._state = TransactionState.ROLLEDBACK
```

--------------------------------

### Pool Connection Acquisition

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Demonstrates how to acquire a database connection from the pool using async with or await.

```APIDOC
## acquire(timeout=None)

### Description
Acquire a database connection from the pool.

### Method
GET

### Endpoint
/pool/acquire

### Parameters
#### Query Parameters
- **timeout** (float) - Optional - A timeout for acquiring a Connection.

### Request Example
```python
async with pool.acquire() as con:
    await con.execute(...)
```

Or:

```python
con = await pool.acquire()
try:
    await con.execute(...)
finally:
    await pool.release(con)
```

### Response
#### Success Response (200)
- **connection** (:class:`~asyncpg.connection.Connection`) - An instance of the acquired connection.
```

--------------------------------

### Get Server PID

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Retrieves the Process ID (PID) of the PostgreSQL server process to which the current connection is established. This can be useful for correlating connection activity with server-side processes.

```python
return self._protocol.get_server_pid()
```

--------------------------------

### Get Prepared Statement Attributes

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/prepared_stmt.html

Retrieve a description of the relation attributes (columns) for a prepared statement's result set. This is useful for understanding the structure of the returned data.

```python
st = await self.con.prepare('''
                SELECT typname, typnamespace FROM pg_type
            ''')
print(st.get_attributes())
```

--------------------------------

### Create New Connection

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/pool.html

Internal method to establish a new connection and execute the initialization callback.

```python
    async def _get_new_connection(self):
        con = await self._connect(
            *self._connect_args,
            loop=self._loop,
            connection_class=self._connection_class,
            record_class=self._record_class,
            **self._connect_kwargs,
        )
        if not isinstance(con, self._connection_class):
            good = self._connection_class
            good_n = f'{good.__module__}.{good.__name__}'
            bad = type(con)
            if bad.__module__ == "builtins":
                bad_n = bad.__name__
            else:
                bad_n = f'{bad.__module__}.{bad.__name__}'
            raise exceptions.InterfaceError(
                "expected pool connect callback to return an instance of "
                f"'{good_n}', got " f"'{bad_n}'"
            )

        if self._init is not None:
            try:
                await self._init(con)
            except (Exception, asyncio.CancelledError) as ex:
                # If a user-defined `init` function fails, we don't
                # know if the connection is safe for re-use, hence
                # we close it.  A new connection will be created
                # when `acquire` is called again.
                try:
                    # Use `close()` to close the connection gracefully.
                    # An exception in `init` isn't necessarily caused
                    # by an IO or a protocol error.  close() will
```

--------------------------------

### Configure SSL with require

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Programmatically configure an SSL context equivalent to sslmode=require, which skips server certificate and host verification.

```python
>>> import asyncpg
>>> import asyncio
>>> import ssl
>>> async def main():
...     sslctx = ssl.create_default_context(
...         ssl.Purpose.SERVER_AUTH)
...     sslctx.check_hostname = False
...     sslctx.verify_mode = ssl.CERT_NONE
...     con = await asyncpg.connect(user='postgres', ssl=sslctx)
...     await con.close()
>>> asyncio.run(main())
```

--------------------------------

### Get Prepared Statement Status Message

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/prepared_stmt.html

Retrieve the status message from the last executed command of a prepared statement. This typically indicates the success or outcome of operations like CREATE, INSERT, UPDATE, DELETE.

```python
stmt = await connection.prepare('CREATE TABLE mytab (a int)')
await stmt.fetch()
assert stmt.get_statusmsg() == "CREATE TABLE"
```

--------------------------------

### acquire()

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Acquire a database connection from the pool.

```APIDOC
## acquire(timeout=None)

### Description
Acquire a database connection from the pool.

### Parameters
#### Query Parameters
- **timeout** (float) - Optional - A timeout for acquiring a Connection.

### Response
- **Connection** (object) - An instance of Connection.
```

--------------------------------

### Establish PostgreSQL Connection

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

This section describes how to establish a connection to a PostgreSQL server using the asyncpg library. You can specify connection parameters via a DSN string or individual keyword arguments.

```APIDOC
## POST /connect

### Description
A coroutine to establish a connection to a PostgreSQL server.

### Method
POST

### Endpoint
/connect

### Parameters
#### Query Parameters
- **dsn** (string) - Optional - Connection arguments specified using as a single string in the `libpq connection URI format`_.
- **host** (string) - Optional - Database host address.
- **port** (integer) - Optional - Port number to connect to at the server host.
- **user** (string) - Optional - The name of the database role used for authentication.
- **database** (string) - Optional - The name of the database to connect to.
- **password** (string or callable) - Optional - Password to be used for authentication.
- **passfile** (string) - Optional - The name of the file used to store passwords.
- **service** (string) - Optional - The name of the postgres connection service stored in the postgres connection service file.
- **timeout** (integer) - Optional - Connection timeout in seconds. Defaults to 60.
- **statement_cache_size** (integer) - Optional - The maximum number of statements to cache. Defaults to 100.
- **max_cached_statement_lifetime** (integer) - Optional - The maximum lifetime in seconds for cached statements. Defaults to 300.
- **max_cacheable_statement_size** (integer) - Optional - The maximum size in bytes for a statement to be cacheable. Defaults to 15360.
- **command_timeout** (integer) - Optional - The timeout in seconds for executing commands.
- **ssl** (boolean or object) - Optional - SSL configuration.
- **direct_tls** (boolean) - Optional - Whether to use direct TLS connection.
- **connection_class** (class) - Optional - The connection class to use. Defaults to `asyncpg.connection.Connection`.
- **record_class** (class) - Optional - The record class to use. Defaults to `protocol.Record`.
- **server_settings** (dict) - Optional - Dictionary of server settings to apply.
- **target_session_attrs** (string) - Optional - Target session attributes.
- **krbsrvname** (string) - Optional - Kerberos service name.
- **gsslib** (string) - Optional - GSSAPI library to use.

### Request Example
```json
{
  "dsn": "postgres://user:password@host:port/database?option=value",
  "timeout": 30
}
```

### Response
#### Success Response (200)
- **connection** (object) - A new `asyncpg.connection.Connection` object.

#### Response Example
```json
{
  "connection": "<asyncpg.connection.Connection object at 0x...>"
}
```
```

--------------------------------

### Cursor with Prepared Statements

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Illustrates creating and iterating over a cursor from a prepared statement. This allows for efficient execution of parameterized queries.

```APIDOC
## PUT /api/users/{id}

### Description
Updates an existing user's information.

### Method
PUT

### Endpoint
/api/users/{id}

### Parameters
#### Path Parameters
- **id** (integer) - Required - The unique identifier of the user to update.

#### Request Body
- **username** (string) - Optional - The new username for the account.
- **email** (string) - Optional - The new email address for the account.

### Request Example
{
  "email": "john.doe.updated@example.com"
}

### Response
#### Success Response (200)
- **id** (integer) - The unique identifier for the updated user.
- **username** (string) - The updated username.
- **email** (string) - The updated email address.
- **updated_at** (string) - The timestamp when the user was last updated.

#### Response Example
{
  "id": 101,
  "username": "john_doe",
  "email": "john.doe.updated@example.com",
  "updated_at": "2023-10-27T11:00:00Z"
}
```

--------------------------------

### Set Custom Type Codec for Interval Type

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Example of setting a custom codec for the PostgreSQL 'interval' type using a tuple format. This allows for custom encoding and decoding of interval values. Requires dateutil library.

```pycon
>>> import asyncpg
>>> import asyncio
>>> import datetime
>>> from dateutil.relativedelta import relativedelta
>>> async def run():
...     con = await asyncpg.connect(user='postgres')
...     def encoder(delta):
...         ndelta = delta.normalized()
...         return (ndelta.years * 12 + ndelta.months,
...                 ndelta.days,
...                 ((ndelta.hours * 3600 +
...                    ndelta.minutes * 60 +
...                    ndelta.seconds) * 1000000 +
...                  ndelta.microseconds))
...     def decoder(tup):
...         return relativedelta(months=tup[0], days=tup[1],
...                              microseconds=tup[2])
...     await con.set_type_codec(
...         'interval', schema='pg_catalog', encoder=encoder,
...         decoder=decoder, format='tuple')
...     result = await con.fetchval(
...         "SELECT '2 years 3 mons 1 day'::interval")
...     print(result)
...     print(datetime.datetime(2002, 1, 1) + result)
... 
>>> asyncio.run(run())
```

--------------------------------

### Connection Class Overview

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/connection.html

Provides an overview of the Connection class, its initialization, and its role as a database session representation.

```APIDOC
## Connection Class

### Description
A representation of a database session.

Connections are created by calling :func:`~asyncpg.connection.connect`.

### Attributes
- `_protocol`: Internal protocol handler.
- `_transport`: Network transport layer.
- `_loop`: The asyncio event loop.
- `_top_xact`: The top-level transaction object.
- `_aborted`: Flag indicating if the connection is aborted.
- `_pool_release_ctr`: Counter for connection releases to a pool.
- `_stmt_cache`: Cache for prepared statements.
- `_stmts_to_close`: Set of statements to be closed.
- `_stmt_cache_enabled`: Boolean indicating if statement caching is enabled.
- `_listeners`: Dictionary for event listeners.
- `_server_version`: The version of the PostgreSQL server.
- `_server_caps`: Detected server capabilities.
- `_intro_query`: Query used for type introspection.
- `_reset_query`: Query used to reset the connection state.
- `_proxy`: Proxy object for the connection.
- `_stmt_exclusive_section`: Atomic section for statement operations.
- `_config`: Client configuration object.
- `_params`: Connection parameters.
- `_addr`: Network address of the server.
- `_log_listeners`: Listeners for logging events.
- `_termination_listeners`: Listeners for termination events.
- `_cancellations`: Set for handling cancellations.
- `_source_traceback`: Traceback of connection origin if debug mode is enabled.
- `_query_loggers`: Loggers for queries.

### Initialization
Connections are initialized internally by the `asyncpg.connect` function, setting up the protocol, transport, loop, and configuration parameters.
```

--------------------------------

### Transaction Management

Source: https://magicstack.github.io/asyncpg/current/_modules/asyncpg/transaction.html

The Transaction class provides an interface for managing database transactions. It supports various isolation levels, read-only modes, and deferrable transactions. Transactions can be started, committed, and rolled back, and they can be used as asynchronous context managers.

```APIDOC
## Transaction Class

### Description
Represents a transaction or savepoint block. Transactions are created by calling the :meth:`Connection.transaction() <connection.Connection.transaction>` function.

### Initialization

`__init__(self, connection, isolation, readonly, deferrable)`

- **connection**: The database connection object.
- **isolation** (str): The transaction isolation level. Expected values are 'read_committed', 'read_uncommitted', 'serializable', 'repeatable_read'.
- **readonly** (bool): If True, the transaction is read-only.
- **deferrable** (bool): If True, the transaction is deferrable.

### Methods

`start()`
  - **Description**: Enters the transaction or savepoint block. This method is called automatically when entering an `async with` block.
  - **Guarded**: Yes, ensures the connection is valid.

`__aenter__()`
  - **Description**: Asynchronous context manager entry. Starts the transaction.

`__aexit__(self, extype, ex, tb)`
  - **Description**: Asynchronous context manager exit. Commits the transaction if no exception occurred, otherwise rolls it back.

### Transaction States

`TransactionState` is an enum with the following states:
- `NEW`: The transaction has been created but not yet started.
- `STARTED`: The transaction has been started.
- `COMMITTED`: The transaction has been committed.
- `ROLLEDBACK`: The transaction has been rolled back.
- `FAILED`: The transaction has failed.

### Isolation Levels

Supported isolation levels:
- `read_committed`
- `read_uncommitted`
- `serializable`
- `repeatable_read`

### Example Usage

```python
import asyncpg

async def example(conn):
    async with conn.transaction(isolation='read_committed', readonly=True) as trans:
        # Perform database operations within the transaction
        await conn.execute("INSERT INTO users (name) VALUES ($1)", 'Alice')
        # If an error occurs, the transaction will be rolled back automatically
        # If the block completes without error, the transaction will be committed
```

### Nested Transactions and Savepoints

- When a `Transaction` object is used within another active transaction (i.e., `con._top_xact` is not `None`), it creates a savepoint.
- Nested transactions must have the same isolation level as the outer transaction.
- The `_nested` attribute is set to `True` for savepoints.
- The query for a savepoint is `SAVEPOINT <savepoint_id>;`.
```

--------------------------------

### fetch()

Source: https://magicstack.github.io/asyncpg/current/api/index.html

Run a query and return the results as a list of Records.

```APIDOC
## fetch(query, *args, timeout=None, record_class=None)

### Description
Run a query and return the results as a list of Record objects.

### Parameters
- **query** (str) - Required - The SQL query to run.
- **args** (list) - Optional - Arguments for the query.
- **timeout** (float) - Optional - Timeout for the operation.
- **record_class** (type) - Optional - Custom record class.

### Response
- **list** - A list of Record objects.
```