# patroni.dcs.consul module

class patroni.dcs.consul.Consul(config: Dict[str, Any])

  Bases: `AbstractDCS`

  __init__(config: Dict[str, Any]) → None

    Prepare DCS paths, Citus group ID, initial values for state information and processing dependencies.

    Variables:

      config – `dict`, reference to config section of selected DCS. i.e.: `zookeeper` for zookeeper, `etcd` for etcd, etc…

  _abc_impl = <_abc._abc_data object>
  _citus_cluster_loader(path: str) → Dict[int, Cluster]

    Load and build all Patroni clusters from a single Citus cluster.

    Parameters:

      path – the path in DCS where to load Cluster(s) from.

    Returns:

      all Citus groups as `dict`, with group IDs as keys and `Cluster` objects as values or a `Cluster` object representing the coordinator with filled Cluster.workers attribute.

  _cluster_from_nodes(nodes: Dict[str, Any]) → Cluster
  _cluster_loader(path: str) → Cluster

    Load and build the `Cluster` object from DCS, which represents a single Patroni or Citus cluster.

    Parameters:

      path – the path in DCS where to load Cluster(s) from.

    Returns:

      `Cluster` instance.

  property _consistency: str
  _delete_leader(**kwargs: Any) → Any

    Remove leader key from DCS.

    This method should remove leader key if current instance is the leader.

    Parameters:

      leader – `Leader` object with information about the leader.

    Returns:

      `True` if successfully committed to DCS.

  _do_attempt_to_acquire_leader(retry: Retry) → bool
  _do_refresh_session(force: bool = False) → bool
    Returns:

      !True if it had to create new session

  _load_cluster(path: str, loader: Callable[[str], Cluster | Dict[int, Cluster]]) → Cluster | Dict[int, Cluster]

    Main abstract method that implements the loading of `Cluster` instance.

    Note

    Internally this method should call the loader method that will build `Cluster` object which represents current state and topology of the cluster in DCS. This method supposed to be called only by the `get_cluster()` method.

    Parameters:

      - path – the path in DCS where to load Cluster(s) from.

      - loader – one of `_cluster_loader()` or `_citus_cluster_loader()`.

    Raise:

      `DCSError` in case of communication problems with DCS. If the current node was running as a primary and exception raised, instance would be demoted.

  static _run_and_handle_exceptions(method: Callable[[...], Any], *args: Any, **kwargs: Any) → Any
  _set_service_name() → None
  _update_leader(**kwargs: Any)

    Update `leader` key (or session) ttl.

    Note

    You have to use CAS (Compare And Swap) operation in order to update leader key, for example for etcd `prevValue` parameter must be used.

    If update fails due to DCS not being accessible or because it is not able to process requests (hopefully temporary), the `DCSError` exception should be raised.

    Parameters:

      leader – a reference to a current `leader` object.

    Returns:

      `True` if `leader` key (or session) has been updated successfully.

  _update_service(data: Dict[str, Any]) → bool | None
  _write_failsafe(**kwargs: Any) → Any

    Write current cluster topology to DCS that will be used by failsafe mechanism (if enabled).

    Parameters:

      value – failsafe topology serialized in JSON format.

    Returns:

      `True` if successfully committed to DCS.

  _write_leader_optime(**kwargs: Any) → Any

    Write current WAL LSN into `/optime/leader` key in DCS.

    Parameters:

      last_lsn – absolute WAL LSN in bytes.

    Returns:

      `True` if successfully committed to DCS.

  _write_status(**kwargs: Any) → Any

    Write current WAL LSN and `confirmed_flush_lsn` of permanent slots into the `/status` key in DCS.

    Parameters:

      value – status serialized in JSON format.

    Returns:

      `True` if successfully committed to DCS.

  adjust_ttl() → None
  attempt_to_acquire_leader(**kwargs: Any)

    Attempt to acquire leader lock.

    Note

    This method should create `/leader` key with the value `_name`.

    The key must be created atomically. In case the key already exists it should not be overwritten and `False` must be returned.

    If key creation fails due to DCS not being accessible or because it is not able to process requests (hopefully temporary), the `DCSError` exception should be raised.

    Returns:

      `True` if key has been created successfully.

  cancel_initialization(**kwargs: Any) → Any

    Removes the `initialize` key for a cluster.

    Returns:

      `True` if successfully committed to DCS.

  create_session() → None
  delete_cluster(**kwargs: Any) → Any

    Delete cluster from DCS.

    Returns:

      `True` if successfully committed to DCS.

  delete_sync_state(**kwargs: Any) → Any

    Delete the synchronous state from DCS.

    Parameters:

      version – for conditional deletion of the key/object.

    Returns:

      `True` if delete successful.

  deregister_service(**kwargs: Any) → Any
  initialize(**kwargs: Any) → Any

    Race for cluster initialization.

    This method should atomically create `initialize` key and return `True`, otherwise it should return `False`.

    Parameters:

      - create_new – `False` if the key should already exist (in the case we are setting the system_id).

      - sysid – PostgreSQL cluster system identifier, if specified, is written to the key.

    Returns:

      `True` if key has been created successfully.

  static member(node: Dict[str, str]) → Member
  refresh_session() → bool
  register_service(**kwargs: Any) → Any
  reload_config(config: Config | Dict[str, Any]) → None

    Load and set relevant values from configuration.

    Sets `loop_wait`, `ttl` and `retry_timeout` properties.

    Parameters:

      config – Loaded configuration information object or dictionary of key value pairs.

  retry(method: Callable[[...], Any], *args: Any, **kwargs: Any) → Any
  set_config_value(**kwargs: Any) → Any

    Create or update `/config` key in DCS.

    Parameters:

      - value – new value to set in the `config` key.

      - version – for conditional update of the key/object.

    Returns:

      `True` if successfully committed to DCS.

  set_failover_value(**kwargs: Any) → Any

    Create or update `/failover` key.

    Parameters:

      - value – value to set.

      - version – for conditional update of the key/object.

    Returns:

      `True` if successfully committed to DCS.

  set_history_value(**kwargs: Any) → Any

    Set value for `history` in DCS.

    Parameters:

      value – new value of `history` key/object.

    Returns:

      `True` if successfully committed to DCS.

  set_retry_timeout(retry_timeout: int) → None

    Set the new value for retry_timeout.

  set_sync_state_value(**kwargs: Any) → Any

    Set synchronous state in DCS.

    Parameters:

      - value – the new value of `/sync` key.

      - version – for conditional update of the key/object.

    Returns:

      version of the new object or `False` in case of error.

  set_ttl(ttl: int) → bool | None

    Set the new ttl value for DCS keys.

  take_leader() → bool

    Establish a new leader in DCS.

    Note

    This method should create leader key with value of `_name` and `ttl` of `ttl`.

    Since it could be called only on initial cluster bootstrap it could create this key regardless, overwriting the key if necessary.

    Returns:

      `True` if successfully committed to DCS.

  touch_member(**kwargs: Any) → Any

    Update member key in DCS.

    Note

    This method should create or update key with the name with `/members/` + `_name` and the value of data in a given DCS.

    Parameters:

      data – information about an instance (including connection strings).

    Returns:

      `True` if successfully committed to DCS.

  property ttl: int

    Get current `ttl` value.

  update_service(**kwargs: Any) → Any
  watch(leader_version: int | None, timeout: float) → bool

    Sleep if the current node is a leader, otherwise, watch for changes of leader key with a given timeout.

    Parameters:

      - leader_version – version of a leader key.

      - timeout – timeout in seconds.

    Returns:

      if `True` this will reschedule the next run of the HA cycle.

class patroni.dcs.consul.ConsulClient(*args: Any, **kwargs: Any)

  Bases: `Consul`

  __init__(*args: Any, **kwargs: Any) → None

    Consul client with Patroni customisations.

    Note

    Parameters, token, cert and ca_cert are not passed to the parent class `consul.base.Consul`.

    Original class documentation,

    > token is an optional `ACL token`. If supplied it will be used by default for all requests made with this client session. It’s still possible to override this token by passing a token explicitly for a request.

    > consistency sets the consistency mode to use by default for all reads that support the consistency option. It’s still possible to override this by passing explicitly for a given request. consistency can be either ‘default’, ‘consistent’ or ‘stale’.

    > dc is the datacenter that this agent will communicate with. By default, the datacenter of the host is used.

    > verify is whether to verify the SSL certificate for HTTPS requests

    > cert client side certificates for HTTPS requests

    Parameters:

      - args – positional arguments to pass to `consul.base.Consul`

      - kwargs – keyword arguments, with cert, ca_cert and token removed, passed to `consul.base.Consul`

  connect(*args: Any, **kwargs: Any) → HTTPClient
  http_connect(*args: Any, **kwargs: Any) → HTTPClient
  reload_config(config: Dict[str, Any]) → None

exception patroni.dcs.consul.ConsulError(value: Any)

  Bases: `DCSError`

exception patroni.dcs.consul.ConsulInternalError

  Bases: `ConsulException`

  An internal Consul server error occurred

class patroni.dcs.consul.HTTPClient(host: str = '127.0.0.1', port: int = 8500, token: str | None = None, scheme: str = 'http', verify: bool = True, cert: str | None = None, ca_cert: str | None = None)

  Bases: `object`

  __init__(host: str = '127.0.0.1', port: int = 8500, token: str | None = None, scheme: str = 'http', verify: bool = True, cert: str | None = None, ca_cert: str | None = None) → None
  static response(response: HTTPResponse) → Response
  set_read_timeout(timeout: float) → None
  set_ttl(ttl: int) → bool
  property ttl: int
  uri(path: str, params: None | Dict[str, Any] | List[Tuple[str, Any]] | Tuple[Tuple[str, Any], ...] = None) → str

exception patroni.dcs.consul.InvalidSession

  Bases: `ConsulException`

  invalid session

exception patroni.dcs.consul.InvalidSessionTTL

  Bases: `ConsulException`

  Session TTL is too small or too big

class patroni.dcs.consul.Response(code, headers, body, content)

  Bases: `NamedTuple`

  _asdict()

    Return a new dict which maps field names to their values.

  _field_defaults = {}
  _fields = ('code', 'headers', 'body', 'content')
  classmethod _make(iterable)

    Make a new Response object from a sequence or iterable

  _replace(**kwds)

    Return a new Response object replacing specified fields with new values

  body: str

    Alias for field number 2

  code: int

    Alias for field number 0

  content: bytes

    Alias for field number 3

  headers: Mapping[str, str] | Mapping[bytes, bytes] | None

    Alias for field number 1

patroni.dcs.consul.catch_consul_errors(func: Callable[[...], Any]) → Callable[[...], Any]

patroni.dcs.consul.force_if_last_failed(func: Callable[[...], Any]) → Callable[[...], Any]

patroni.dcs.consul.service_name_from_scope_name(scope_name: str) → str

  Translate scope name to service name which can be used in dns.

  230 = 253 - len(‘replica.’) - len(‘.service.consul’)