Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use create_ipam_pool method in localstack

Best Python code snippet using localstack_python

__init__.py

Source:__init__.py

1# -*- coding: utf-8 -*-2"""3Common logic used by the docker state and execution module4This module contains logic to accommodate docker/salt CLI usage, as well as5input as formatted by states.6"""7# Import Python libs8from __future__ import absolute_import, print_function, unicode_literals9import copy10import logging11# Import Salt libs12import salt.utils.args13import salt.utils.data14import salt.utils.dockermod.translate15from salt.exceptions import CommandExecutionError, SaltInvocationError16# Import 3rd-party libs17from salt.ext import six18from salt.utils.args import get_function_argspec as _argspec19from salt.utils.dockermod.translate.helpers import split as _split20try:21    import docker22except ImportError:23    docker = None24# These next two imports are only necessary to have access to the needed25# functions so that we can get argspecs for the container config, host config,26# and networking config (see the get_client_args() function).27try:28    import docker.types29except ImportError:30    pass31try:32    import docker.utils33except ImportError:34    pass35NOTSET = object()36__virtualname__ = "docker"37# Default timeout as of docker-py 1.0.038CLIENT_TIMEOUT = 6039# Timeout for stopping the container, before a kill is invoked40SHUTDOWN_TIMEOUT = 1041log = logging.getLogger(__name__)42def __virtual__():43    if docker is None:44        return False45    return __virtualname__46def get_client_args(limit=None):47    if docker is None:48        raise CommandExecutionError("docker Python module not imported")49    limit = salt.utils.args.split_input(limit or [])50    ret = {}51    if not limit or any(52        x in limit53        for x in ("create_container", "host_config", "connect_container_to_network")54    ):55        try:56            ret["create_container"] = _argspec(docker.APIClient.create_container).args57        except AttributeError:58            try:59                ret["create_container"] = _argspec(docker.Client.create_container).args60            except AttributeError:61                raise CommandExecutionError("Coult not get create_container argspec")62        try:63            ret["host_config"] = _argspec(docker.types.HostConfig.__init__).args64        except AttributeError:65            try:66                ret["host_config"] = _argspec(docker.utils.create_host_config).args67            except AttributeError:68                raise CommandExecutionError("Could not get create_host_config argspec")69        try:70            ret["connect_container_to_network"] = _argspec(71                docker.types.EndpointConfig.__init__72            ).args73        except AttributeError:74            try:75                ret["connect_container_to_network"] = _argspec(76                    docker.utils.utils.create_endpoint_config77                ).args78            except AttributeError:79                try:80                    ret["connect_container_to_network"] = _argspec(81                        docker.utils.create_endpoint_config82                    ).args83                except AttributeError:84                    raise CommandExecutionError(85                        "Could not get connect_container_to_network argspec"86                    )87    for key, wrapped_func in (88        ("logs", docker.api.container.ContainerApiMixin.logs),89        ("create_network", docker.api.network.NetworkApiMixin.create_network),90    ):91        if not limit or key in limit:92            try:93                func_ref = wrapped_func94                if six.PY2:95                    try:96                        # create_network is decorated, so we have to dig into the97                        # closure created by functools.wraps98                        ret[key] = _argspec(99                            func_ref.__func__.__closure__[0].cell_contents100                        ).args101                    except (AttributeError, IndexError):102                        # functools.wraps changed (unlikely), bail out103                        ret[key] = []104                else:105                    try:106                        # functools.wraps makes things a little easier in Python 3107                        ret[key] = _argspec(func_ref.__wrapped__).args108                    except AttributeError:109                        # functools.wraps changed (unlikely), bail out110                        ret[key] = []111            except AttributeError:112                # Function moved, bail out113                ret[key] = []114    if not limit or "ipam_config" in limit:115        try:116            ret["ipam_config"] = _argspec(docker.types.IPAMPool.__init__).args117        except AttributeError:118            try:119                ret["ipam_config"] = _argspec(docker.utils.create_ipam_pool).args120            except AttributeError:121                raise CommandExecutionError("Could not get ipam args")122    for item in ret:123        # The API version is passed automagically by the API code that imports124        # these classes/functions and is not an arg that we will be passing, so125        # remove it if present. Similarly, don't include "self" if it shows up126        # in the arglist.127        for argname in ("version", "self"):128            try:129                ret[item].remove(argname)130            except ValueError:131                pass132    # Remove any args in host or endpoint config from the create_container133    # arglist. This keeps us from accidentally allowing args that docker-py has134    # moved from the create_container function to the either the host or135    # endpoint config.136    for item in ("host_config", "connect_container_to_network"):137        for val in ret.get(item, []):138            try:139                ret["create_container"].remove(val)140            except ValueError:141                # Arg is not in create_container arglist142                pass143    for item in ("create_container", "host_config", "connect_container_to_network"):144        if limit and item not in limit:145            ret.pop(item, None)146    try:147        ret["logs"].remove("container")148    except (KeyError, ValueError, TypeError):149        pass150    return ret151def translate_input(152    translator,153    skip_translate=None,154    ignore_collisions=False,155    validate_ip_addrs=True,156    **kwargs157):158    """159    Translate CLI/SLS input into the format the API expects. The ``translator``160    argument must be a module containing translation functions, within161    salt.utils.dockermod.translate. A ``skip_translate`` kwarg can be passed to162    control which arguments are translated. It can be either a comma-separated163    list or an iterable containing strings (e.g. a list or tuple), and members164    of that tuple will have their translation skipped. Optionally,165    skip_translate can be set to True to skip *all* translation.166    """167    kwargs = copy.deepcopy(salt.utils.args.clean_kwargs(**kwargs))168    invalid = {}169    collisions = []170    if skip_translate is True:171        # Skip all translation172        return kwargs173    else:174        if not skip_translate:175            skip_translate = ()176        else:177            try:178                skip_translate = _split(skip_translate)179            except AttributeError:180                pass181            if not hasattr(skip_translate, "__iter__"):182                log.error("skip_translate is not an iterable, ignoring")183                skip_translate = ()184    try:185        # Using list(kwargs) here because if there are any invalid arguments we186        # will be popping them from the kwargs.187        for key in list(kwargs):188            real_key = translator.ALIASES.get(key, key)189            if real_key in skip_translate:190                continue191            # ipam_pools is designed to be passed as a list of actual192            # dictionaries, but if each of the dictionaries passed has a single193            # element, it will be incorrectly repacked.194            if key != "ipam_pools" and salt.utils.data.is_dictlist(kwargs[key]):195                kwargs[key] = salt.utils.data.repack_dictlist(kwargs[key])196            try:197                kwargs[key] = getattr(translator, real_key)(198                    kwargs[key],199                    validate_ip_addrs=validate_ip_addrs,200                    skip_translate=skip_translate,201                )202            except AttributeError:203                log.debug("No translation function for argument '%s'", key)204                continue205            except SaltInvocationError as exc:206                kwargs.pop(key)207                invalid[key] = exc.strerror208        try:209            translator._merge_keys(kwargs)210        except AttributeError:211            pass212        # Convert CLI versions of commands to their docker-py counterparts213        for key in translator.ALIASES:214            if key in kwargs:215                new_key = translator.ALIASES[key]216                value = kwargs.pop(key)217                if new_key in kwargs:218                    collisions.append(new_key)219                else:220                    kwargs[new_key] = value221        try:222            translator._post_processing(kwargs, skip_translate, invalid)223        except AttributeError:224            pass225    except Exception as exc:  # pylint: disable=broad-except226        error_message = exc.__str__()227        log.error("Error translating input: '%s'", error_message, exc_info=True)228    else:229        error_message = None230    error_data = {}231    if error_message is not None:232        error_data["error_message"] = error_message233    if invalid:234        error_data["invalid"] = invalid235    if collisions and not ignore_collisions:236        for item in collisions:237            error_data.setdefault("collisions", []).append(238                "'{0}' is an alias for '{1}', they cannot both be used".format(239                    translator.ALIASES_REVMAP[item], item240                )241            )242    if error_data:243        raise CommandExecutionError("Failed to translate input", info=error_data)244    return kwargs245def create_ipam_config(*pools, **kwargs):246    """247    Builds an IP address management (IPAM) config dictionary248    """249    kwargs = salt.utils.args.clean_kwargs(**kwargs)250    try:251        # docker-py 2.0 and newer252        pool_args = salt.utils.args.get_function_argspec(253            docker.types.IPAMPool.__init__254        ).args255        create_pool = docker.types.IPAMPool256        create_config = docker.types.IPAMConfig257    except AttributeError:258        # docker-py < 2.0259        pool_args = salt.utils.args.get_function_argspec(260            docker.utils.create_ipam_pool261        ).args262        create_pool = docker.utils.create_ipam_pool263        create_config = docker.utils.create_ipam_config264    for primary_key, alias_key in (("driver", "ipam_driver"), ("options", "ipam_opts")):265        if alias_key in kwargs:266            alias_val = kwargs.pop(alias_key)267            if primary_key in kwargs:268                log.warning(269                    "docker.create_ipam_config: Both '%s' and '%s' "270                    "passed. Ignoring '%s'",271                    alias_key,272                    primary_key,273                    alias_key,274                )275            else:276                kwargs[primary_key] = alias_val277    if salt.utils.data.is_dictlist(kwargs.get("options")):278        kwargs["options"] = salt.utils.data.repack_dictlist(kwargs["options"])279    # Get all of the IPAM pool args that were passed as individual kwargs280    # instead of in the *pools tuple281    pool_kwargs = {}282    for key in list(kwargs):283        if key in pool_args:284            pool_kwargs[key] = kwargs.pop(key)285    pool_configs = []286    if pool_kwargs:287        pool_configs.append(create_pool(**pool_kwargs))288    pool_configs.extend([create_pool(**pool) for pool in pools])289    if pool_configs:290        # Sanity check the IPAM pools. docker-py's type/function for creating291        # an IPAM pool will allow you to create a pool with a gateway, IP292        # range, or map of aux addresses, even when no subnet is passed.293        # However, attempting to use this IPAM pool when creating the network294        # will cause the Docker Engine to throw an error.295        if any("Subnet" not in pool for pool in pool_configs):296            raise SaltInvocationError("A subnet is required in each IPAM pool")297        else:298            kwargs["pool_configs"] = pool_configs299    ret = create_config(**kwargs)300    pool_dicts = ret.get("Config")301    if pool_dicts:302        # When you inspect a network with custom IPAM configuration, only303        # arguments which were explictly passed are reflected. By contrast,304        # docker-py will include keys for arguments which were not passed in305        # but set the value to None. Thus, for ease of comparison, the below306        # loop will remove all keys with a value of None from the generated307        # pool configs.308        for idx, _ in enumerate(pool_dicts):309            for key in list(pool_dicts[idx]):310                if pool_dicts[idx][key] is None:311                    del pool_dicts[idx][key]...

__init__.pyi

Source:__init__.pyi

1"""This type stub file was generated by pyright."""2# pylint: disable=C,E,W,R3from __future__ import annotations4from docker.utils.build import create_archive, exclude_paths, mkbuildcontext, tar5from docker.utils.decorators import check_resource, minimum_version, update_headers6from docker.utils.utils import (7    compare_version,8    convert_filters,9    convert_port_bindings,10    convert_service_networks,11    convert_volume_binds,12    create_host_config,13    create_ipam_config,14    create_ipam_pool,15    datetime_to_timestamp,16    decode_json_header,17    format_environment,18    format_extra_hosts,19    kwargs_from_env,20    normalize_links,21    parse_bytes,22    parse_devices,23    parse_env_file,24    parse_host,25    parse_repository_tag,26    split_command,27    version_gte,28    version_lt,29)30__all__ = [31    "check_resource",32    "compare_version",33    "convert_filters",34    "convert_port_bindings",35    "convert_service_networks",36    "convert_volume_binds",37    "create_archive",38    "create_host_config",39    "create_ipam_config",40    "create_ipam_pool",41    "datetime_to_timestamp",42    "decode_json_header",43    "exclude_paths",44    "format_environment",45    "format_extra_hosts",46    "kwargs_from_env",47    "minimum_version",48    "mkbuildcontext",49    "normalize_links",50    "parse_bytes",51    "parse_devices",52    "parse_env_file",53    "parse_host",54    "parse_repository_tag",55    "split_command",56    "tar",57    "update_headers",58    "version_gte",59    "version_lt",...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.