[PYTHON] Correction de la base de données avec pytest-docker

Ce qui suit est préparé comme "Exemple d'environnement pytest pour réparer la base de données avec Docker".

• Article Qiita
• GitHub

Cet article était MySQL, mais je l'ai également vérifié avec PostgreSQL.

En tant que bibliothèque d'interface de connexion à la base de données,

--Pour MySQL: PyMySQL --Pour PostgreSQL: psycopg2

J'ai utilisé, mais je pense que c'est presque la même chose lorsque j'utilise d'autres choses.

Exemple de fixture dans conftest.py

Dans l'exemple de pytest-docker, en attente du démarrage de httpbin

def is_responsive(url):
    try:
        response = requests.get(url)
        if response.status_code == 200:
            return True
    except ConnectionError:
        return False


@pytest.fixture(scope="session")
def http_service(docker_ip, docker_services):
    """Ensure that HTTP service is up and responsive."""

    # `port_for` takes a container port and returns the corresponding host port
    port = docker_services.port_for("httpbin", 80)
    url = "http://{}:{}".format(docker_ip, port)
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_responsive(url)
    )
    return url

Version de la base de données de ceci. Du côté de l'application, même en utilisant ORM comme SQLAlchemy, j'ai essayé de me connecter avec la bibliothèque de connexion directe. Je ne suis pas sûr que ce soit le timing de create_engine () de SQLAlchemy via ORM. C'est comme indiquer clairement que vous vous connectez via la bibliothèque de pilotes et que vous vérifiez le démarrage.

Pour le conteneur MySQL

Attendez que le conteneur MySQL démarre et qu'il n'y ait aucune exception à connecter ()

def is_mysqld_ready(docker_ip):
    try:
        pymysql.connect(
            host=docker_ip,
            user=os.getenv('MYSQL_USER', ''),
            password=os.getenv('MYSQL_PASSWORD', ''),
            db=os.getenv('MYSQL_DATABASE', '')
        )
        return True
    except:
        return False

@pytest.fixture(scope="session")
def database_service(docker_ip, docker_services):
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_mysqld_ready(docker_ip)
    )
    return

Pour le conteneur PostgreSQL

C'est le même.

def is_postgresql_ready(docker_ip):
    try:
        psycopg2.connect(
            "postgresql://{user}:{password}@{host}/{db}".format(
                user=os.getenv('POSTGRES_USER', ''),
                password=os.getenv('POSTGRES_PASSWORD', ''),
                host=docker_ip,
                db=os.getenv('POSTGRES_DB', '')
            )
        )
        return True
    except:
        return False

@pytest.fixture(scope="session")
def database_service(docker_ip, docker_services):
    docker_services.wait_until_responsive(
        timeout=30.0, pause=0.1, check=lambda: is_postgresql_ready(docker_ip)
    )
    return

docker-compose.yml

#Pour MySQL
version: "3"

services:
  database:
    image: mysql:5.7
    ports:
      - 3306:3306
    volumes:
      - ./initdb.d:/docker-entrypoint-initdb.d
    environment:
      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
      - MYSQL_USER=${MYSQL_USER}
      - MYSQL_PASSWORD=${MYSQL_PASSWORD}
      - MYSQL_DATABASE=${MYSQL_DATABASE}

#Pour PostgreSQL
version: "3"

services:
  database:
    image: postgres:13
    ports:
      - 5432:5432
    volumes:
      - ./initdb.d:/docker-entrypoint-initdb.d
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=${POSTGRES_DB}

Préparation du schéma et des données de test

Ensuite, montez le volume avec docker-compose et initialisez-le. Placez SQL et le script shell d'initialisation, etc. dans le dossier . / Initdb.d de ./initdb.d: / docker-entrypoint-initdb.d. Si vous le faites, vous pouvez le tester.

L'article de Qiita inclut Initialiser les données lorsque MySQL démarre avec Docker.