starts an openseach container with python

es_container.py

import time

import docker
from docker.models.containers import Container

keep_container = False
docker_client = docker.from_env()


def wait_healthy(container_name):
    print(f"Waiting for ES container named {container_name} to become health")
    while True:
        try:
            container = docker_client.containers.get(container_name)
        except docker.errors.NotFound as e:
            raise ValueError(f"No container found with name {container_name}") from e

        state = container.attrs["State"]
        if not (health := state.get("Health")):
            raise ValueError("No healthcheck configured for that container")

        status = health.get("Status")
        if status is None:
            raise ValueError(state)

        if status == "healthy":
            return

        time.sleep(0.1)


def create_es_container(container_name) -> Container:
    print(f"Creating ES container named {container_name}")
    container = docker_client.containers.run(
        "opensearchproject/opensearch:2.5.0",
        name=container_name,
        detach=True,
        remove=not keep_container,
        ports={"9200/tcp": None, "9300/tcp": None},
        healthcheck={
            "test": [
                "CMD",
                "curl",
                "--silent",
                "--fail",
                "http://localhost:9200/_cluster/health",
            ],
            "interval": 500_000_000,  # 5 s
            "timeout": 500_000_000,  # 500 ms
            "retries": 6,
            "start_interval": 50_000_000,  # 50 ms
        },
        environment={
            "discovery.type": "single-node",
            "network.host": "0.0.0.0",
            "path.repo": "/usr/share/opensearch/backup",
            "plugins.security.disabled": "true",  # The boolean True would fail lol
        },
    )
    print(f"Created ES container named {container_name}")
    return container


def es_container(container_name) -> Container:
    try:
        container = docker_client.containers.get(container_name)
        print(f"Found existing ES container named {container_name}")
    except docker.errors.NotFound:
        container = create_es_container(container_name)

    wait_healthy(container_name)
    return container


def host_port(container: Container, container_port: int) -> int:
    container_name = container.name
    assert container_name is not None

    while True:
        container = docker_client.containers.get(container_name)
        ports = container.attrs.get("NetworkSettings", {}).get("Ports")
        if not ports:
            continue
        mapping = ports.get(f"{container_port}/tcp", [])
        if not mapping:
            continue
        host_port = mapping[0].get("HostPort")
        if not host_port:
            continue
        return int(host_port)


def es_port(container: Container) -> int:
    return host_port(container, 9200)


print(f"Starting ES Container")
start = time.time()
container = es_container("test-es")
elapsed = time.time() - start
print(f"ES Container became ready in {elapsed}")
port = es_port(container)
print(f"ES is listening on port: {port}")