This document is for Kombu's development version, which can be significantly different from previous releases. Get the stable docs here: 5.3.

# Connection - `kombu.connection`

Client (Connection).

## Connection

class kombu.connection.Connection(hostname='localhost', userid=None, password=None, virtual_host=None, port=None, insist=False, ssl=False, transport=None, connect_timeout=5, transport_options=None, login_method=None, uri_prefix=None, heartbeat=0, failover_strategy='round-robin', alternates=None, credential_provider=None, **kwargs)

  A connection to the broker.

  ### Example:

      >>> Connection('amqp://guest:guest@localhost:5672//')
      >>> Connection('amqp://foo;amqp://bar',
      ...            failover_strategy='round-robin')
      >>> Connection('redis://', transport_options={
      ...     'visibility_timeout': 3000,
      ... })

      >>> import ssl
      >>> Connection('amqp://', login_method='EXTERNAL', ssl={
      ...    'ca_certs': '/etc/pki/tls/certs/something.crt',
      ...    'keyfile': '/etc/something/system.key',
      ...    'certfile': '/etc/something/system.cert',
      ...    'cert_reqs': ssl.CERT_REQUIRED,
      ... })

  ### Note:

  > SSL currently only works with the py-amqp, qpid and redis transports. For other transports you can use stunnel.

  ### Arguments:

  > URL (str, Sequence): Broker URL, or a list of URLs.

  ### Keyword Arguments:

  > ssl (bool/dict): Use SSL to connect to the server.

  >   Default is `False`. May not be supported by the specified transport.

  > transport (Transport): Default transport if not specified in the URL. connect_timeout (float): Timeout in seconds for connecting to the

  > > server. May not be supported by the specified transport.

  > transport_options (Dict): A dict of additional connection arguments to

  >   pass to alternate kombu channel implementations. Consult the transport documentation for available options.

  > heartbeat (float): Heartbeat interval in int/float seconds.

  >   Note that if heartbeats are enabled then the `heartbeat_check()` method must be called regularly, around once per second.

  ### Note:

  > The connection is established lazily when needed. If you need the connection to be established, then force it by calling `connect()`:

  >     >>> conn = Connection('amqp://')
  >     >>> conn.connect()

  > and always remember to close the connection:

  >     >>> conn.release()

  These options have been replaced by the URL argument, but are still supported for backwards compatibility:

  keyword hostname:

    Host name/address. NOTE: You cannot specify both the URL argument and use the hostname keyword argument at the same time.

  keyword userid:

    Default user name if not provided in the URL.

  keyword password:

    Default password if not provided in the URL.

  keyword virtual_host:

    Default virtual host if not provided in the URL.

  keyword port:

    Default port if not provided in the URL.

  ChannelPool(limit=None, **kwargs)

    Pool of channels.

    #### Arguments:

    > limit (int): Maximum number of active channels.

    >   Default is no limit.

    #### Example:

        >>> connection = Connection('amqp://')
        >>> pool = connection.ChannelPool(2)
        >>> c1 = pool.acquire()
        >>> c2 = pool.acquire()
        >>> c3 = pool.acquire()
        Traceback (most recent call last):
          File "<stdin>", line 1, in <module>
          File "kombu/connection.py", line 354, in acquire
          raise ChannelLimitExceeded(self.limit)
            kombu.connection.ChannelLimitExceeded: 2
        >>> c1.release()
        >>> c3 = pool.acquire()

  Consumer(queues=None, channel=None, *args, **kwargs)

    Create new `kombu.Consumer` instance.

  Pool(limit=None, **kwargs)

    Pool of connections.

    #### Arguments:

    > limit (int): Maximum number of active connections.

    >   Default is no limit.

    #### Example:

        >>> connection = Connection('amqp://')
        >>> pool = connection.Pool(2)
        >>> c1 = pool.acquire()
        >>> c2 = pool.acquire()
        >>> c3 = pool.acquire()
        Traceback (most recent call last):
          File "<stdin>", line 1, in <module>
          File "kombu/connection.py", line 354, in acquire
          raise ConnectionLimitExceeded(self.limit)
            kombu.exceptions.ConnectionLimitExceeded: 2
        >>> c1.release()
        >>> c3 = pool.acquire()

  Producer(channel=None, *args, **kwargs)

    Create new `kombu.Producer` instance.

  SimpleBuffer(name, no_ack=None, queue_opts=None, queue_args=None, exchange_opts=None, channel=None, **kwargs)

    Simple ephemeral queue API.

    Create new `SimpleQueue` using a channel from this connection.

  SimpleQueue(name, no_ack=None, queue_opts=None, queue_args=None, exchange_opts=None, channel=None, **kwargs)

    Simple persistent queue API.

    Create new `SimpleQueue`, using a channel from this connection.

    If `name` is a string, a queue and exchange will be automatically created using that name as the name of the queue and exchange, also it will be used as the default routing key.

    #### Arguments:

    > name (str, kombu.Queue): Name of the queue/or a queue. no_ack (bool): Disable acknowledgments. Default is false. queue_opts (Dict): Additional keyword arguments passed to the

    > > constructor of the automatically created `Queue`.

    > queue_args (Dict): Additional keyword arguments passed to the

    >   constructor of the automatically created `Queue` for setting implementation extensions (e.g., in RabbitMQ).

    > exchange_opts (Dict): Additional keyword arguments passed to the

    >   constructor of the automatically created `Exchange`.

    > channel (ChannelT): Custom channel to use. If not specified the

    >   connection default channel is used.

  as_uri(include_password=False, mask='**', getfields=operator.itemgetter('port', 'userid', 'password', 'virtual_host', 'transport')) → str

    Convert connection parameters to URL form.

  autoretry(fun, channel=None, **ensure_options)

    Decorator for functions supporting a `channel` keyword argument.

    The resulting callable will retry calling the function if it raises connection or channel related errors. The return value will be a tuple of `(retval, last_created_channel)`.

    If a `channel` is not provided, then one will be automatically acquired (remember to close it afterwards).

    #### Example:

        >>> channel = connection.channel()
        >>> try:
        ...    ret, channel = connection.autoretry(
        ...         publish_messages, channel)
        ... finally:
        ...    channel.close()

  channel()

    Create and return a new channel.

  property channel_errors

    List of exceptions that may be raised by the channel.

  clone(**kwargs)

    Create a copy of the connection with same settings.

  close()

    Close the connection (if open).

  collect(socket_timeout=None)
  completes_cycle(retries)

    Return true if the cycle is complete after number of retries.

  connect()

    Establish connection to server immediately.

  connect_timeout = 5
  property connected

    Return true if the connection has been established.

  property connection

    The underlying connection object.

    #### Warning:

    > This instance is transport specific, so do not depend on the interface of this object.

  property connection_errors

    List of exceptions that may be raised by the connection.

  create_transport()
  cycle = None

    Iterator returning the next broker URL to try in the event of connection failure (initialized by `failover_strategy`).

  declared_entities = None

    The cache of declared entities is per connection, in case the server loses data.

  property default_channel: Channel

    Default channel.

    Created upon access and closed when the connection is closed.

    #### Note:

    > Can be used for automatic channel handling when you only need one channel, and also it is the channel implicitly used if a connection is passed instead of a channel, to functions that require a channel.

  drain_events(**kwargs)

    Wait for a single event from the server.

    #### Arguments:

    > timeout (float): Timeout in seconds before we give up.

    raises socket.timeout:

      if the timeout is exceeded.:

  ensure(obj, fun, errback=None, max_retries=None, interval_start=1, interval_step=1, interval_max=1, on_revive=None, retry_errors=None)

    Ensure operation completes.

    Regardless of any channel/connection errors occurring.

    Retries by establishing the connection, and reapplying the function.

    #### Arguments:

    > obj: The object to ensure an action on. fun (Callable): Method to apply.

    > errback (Callable): Optional callback called each time the

    >   connection can’t be established. Arguments provided are the exception raised and the interval that will be slept `(exc, interval)`.

    > max_retries (int): Maximum number of times to retry.

    >   If this limit is exceeded the connection error will be re-raised.

    > interval_start (float): The number of seconds we start

    >   sleeping for.

    > interval_step (float): How many seconds added to the interval

    >   for each retry.

    > interval_max (float): Maximum number of seconds to sleep between

    >   each retry.

    > on_revive (Callable): Optional callback called whenever

    >   revival completes successfully

    > retry_errors (tuple): Optional list of errors to retry on

    >   regardless of the connection state.

    Examples

        >>> from kombu import Connection, Producer
        >>> conn = Connection('amqp://')
        >>> producer = Producer(conn)

        >>> def errback(exc, interval):
        ...     logger.error('Error: %r', exc, exc_info=1)
        ...     logger.info('Retry in %s seconds.', interval)

        >>> publish = conn.ensure(producer, producer.publish,
        ...                       errback=errback, max_retries=3)
        >>> publish({'hello': 'world'}, routing_key='dest')

  ensure_connection(*args, **kwargs)

    Public interface of _ensure_connection for retro-compatibility.

    Returns kombu.Connection instance.

  failover_strategies = {'round-robin': <class 'itertools.cycle'>, 'shuffle': <function shufflecycle>}
  failover_strategy = 'round-robin'

    Strategy used to select new hosts when reconnecting after connection failure. One of “round-robin”, “shuffle” or any custom iterator constantly yielding new URLs to try.

  get_heartbeat_interval()
  get_manager(*args, **kwargs)
  get_transport_cls()

    Get the currently used transport class.

  heartbeat = None

    Heartbeat value, currently only supported by the py-amqp transport.

  heartbeat_check(rate=2)

    Check heartbeats.

    Allow the transport to perform any periodic tasks required to make heartbeats work. This should be called approximately every second.

    If the current transport does not support heartbeats then this is a noop operation.

    #### Arguments:

    > rate (int): Rate is how often the tick is called

    >   compared to the actual heartbeat value. E.g. if the heartbeat is set to 3 seconds, and the tick is called every 3 / 2 seconds, then the rate is 2. This value is currently unused by any transports.

  property host

    The host as a host name/port pair separated by colon.

  hostname = None
  info()

    Get connection info.

  property is_evented
  login_method = None
  property manager

    AMQP Management API.

    Experimental manager that can be used to manage/monitor the broker instance.

    Not available for all transports.

  maybe_close_channel(channel)

    Close given channel, but ignore connection and channel errors.

  maybe_switch_next()

    Switch to next URL given by the current failover strategy.

  password = None
  port = None
  property qos_semantics_matches_spec
  property recoverable_channel_errors

    Recoverable channel errors.

    List of channel related exceptions that can be automatically recovered from without re-establishing the connection.

  property recoverable_connection_errors

    Recoverable connection errors.

    List of connection related exceptions that can be recovered from, but where the connection must be closed and re-established first.

  register_with_event_loop(loop)
  release()

    Close the connection (if open).

  resolve_aliases = {'librabbitmq': 'amqp', 'pyamqp': 'amqp'}
  revive(new_channel)

    Revive connection after connection re-established.

  ssl = None
  supports_exchange_type(exchange_type)
  property supports_heartbeats
  switch(conn_str)

    Switch connection parameters to use a new URL or hostname.

    #### Note:

    > Does not reconnect!

    #### Arguments:

    > conn_str (str): either a hostname or URL.

  property transport
  transport_options = None

    Additional transport specific options, passed on to the transport instance.

  uri_prefix = None
  userid = None
  virtual_host = '/'

## Pools

See also

The shortcut methods `Connection.Pool()` and `Connection.ChannelPool()` is the recommended way to instantiate these classes.

class kombu.connection.ConnectionPool(connection, limit=None, **kwargs)

  Pool of connections.

  LimitExceeded = <class 'kombu.exceptions.ConnectionLimitExceeded'>
  acquire(block=False, timeout=None)

    Acquire resource.

    ### Arguments:

    > block (bool): If the limit is exceeded,

    >   then block until there is an available item.

    > timeout (float): Timeout to wait

    >   if `block` is true. Default is `None` (forever).

    raises LimitExceeded:

      if block is false and the limit has been exceeded.:

  release(resource)
  force_close_all(close_pool=True)

    Close and remove all resources in the pool (also those in use).

    Used to close resources from parent processes after fork (e.g. sockets/connections).

    ### Arguments:

    > close_pool (bool): If True (default) then the pool is marked

    >   as closed. In case of False the pool can be reused.

class kombu.connection.ChannelPool(connection, limit=None, **kwargs)

  Pool of channels.

  LimitExceeded = <class 'kombu.exceptions.ChannelLimitExceeded'>
  acquire(block=False, timeout=None)

    Acquire resource.

    ### Arguments:

    > block (bool): If the limit is exceeded,

    >   then block until there is an available item.

    > timeout (float): Timeout to wait

    >   if `block` is true. Default is `None` (forever).

    raises LimitExceeded:

      if block is false and the limit has been exceeded.:

  release(resource)
  force_close_all(close_pool=True)

    Close and remove all resources in the pool (also those in use).

    Used to close resources from parent processes after fork (e.g. sockets/connections).

    ### Arguments:

    > close_pool (bool): If True (default) then the pool is marked

    >   as closed. In case of False the pool can be reused.