KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use publish_invocation method in localstack

Best Python code snippet using localstack_python

localstack.py

Source:localstack.py

1import json2import logging3import os4import sys5from typing import Dict, List, Optional6from localstack import config7from localstack.utils.analytics.cli import publish_invocation8if sys.version_info >= (3, 8):9    from typing import TypedDict10else:11    from typing_extensions import TypedDict12import click13from localstack import __version__14from .console import BANNER, console15from .plugin import LocalstackCli, load_cli_plugins16def create_with_plugins() -> LocalstackCli:17    """18    Creates a LocalstackCli instance with all cli plugins loaded.19    :return: a LocalstackCli instance20    """21    cli = LocalstackCli()22    cli.group = localstack23    load_cli_plugins(cli)24    return cli25def _setup_cli_debug():26    from localstack.logging.setup import setup_logging_for_cli27    config.DEBUG = True28    os.environ["DEBUG"] = "1"29    setup_logging_for_cli(logging.DEBUG if config.DEBUG else logging.INFO)30@click.group(name="localstack", help="The LocalStack Command Line Interface (CLI)")31@click.version_option(version=__version__, message="%(version)s")32@click.option("--debug", is_flag=True, help="Enable CLI debugging mode")33@click.option("--profile", type=str, help="Set the configuration profile")34def localstack(debug, profile):35    if profile:36        os.environ["CONFIG_PROFILE"] = profile37    if debug:38        _setup_cli_debug()39    from localstack.utils.files import cache_dir40    # overwrite the config variable here to defer import of cache_dir41    if not config.LEGACY_DIRECTORIES and not os.environ.get("LOCALSTACK_VOLUME_DIR", "").strip():42        config.VOLUME_DIR = cache_dir() / "volume"43@localstack.group(name="config", help="Inspect your LocalStack configuration")44def localstack_config():45    pass46@localstack.group(47    name="status",48    help="Print status information about the LocalStack runtime",49    invoke_without_command=True,50)51@click.pass_context52def localstack_status(ctx):53    if ctx.invoked_subcommand is None:54        ctx.invoke(localstack_status.get_command(ctx, "docker"))55@localstack_status.command(56    name="docker", help="Query information about the LocalStack Docker image and runtime"57)58@click.option("--format", type=click.Choice(["table", "plain", "dict", "json"]), default="table")59@publish_invocation60def cmd_status_docker(format):61    with console.status("Querying Docker status"):62        print_docker_status(format)63@localstack_status.command(name="services", help="Query information about running services")64@click.option("--format", type=click.Choice(["table", "plain", "dict", "json"]), default="table")65@publish_invocation66def cmd_status_services(format):67    import requests68    url = config.get_edge_url()69    try:70        health = requests.get(f"{url}/health", timeout=2)71        doc = health.json()72        services = doc.get("services", [])73        if format == "table":74            print_service_table(services)75        if format == "plain":76            for service, status in services.items():77                console.print(f"{service}={status}")78        if format == "dict":79            console.print(services)80        if format == "json":81            console.print(json.dumps(services))82    except requests.ConnectionError:83        error = f"could not connect to LocalStack health endpoint at {url}"84        print_error(format, error)85        if config.DEBUG:86            console.print_exception()87        sys.exit(1)88@localstack.command(name="start", help="Start LocalStack")89@click.option("--docker", is_flag=True, help="Start LocalStack in a docker container (default)")90@click.option("--host", is_flag=True, help="Start LocalStack directly on the host")91@click.option("--no-banner", is_flag=True, help="Disable LocalStack banner", default=False)92@click.option(93    "-d", "--detached", is_flag=True, help="Start LocalStack in the background", default=False94)95@publish_invocation96def cmd_start(docker: bool, host: bool, no_banner: bool, detached: bool):97    if docker and host:98        raise click.ClickException("Please specify either --docker or --host")99    if host and detached:100        raise click.ClickException("Cannot start detached in host mode")101    if not no_banner:102        print_banner()103        print_version()104        console.line()105    from localstack.utils import bootstrap106    if not no_banner:107        if host:108            console.log("starting LocalStack in host mode :laptop_computer:")109        else:110            console.log("starting LocalStack in Docker mode :whale:")111    bootstrap.prepare_host()112    if not no_banner and not detached:113        console.rule("LocalStack Runtime Log (press [bold][yellow]CTRL-C[/yellow][/bold] to quit)")114    if host:115        bootstrap.start_infra_locally()116    else:117        if detached:118            bootstrap.start_infra_in_docker_detached(console)119        else:120            bootstrap.start_infra_in_docker()121@localstack.command(name="stop", help="Stop the running LocalStack container")122@publish_invocation123def cmd_stop():124    from localstack.utils.docker_utils import DOCKER_CLIENT125    from ..utils.container_utils.container_client import NoSuchContainer126    container_name = config.MAIN_CONTAINER_NAME127    try:128        DOCKER_CLIENT.stop_container(container_name)129        console.print("container stopped: %s" % container_name)130    except NoSuchContainer:131        console.print("no such container: %s" % container_name)132        sys.exit(1)133@localstack.command(name="logs", help="Show the logs of the LocalStack container")134@click.option(135    "-f",136    "--follow",137    is_flag=True,138    help="Block the terminal and follow the log output",139    default=False,140)141@publish_invocation142def cmd_logs(follow: bool):143    from localstack.utils.bootstrap import LocalstackContainer144    from localstack.utils.docker_utils import DOCKER_CLIENT145    container_name = config.MAIN_CONTAINER_NAME146    logfile = LocalstackContainer(container_name).logfile147    if not DOCKER_CLIENT.is_container_running(container_name):148        console.print("localstack container not running")149        if os.path.exists(logfile):150            console.print("printing logs from previous run")151            with open(logfile) as fd:152                for line in fd:153                    click.echo(line, nl=False)154        sys.exit(1)155    if follow:156        for line in DOCKER_CLIENT.stream_container_logs(container_name):157            print(line.decode("utf-8").rstrip("\r\n"))158    else:159        print(DOCKER_CLIENT.get_container_logs(container_name))160@localstack.command(name="wait", help="Wait on the LocalStack container to start")161@click.option(162    "-t",163    "--timeout",164    type=float,165    help="The amount of time in seconds to wait before raising a timeout error",166    default=None,167)168@publish_invocation169def cmd_wait(timeout: Optional[float] = None):170    from localstack.utils.bootstrap import wait_container_is_ready171    if not wait_container_is_ready(timeout=timeout):172        raise click.ClickException("timeout")173@localstack_config.command(174    name="validate", help="Validate your LocalStack configuration (e.g., your docker-compose.yml)"175)176@click.option(177    "--file",178    default="docker-compose.yml",179    type=click.Path(exists=True, file_okay=True, readable=True),180)181@publish_invocation182def cmd_config_validate(file):183    from rich.panel import Panel184    from localstack.utils import bootstrap185    try:186        if bootstrap.validate_localstack_config(file):187            console.print("[green]:heavy_check_mark:[/green] config valid")188            sys.exit(0)189        else:190            console.print("[red]:heavy_multiplication_x:[/red] validation error")191            sys.exit(1)192    except Exception as e:193        console.print(Panel(str(e), title="[red]Error[/red]", expand=False))194        console.print("[red]:heavy_multiplication_x:[/red] validation error")195        sys.exit(1)196@localstack_config.command(name="show", help="Print the current LocalStack config values")197@click.option("--format", type=click.Choice(["table", "plain", "dict", "json"]), default="table")198@publish_invocation199def cmd_config_show(format):200    # TODO: parse values from potential docker-compose file?201    from localstack_ext import config as ext_config202    assert config203    assert ext_config204    if format == "table":205        print_config_table()206    elif format == "plain":207        print_config_pairs()208    elif format == "dict":209        print_config_dict()210    elif format == "json":211        print_config_json()212    else:213        print_config_pairs()  # fall back to plain214def print_config_json():215    import json216    console.print(json.dumps(dict(config.collect_config_items())))217def print_config_pairs():218    for key, value in config.collect_config_items():219        console.print(f"{key}={value}")220def print_config_dict():221    console.print(dict(config.collect_config_items()))222def print_config_table():223    from rich.table import Table224    grid = Table(show_header=True)225    grid.add_column("Key")226    grid.add_column("Value")227    for key, value in config.collect_config_items():228        grid.add_row(key, str(value))229    console.print(grid)230@localstack.command(name="ssh", help="Obtain a shell in the running LocalStack container")231@publish_invocation232def cmd_ssh():233    from localstack.utils.docker_utils import DOCKER_CLIENT234    from localstack.utils.run import run235    if not DOCKER_CLIENT.is_container_running(config.MAIN_CONTAINER_NAME):236        raise click.ClickException(237            'Expected a running container named "%s", but found none' % config.MAIN_CONTAINER_NAME238        )239    try:240        process = run("docker exec -it %s bash" % config.MAIN_CONTAINER_NAME, tty=True)241        process.wait()242    except KeyboardInterrupt:243        pass244@localstack.group(name="update", help="Update LocalStack components")245def localstack_update():246    pass247@localstack_update.command(name="all", help="Update all LocalStack components")248@click.pass_context249@publish_invocation250def cmd_update_all(ctx):251    ctx.invoke(localstack_update.get_command(ctx, "localstack-cli"))252    ctx.invoke(localstack_update.get_command(ctx, "docker-images"))253@localstack_update.command(name="localstack-cli", help="Update LocalStack CLI tools")254@publish_invocation255def cmd_update_localstack_cli():256    import subprocess257    from subprocess import CalledProcessError258    console.rule("Updating LocalStack CLI")259    with console.status("Updating LocalStack CLI..."):260        try:261            subprocess.check_output(262                [sys.executable, "-m", "pip", "install", "--upgrade", "localstack"]263            )264            console.print(":heavy_check_mark: LocalStack CLI updated")265        except CalledProcessError:266            console.print(":heavy_multiplication_x: LocalStack CLI update failed", style="bold red")267@localstack_update.command(268    name="docker-images", help="Update container images LocalStack depends on"269)270@publish_invocation271def cmd_update_docker_images():272    from localstack.utils.docker_utils import DOCKER_CLIENT273    console.rule("Updating docker images")274    all_images = DOCKER_CLIENT.get_docker_image_names(strip_latest=False)275    image_prefixes = ["localstack/", "lambci/lambda:", "mlupin/docker-lambda:"]276    localstack_images = [277        image278        for image in all_images279        if any(280            image.startswith(image_prefix) or image.startswith(f"docker.io/{image_prefix}")281            for image_prefix in image_prefixes282        )283    ]284    update_images(localstack_images)285def update_images(image_list: List[str]):286    from rich.markup import escape287    from rich.progress import MofNCompleteColumn, Progress288    from localstack.utils.container_utils.container_client import ContainerException289    from localstack.utils.docker_utils import DOCKER_CLIENT290    updated_count = 0291    failed_count = 0292    progress = Progress(293        *Progress.get_default_columns(), MofNCompleteColumn(), transient=True, console=console294    )295    with progress:296        for image in progress.track(image_list, description="Processing image..."):297            try:298                updated = False299                hash_before_pull = DOCKER_CLIENT.inspect_image(image_name=image, pull=False)["Id"]300                DOCKER_CLIENT.pull_image(image)301                if (302                    hash_before_pull303                    != DOCKER_CLIENT.inspect_image(image_name=image, pull=False)["Id"]304                ):305                    updated = True306                    updated_count += 1307                console.print(308                    f":heavy_check_mark: Image {escape(image)} {'updated' if updated else 'up-to-date'}.",309                    style="bold" if updated else None,310                    highlight=False,311                )312            except ContainerException as e:313                console.print(314                    f":heavy_multiplication_x: Image {escape(image)} pull failed: {e.message}",315                    style="bold red",316                    highlight=False,317                )318                failed_count += 1319    console.rule()320    console.print(321        f"Images updated: {updated_count}, Images failed: {failed_count}, total images processed: {len(image_list)}."322    )323# legacy support324@localstack.group(325    name="infra",326    help="Manipulate LocalStack infrastructure (legacy)",327)328def infra():329    pass330@infra.command("start")331@click.pass_context332@click.option("--docker", is_flag=True, help="Start LocalStack in a docker container (default)")333@click.option("--host", is_flag=True, help="Start LocalStack directly on the host")334@publish_invocation335def cmd_infra_start(ctx, *args, **kwargs):336    ctx.invoke(cmd_start, *args, **kwargs)337class DockerStatus(TypedDict, total=False):338    running: bool339    runtime_version: str340    image_tag: str341    image_id: str342    image_created: str343    container_name: Optional[str]344    container_ip: Optional[str]345def print_docker_status(format):346    from localstack.utils import docker_utils347    from localstack.utils.bootstrap import get_docker_image_details, get_server_version348    from localstack.utils.container_networking import get_main_container_ip, get_main_container_name349    img = get_docker_image_details()350    cont_name = config.MAIN_CONTAINER_NAME351    running = docker_utils.DOCKER_CLIENT.is_container_running(cont_name)352    status = DockerStatus(353        runtime_version=get_server_version(),354        image_tag=img["tag"],355        image_id=img["id"],356        image_created=img["created"],357        running=running,358    )359    if running:360        status["container_name"] = get_main_container_name()361        status["container_ip"] = get_main_container_ip()362    if format == "dict":363        console.print(status)364    if format == "table":365        print_docker_status_table(status)366    if format == "json":367        console.print(json.dumps(status))368    if format == "plain":369        for key, value in status.items():370            console.print(f"{key}={value}")371def print_docker_status_table(status: DockerStatus):372    from rich.table import Table373    grid = Table(show_header=False)374    grid.add_column()375    grid.add_column()376    grid.add_row("Runtime version", f'[bold]{status["runtime_version"]}[/bold]')377    grid.add_row(378        "Docker image",379        f"tag: {status['image_tag']}, "380        f"id: {status['image_id']}, "381        f":calendar: {status['image_created']}",382    )383    cont_status = "[bold][red]:heavy_multiplication_x: stopped"384    if status["running"]:385        cont_status = (386            f"[bold][green]:heavy_check_mark: running[/green][/bold] "387            f'(name: "[italic]{status["container_name"]}[/italic]", IP: {status["container_ip"]})'388        )389    grid.add_row("Runtime status", cont_status)390    console.print(grid)391def print_service_table(services: Dict[str, str]):392    from rich.table import Table393    status_display = {394        "running": "[green]:heavy_check_mark:[/green] running",395        "starting": ":hourglass_flowing_sand: starting",396        "available": "[grey]:heavy_check_mark:[/grey] available",397        "error": "[red]:heavy_multiplication_x:[/red] error",398    }399    table = Table()400    table.add_column("Service")401    table.add_column("Status")402    services = list(services.items())403    services.sort(key=lambda item: item[0])404    for service, status in services:405        if status in status_display:406            status = status_display[status]407        table.add_row(service, status)408    console.print(table)409def print_version():410    console.print(" :laptop_computer: [bold]LocalStack CLI[/bold] [blue]%s[/blue]" % __version__)411def print_error(format, error):412    if format == "table":413        symbol = "[bold][red]:heavy_multiplication_x: ERROR[/red][/bold]"414        console.print(f"{symbol}: {error}")415    if format == "plain":416        console.print(f"error={error}")417    if format == "dict":418        console.print({"error": error})419    if format == "json":420        console.print(json.dumps({"error": error}))421def print_banner():...

cli.py

Source:cli.py

...26    while parent is not None:27        parent_commands.insert(0, parent.command.name)28        parent = parent.parent29    return parent_commands30def publish_invocation(fn):31    """32    Decorator for capturing CLI commands from Click and publishing them to the backend as analytics events.33    This decorator should only be used on outermost subcommands, e.g. "localstack status docker" not "localstack status"34    otherwise it may publish multiple events for a single invocation.35    If DISABLE_EVENTS is set then nothing is collected.36    For performance reasons, the API call to the backend runs on a separate process and is killed if it takes longer37    than ANALYTICS_API_RESPONSE_TIMEOUT_SECS.38    The emitted event contains the invoked command, plus any parameter names if their associated values are truthy (but39    not the values themselves).40    """41    @functools.wraps(fn)42    def publisher_wrapper(*args, **kwargs):43        if config.DISABLE_EVENTS:44            return fn(*args, **kwargs)...

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.