Using MySQL in Docker for local testing in Python

Imagine you’re writing a web service that uses MySQL. You deploy on Linux but write code and test on Mac.

What is a good way to setup a MySQL database for local testing?

You can use native MySQL installer or install MySQL via Homebrew but my favorite way is to use docker.

With docker you can have multiple MySQL instances running and you can easily install exactly the same version of MySQL as the one in production. I’ve learned the hard way that using newer version for development can bite you.

One wrinkle in the scheme is that you have to make sure that the database container is running, get the ip address of docker instance and port on which the database is listening on.

Doing this manually would be annoying and I like to automate. I wrote a re-usable script that ensures MySQL docker instance is running and then passes database ip address/port to the program.

It’s just a matter of running docker commands and parsing their outputs but it’s difficult enough to worth sharing the complete solution.

Conceptually, what we do is:

run docker ps -a and parse the output. If the desired container is already running, parse ip address/port from the output and return that
if the container is stopped, re-start it with docker start
otherwise start it with docker run

MySQL database is stored in a mounted directory so that it persists even if the container is stopped.

The script is very re-usable. You can customize it by changing:

the base MySQL container (mysql:5.6 in my case)
where MySQL data is stored
name of the container, which should by unique for the project
port on which the database is exposed locally (in the container MySQL listens on standard port 3306, it must be exposed locally on a unique port)

This can be adapted for other databases, like PostgreSQL.

Implementation in Python

Python scripts works a bit differently in that it extracts ip address of the docker instance and port on which MySQL is listening and writes this information to stdout. We can parse its output in run.sh and pass it to arbitrary program.

Let’s start with run.sh script that starts a local instance.

#!/bin/bash
set -u -e -o pipefail

# ./start_docker.py writes '<ip_address>:<db port>' to stdout
IFS=\; read -a ip_port <<<"`./start_docker.py`"
ip="${ip_port[0]}"
port="${ip_port[1]}"

./quicknotes -local -verbose -db-host ${ip} -db-port ${port} $@

Assume that ./quicknotes is the executable for the service.

The real meat of the operation is start_docker.py script which checks if docker instance is already running. If not, it starts it.

start-mysql-in-docker/start_docker.py:

#!/usr/local/bin/python3

import sys, os, os.path, time, subprocess

kStatusRunning = "running"
kStatusExited = "exited"

# using https://hub.docker.com/_/mysql/
# to use the latest mysql, use mysql:8
imageName = "mysql:5.6"
# name must be unique across containers runing on this computer
containerName = "mysql-db-multi"
# this is where mysql database files are stored, so that
# they persist even if container goes away
dbDir = os.path.expanduser("~/data/db-multi")
# 3306 is standard MySQL port, I use a unique port to be able
# to run multiple mysql instances for different projects
dockerDbLocalPort = "7200"

def eprint(*args, **kwargs):
  print(*args, file=sys.stderr, **kwargs)

def print_cmd(cmd):
  eprint("cmd:" + " ".join(cmd))

def run_cmd(cmd):
  print_cmd(cmd)
  res = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  eprint(res.stdout.decode('utf-8'))

def run_cmd_out(cmd):
  print_cmd(cmd)
  s = subprocess.check_output(cmd)
  return s.decode("utf-8")

def run_cmd_show_progress(cmd):
  eprint("Running '%s'" % cmd)
  p = subprocess.Popen(cmd, stdout = subprocess.PIPE,
          stderr = subprocess.STDOUT, shell = True)
  while True:
    line = p.stdout.readline()
    if not line:
      break
    sys.stdout.buffer.write(line)
    sys.stdout.flush()
  #eprint("Finished runnign '%s'" % " ".join(cmd))

def verify_docker_running():
  try:
    run_cmd(["docker", "ps"])
  except:
    eprint("docker is not running! must run docker")
    sys.exit(10)

# not sure if this covers all cases
def decode_status(status_verbose):
  if "Exited" in status_verbose:
    return kStatusExited
  return kStatusRunning

# given:
# 0.0.0.0:7200->3306/tcp
# return (0.0.0.0, 7200) or None if doesn't match
def decode_ip_port(mappings):
  parts = mappings.split("->")
  if len(parts) != 2:
    return None
  parts = parts[0].split(":")
  if len(parts) != 2:
    return None
  return parts

# returns:
#  - container id
#  - status
#  - (ip, port) in the host that maps to exposed port inside the container (or None)
# returns (None, None, None) if no container of that name
def docker_container_info(containerName):
  s = run_cmd_out(["docker", "ps", "-a", "--format", "{{.ID}}|{{.Status}}|{{.Ports}}|{{.Names}}"])
  # this returns a line like:
  # 6c5a934e00fb|Exited (0) 3 months ago|0.0.0.0:7200->3306/tcp|mysql-56-for-quicknotes
  lines = s.split("\n")
  for l in lines:
    if len(l) == 0:
      continue
    parts = l.split("|")
    assert len(parts) == 4, "parts: %s" % parts
    id, status, mappings, names = parts
    if containerName in names:
      status = decode_status(status)
      ip_port = decode_ip_port(mappings)
      return (id, status, ip_port)
  return (None, None, None)

def wait_for_container(containerName):
  # 8 secs is a heuristic
  timeOut = 8
  eprint("waiting %s secs for container to start" % timeOut, end="", flush=True)
  while timeOut > 0:
    (containerId, status, ip_port) = docker_container_info(containerName)
    if status == kStatusRunning:
      return
    eprint(".", end="", flush=True)
    time.sleep(1)
    timeOut -= 1
  eprint("")

def start_container_if_needed(imageName, containerName, portMapping):
  (containerId, status, ip_port) = docker_container_info(containerName)
  if status == kStatusRunning:
    eprint("container %s is already running" % containerName)
    return
  if status == kStatusExited:
    cmd = ["docker", "start", containerId]
  else:
    volumeMapping = "%s:/var/lib/mysql" % dbDir
    cmd = ["docker", "run", "-d", "--name=" + containerName, "-p", portMapping, "-v", volumeMapping, "-e", "MYSQL_ALLOW_EMPTY_PASSWORD=yes", imageName]
  run_cmd(cmd)
  wait_for_container(containerName)

def create_db_dir():
  try:
    os.makedirs(dbDir)
  except:
    # throws if already exists, which is ok
    pass

def main():
  verify_docker_running()
  create_db_dir()
  start_container_if_needed(imageName, containerName, dockerDbLocalPort + ":3306")
  (containerId, status, ip_port) = docker_container_info(containerName)
  assert ip_port is not None
  ip, port = ip_port
  print("%s;%s" % (ip, port))

if __name__ == "__main__":
  main()