Python Snippet: Start Docker Compose in Context Manager

Waits for /readyz

import subprocess
import time
import requests

class DockerComposeRunner:

    def __init__(self, project_folder):
        self.project_folder = project_folder

    def __enter__(self):
        command = ['docker-compose', 'up', '-d', '--build']

        self.port = self.find_free_port()

        # Set the PORT environment variable for docker-compose
        env = os.environ.copy()
        env['PORT'] = str(self.port)

        self.process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            cwd=self.project_folder,
            env=env
        )
        
        self.wait_for_readyz(self.port)

        return self.port
    
    @staticmethod
    def find_free_port():
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(('', 0))
            return s.getsockname()[1]
   
    @staticmethod
    def wait_for_readyz(port, timeout=60, check_every=0.2):
        url = f"http://localhost:{port}/readyz"
        timeout_time = time.time() + timeout

        while True:
            try:
                response = requests.get(url, timeout=1)
                if response.status_code == 200:
                    print('Service is ready')
                    break
            except (requests.ConnectionError, requests.Timeout):
                pass  # Ignore connection errors and timeouts, retrying until timeout is reached

            if time.time() > timeout_time:
                raise TimeoutError(f"Timeout while waiting for /readyz on port {port}")

            time.sleep(check_every)


    def __exit__(self, exc_type, exc_val, exc_tb):
        command = ['docker-compose', 'down']
        subprocess.run(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            cwd=self.project_folder
        )


with DockerComposeRunner('.') as port:
    print(f"API running on port {port}")
    # Perform your tasks here
    # ...