# patroni.validator module

Patroni configuration validation helpers.

This module contains facilities for validating configuration of Patroni processes.

var schema:

  configuration schema of the daemon launched by `patroni` command.

class patroni.validator.AtMostOne(*args: str)

  Bases: `object`

  Mark that at most one option from a `Case` can be supplied.

  Represents a list of possible configuration options in a given scope, where at most one can actually be provided.

  Note

  It should be used together with a `Case` object.

  __init__(*args: str) → None

    Create a :class`AtMostOne` object.

    Parameters:

      *args – any arguments that the caller wants to be stored in this `Or` object.

    Example:

          AtMostOne("nofailover", "failover_priority"): Case({
              "nofailover": bool,
              "failover_priority": IntValidator(min=0, raise_assert=True),
          })

    The :class`AtMostOne` object is used to define that at most one of `nofailover` and `failover_priority` can be provided.

class patroni.validator.BinDirectory(contains: List[str] | None = None, contains_executable: List[str] | None = None)

  Bases: `Directory`

  Check if a Postgres binary directory contains the expected files.

  It is a subclass of `Directory` with an extended capability: translating `BINARIES` according to configured `postgresql.bin_name`, if any.

  Variables:

    BINARIES – list of executable files that should exist directly under a given Postgres binary directory.

  BINARIES = ['pg_ctl', 'initdb', 'pg_controldata', 'pg_basebackup', 'postgres', 'pg_isready']
  validate(name: str) → Iterator[Result]

    Check if the expected executables can be found under name binary directory.

    Parameters:

      name – path to the base directory against which executables will be validated. Check against PATH if name is not provided.

    Yields:

      objects with the error message related to the failure, if any check fails.

class patroni.validator.Case(schema: Dict[str, Any])

  Bases: `object`

  Map how a list of available configuration options should be validated.

  Note

  It should be used together with an `Or` object. The `Or` object will define the list of possible configuration options in a given context, and the `Case` object will dictate how to validate each of them, if they are set.

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

    Create a `Case` object.

    Parameters:

      schema – the schema for validating a set of attributes that may be available in the configuration. Each key is the configuration that is available in a given scope and that should be validated, and the related value is the validation function or expected type.

    Example:

          Case({
              "host": validate_host_port,
              "url": str,
          })

    That will check that `host` configuration, if given, is valid based on `validate_host_port()`, and will also check that `url` configuration, if given, is a `str` instance.

class patroni.validator.Directory(contains: List[str] | None = None, contains_executable: List[str] | None = None)

  Bases: `object`

  Check if a directory contains the expected files.

  The attributes of objects of this class are used by their `validate()` method.

  Parameters:

    - contains – list of paths that should exist relative to a given directory.

    - contains_executable – list of executable files that should exist directly under a given directory.

  __init__(contains: List[str] | None = None, contains_executable: List[str] | None = None) → None

    Create a `Directory` object.

    Parameters:

      - contains – list of paths that should exist relative to a given directory.

      - contains_executable – list of executable files that should exist directly under a given directory.

  _check_executables(path: str | None = None) → Iterator[Result]

    Check that all executables from contains_executable list exist within the given directory or within `PATH`.

    Parameters:

      path – optional path to the base directory against which executables will be validated. If not provided, check within `PATH`.

    Yields:

      objects with the error message containing the name of the executable, if any check fails.

  validate(name: str) → Iterator[Result]

    Check if the expected paths and executables can be found under name directory.

    Parameters:

      name – path to the base directory against which paths and executables will be validated. Check against `PATH` if name is not provided.

    Yields:

      objects with the error message related to the failure, if any check fails.

class patroni.validator.EnumValidator(allowed_values: Tuple[str, ...], case_sensitive: bool = False, raise_assert: bool = False)

  Bases: `object`

  Validate enum setting

  Variables:

    - allowed_values – a `set` or `CaseInsensitiveSet` object with allowed enum values.

    - raise_assert – if an `assert` call should be performed regarding expected type and valid range.

  __init__(allowed_values: Tuple[str, ...], case_sensitive: bool = False, raise_assert: bool = False) → None

    Create an `EnumValidator` object with given allowed values.

    Parameters:

      - allowed_values – a tuple with allowed enum values

      - case_sensitive – set to `True` to do case sensitive comparisons

      - raise_assert – if an `assert` call should be performed regarding expected values.

class patroni.validator.IntValidator(min: int | None = None, max: int | None = None, base_unit: str | None = None, expected_type: Any = None, raise_assert: bool = False)

  Bases: `object`

  Validate an integer setting.

  Variables:

    - min – minimum allowed value for the setting, if any.

    - max – maximum allowed value for the setting, if any.

    - base_unit – the base unit to convert the value to before checking if it’s within min and max range.

    - expected_type – the expected Python type.

    - raise_assert – if an `assert` test should be performed regarding expected type and valid range.

  __init__(min: int | None = None, max: int | None = None, base_unit: str | None = None, expected_type: Any = None, raise_assert: bool = False) → None

    Create an `IntValidator` object with the given rules.

    Parameters:

      - min – minimum allowed value for the setting, if any.

      - max – maximum allowed value for the setting, if any.

      - base_unit – the base unit to convert the value to before checking if it’s within min and max range.

      - expected_type – the expected Python type.

      - raise_assert – if an `assert` test should be performed regarding expected type and valid range.

class patroni.validator.Optional(name: str, default: Any | None = None)

  Bases: `object`

  Mark a configuration option as optional.

  Variables:

    - name – name of the configuration option.

    - default – value to set if the configuration option is not explicitly provided

  __init__(name: str, default: Any | None = None) → None

    Create an `Optional` object.

    Parameters:

      - name – name of the configuration option.

      - default – value to set if the configuration option is not explicitly provided

class patroni.validator.Or(*args: Any)

  Bases: `object`

  Represent the list of options that are available.

  It can represent either a list of configuration options that are available in a given scope, or a list of validation functions and/or expected types for a given configuration option.

  __init__(*args: Any) → None

    Create an `Or` object.

    Parameters:

      *args – any arguments that the caller wants to be stored in this `Or` object.

    Example:

          Or("host", "hosts"): Case({
              "host": validate_host_port,
              "hosts": Or(comma_separated_host_port, [validate_host_port]),
          })

      The outer `Or` is used to define that `host` and `hosts` are possible options in this scope. The inner :class`Or` in the `hosts` key value is used to define that `hosts` option is valid if either of `comma_separated_host_port()` or `validate_host_port()` succeed to validate it.

class patroni.validator.Result(status: bool, error: str | None = "didn't pass validation", level: int = 0, path: str = '', data: Any = '')

  Bases: `object`

  Represent the result of a given validation that was performed.

  Variables:

    - status – If the validation succeeded.

    - path – YAML tree path of the configuration option.

    - data – value of the configuration option.

    - level – error level, in case of error.

    - error – error message if the validation failed, otherwise `None`.

  __init__(status: bool, error: str | None = "didn't pass validation", level: int = 0, path: str = '', data: Any = '') → None

    Create a `Result` object based on the given arguments.

    Note

    `error` attribute is only set if status is failed.

    Parameters:

      - status – if the validation succeeded.

      - error – error message related to the validation that was performed, if the validation failed.

      - level – error level, in case of error.

      - path – YAML tree path of the configuration option.

      - data – value of the configuration option.

class patroni.validator.Schema(validator: Dict[Any, Any] | List[Any] | Any)

  Bases: `object`

  Define a configuration schema.

  It contains all the configuration options that are available in each scope, including the validation(s) that should be performed against each one of them. The validations will be performed whenever the `Schema` object is called, or its `validate()` method is called.

  Variables:

    validator –

    validator of the configuration schema. Can be any of these:

    - `str`: defines that a string value is required; or

    - `type`: any subclass of `type`, defines that a value of the given type is required; or

    - `callable`: any callable object, defines that validation will follow the code defined in the callable object. If the callable object contains an `expected_type` attribute, then it will check if the configuration value is of the expected type before calling the code of the callable object; or

    - `list`: list representing one or more values in the configuration; or

    - `dict`: dictionary representing the YAML configuration tree.

  __init__(validator: Dict[Any, Any] | List[Any] | Any) → None

    Create a `Schema` object.

    Note

    This class is expected to be initially instantiated with a `dict` based validator argument. The idea is that dict represents the full YAML tree of configuration options. The `validate()` method will then walk recursively through the configuration tree, creating new instances of `Schema` with the new “base path”, to validate the structure and the leaf values of the tree. The recursion stops on leaf nodes, when it performs checks of the actual setting values.

    Parameters:

      validator –

      validator of the configuration schema. Can be any of these:

      - `str`: defines that a string value is required; or

      - `type`: any subclass of `type`, defines that a value of the given type is required; or

      - `callable`: Any callable object, defines that validation will follow the code defined in the callable object. If the callable object contains an `expected_type` attribute, then it will check if the configuration value is of the expected type before calling the code of the callable object; or

      - `list`: list representing it expects to contain one or more values in the configuration; or

      - `dict`: dictionary representing the YAML configuration tree.

      The first 3 items in the above list are here referenced as “base validators”, which cause the recursion to stop.

      If validator is a `dict`, then you should follow these rules:

      - For the keys it can be either:

        > - A `str` instance. It will be the name of the configuration option; or

        > - An `Optional` instance. The `name` attribute of that object will be the name of the configuration option, and that class makes this configuration option as optional to the user, allowing it to not be specified in the YAML; or

        > - An `Or` instance. The `args` attribute of that object will contain a tuple of

        >     configuration option names. At least one of them should be specified by the user in the YAML;

      - For the values it can be either:

        > - A new `dict` instance. It will represent a new level in the YAML configuration tree; or

        > - A `Case` instance. This is required if the key of this value is an `Or` instance, and the `Case` instance is used to map each of the `args` in `Or` to their corresponding base validator in `Case`; or

        > - An `Or` instance with one or more base validators; or

        > - A `list` instance with a single item which is the base validator; or

        > - A base validator.

    Example:

          Schema({
              "application_name": str,
              "bind": {
                  "host": validate_host,
                  "port": int,
              },
              "aliases": [str],
              Optional("data_directory"): "/var/lib/myapp",
              Or("log_to_file", "log_to_db"): Case({
                  "log_to_file": bool,
                  "log_to_db": bool,
              }),
              "version": Or(int, float),
          })

      This sample schema defines that your YAML configuration follows these rules:

      > - It must contain an `application_name` entry which value should be a `str` instance;

      > - It must contain a `bind.host` entry which value should be valid as per function `validate_host`;

      > - It must contain a `bind.port` entry which value should be an `int` instance;

      > - It must contain a `aliases` entry which value should be a `list` of `str` instances;

      > - It may optionally contain a `data_directory` entry, with a value which should be a string;

      > - It must contain at least one of `log_to_file` or `log_to_db`, with a value which should be a `bool` instance;

      > - It must contain a `version` entry which value should be either an `int` or a `float` instance.

  _data_key(key: str | Optional | Or | AtMostOne) → Iterator[str]

    Map a key from the `validator` dictionary to the corresponding key(s) in the `data` dictionary.

    Parameters:

      key – key from the `validator` attribute.

    Yields:

      keys that should be used to access corresponding value in the `data` attribute.

  iter_dict() → Iterator[Result]

    Iterate over a `dict` based `validator` to validate the corresponding entries in `data`.

    Yields:

      objects with the error message related to the failure, if any check fails.

  iter_list() → Iterator[Result]

    Iterate over a `data` object and perform validations using the first element of the `validator`.

    Yields:

      objects with the error message related to the failure, if any check fails.

  iter_or() → Iterator[Result]

    Perform all validations defined in an `Or` object for a given configuration option.

    This method can be only called against leaf nodes in the configuration tree. `Or` objects defined in the `validator` keys will be handled by `iter_dict()` method.

    Yields:

      objects with the error message related to the failure, if any check fails.

  validate(data: Any) → Iterator[Result]

    Perform all validations from the schema against the given configuration.

    It first checks that data argument type is compliant with the type of `validator` attribute.

    Additionally:
      - If `validator` attribute is a callable object, calls it to validate data argument. Before doing so, if

          validator contains an `expected_type` attribute, check if data argument is compliant with that expected type.

      - If `validator` attribute is an iterable object (`dict`, `list`, `Directory` or

          `Or`), then it iterates over it to validate each of the corresponding entries in data argument.

    Parameters:

      data – configuration to be validated against `validator`.

    Yields:

      objects with the error message related to the failure, if any check fails.

patroni.validator._get_type_name(python_type: Any) → str

  Get a user-friendly name for a given Python type.

  Parameters:

    python_type – Python type which user friendly name should be taken.

  Returns:

    User friendly name of the given Python type.

patroni.validator.assert_(condition: bool, message: str = 'Wrong value') → None

  Assert that a given condition is `True`.

  If the assertion fails, then throw a message.

  Parameters:

    - condition – result of a condition to be asserted.

    - message – message to be thrown if the condition is `False`.

patroni.validator.comma_separated_host_port(string: str) → bool

  Validate a list of host and port items.

  Call `validate_host_port_list()` with a list represented by the CSV string.

  Parameters:

    string – comma-separated list of host and port items.

  Returns:

    `True` if all items in the CSV string are valid.

patroni.validator.data_directory_empty(data_dir: str) → bool

  Check if PostgreSQL data directory is empty.

  Parameters:

    data_dir – path to the PostgreSQL data directory to be checked.

  Returns:

    `True` if the data directory is empty.

patroni.validator.get_bin_name(bin_name: str) → str

  Get the value of `postgresql.bin_name[*bin_name*]` configuration option.

  Parameters:

    bin_name – a key to be retrieved from `postgresql.bin_name` configuration.

  Returns:

    value of `postgresql.bin_name[*bin_name*]`, if present, otherwise bin_name.

patroni.validator.is_ipv4_address(ip: str) → bool

  Check if ip is a valid IPv4 address.

  Parameters:

    ip – the IP to be checked.

  Returns:

    `True` if the IP is an IPv4 address.

  Raises:

    `ConfigParseError`: if ip is not a valid IPv4 address.

patroni.validator.is_ipv6_address(ip: str) → bool

  Check if ip is a valid IPv6 address.

  Parameters:

    ip – the IP to be checked.

  Returns:

    `True` if the IP is an IPv6 address.

  Raises:

    `ConfigParseError`: if ip is not a valid IPv6 address.

patroni.validator.populate_validate_params(ignore_listen_port: bool = False) → None

  Populate parameters used to fine-tune the validation of the Patroni config.

  Parameters:

    ignore_listen_port – ignore the bind failures for the ports marked as listen.

patroni.validator.validate_binary_name(bin_name: str) → bool

  Validate the value of `postgresql.binary_name[*bin_name*]` configuration option.

  If `postgresql.bin_dir` is set and the value of the bin_name meets these conditions:

  > - The path join of `postgresql.bin_dir` plus the bin_name value exists; and

  > - The path join as above is executable

  If `postgresql.bin_dir` is not set, then validate that the value of bin_name meets this condition:

  > - Is found in the system PATH using `which`

  Parameters:

    bin_name – the value of the `postgresql.bin_name[*bin_name*]`

  Returns:

    `True` if the conditions are true

  Raises:
    `ConfigParseError` if:

      - bin_name is not set; or

      - the path join of the `postgresql.bin_dir` plus bin_name does not exist; or

      - the path join as above is not executable; or

      - the bin_name cannot be found in the system PATH

patroni.validator.validate_connect_address(address: str) → bool

  Check if options related to connection address were properly configured.

  Parameters:

    address – address to be validated in the format `host:ip`.

  Returns:

    `True` if the address is valid.

  Raises:
    `ConfigParseError`:

      - If the address is not in the expected format; or

      - If the host is set to not allowed values (`127.0.0.1`, `0.0.0.0`, `*`, `::1`, or `localhost`).

patroni.validator.validate_data_dir(data_dir: str) → bool

  Validate the value of `postgresql.data_dir` configuration option.

  It requires that `postgresql.data_dir` is set and match one of following conditions:

  > - Point to a path that does not exist yet; or

  > - Point to an empty directory; or

  > - Point to a non-empty directory that seems to contain a valid PostgreSQL data directory.

  Parameters:

    data_dir – the value of `postgresql.data_dir` configuration option.

  Returns:

    `True` if the PostgreSQL data directory is valid.

  Raises:
    `ConfigParseError`:

      - If no data_dir was given; or

      - If data_dir is a file and not a directory; or

      - If data_dir is a non-empty directory and:

          - `PG_VERSION` file is not available in the directory

          - `pg_wal`/`pg_xlog` is not available in the directory

          - `PG_VERSION` content does not match the major version reported by `postgres --version`

patroni.validator.validate_host_port(host_port: str, listen: bool = False, multiple_hosts: bool = False) → bool

  Check if host(s) and port are valid and available for usage.

  Parameters:

    - host_port –

      the host(s) and port to be validated. It can be in either of these formats:

      - `host:ip`, if multiple_hosts is `False`; or

      - `host_1,host_2,...,host_n:port`, if multiple_hosts is `True`.

    - listen – if the address is expected to be available for binding. `False` means it expects to connect to that address, and `True` that it expects to bind to that address.

    - multiple_hosts – if host_port can contain multiple hosts.

  Returns:

    `True` if the host(s) and port are valid.

  Raises:
    `ConfigParseError`:

      - If the host_port is not in the expected format; or

      - If `*` was specified along with more hosts in host_port; or

      - If we are expecting to bind to an address that is already in use; or

      - If we are not able to connect to an address that we are expecting to do so; or

      - If `gaierror` is thrown by socket module when attempting to connect to the given address(es).

patroni.validator.validate_host_port_list(value: List[str]) → bool

  Validate a list of host(s) and port items.

  Call `validate_host_port()` with each item in value.

  Parameters:

    value – list of host(s) and port items to be validated.

  Returns:

    `True` if all items are valid.

patroni.validator.validate_host_port_listen(host_port: str) → bool

  Check if host and port are valid and available for binding.

  Call `validate_host_port()` with listen set to `True`.

  Parameters:

    host_port – the host and port to be validated. Must be in the format `host:ip`.

  Returns:

    `True` if the host and port are valid and available for binding.

patroni.validator.validate_host_port_listen_multiple_hosts(host_port: str) → bool

  Check if host(s) and port are valid and available for binding.

  Call `validate_host_port()` with both listen and multiple_hosts set to `True`.

  Parameters:

    host_port –

    the host(s) and port to be validated. It can be in either of these formats

    - `host:ip`; or

    - `host_1,host_2,...,host_n:port`

  Returns:

    `True` if the host(s) and port are valid and available for binding.

patroni.validator.validate_log_field(field: Any) → bool

  Checks if log field is valid.

  Parameters:

    field – A log field to be validated.

  Returns:

    `True` if the field is either a string or a dictionary with exactly one key that has string value, `False` otherwise.

patroni.validator.validate_log_format(logformat: List[str | Dict[str, Any] | Any] | str | Any) → bool

  Checks if log format is valid.

  Parameters:

    logformat – A log format to be validated.

  Returns:

    `True` if the log format is either a string or a list of valid log fields.

  Raises:
    `ConfigParseError`:

      - If the logformat is not a string or a list; or

      - If the logformat is an empty list; or

      - If the log format is a list and it with values that don’t pass validation using `validate_log_field()`.

patroni.validator.validate_watchdog_mode(value: Any) → None

  Validate `watchdog.mode` configuration option.

  Parameters:

    value – value of `watchdog.mode` to be validated.