In Python, how do I construct a URL to test if my SQL Server instance is running?

user_1235929 0
Viewed 811

I'm using Python 3.8 with the pytest-docker-compose plugin -- https://pypi.org/project/pytest-docker-compose/ . Does anyone know how to write a URL that would eventually tell me if my SQL Server is running?

I have this docker-compose.yml file

version: "3.2"
services:

  sql-server-db:
    build: ./
    container_name: sql-server-db
    image: microsoft/mssql-server-linux:2017-latest
    ports:
      - "1433:1433"
    environment:
      SA_PASSWORD: "password"
      ACCEPT_EULA: "Y"

but I don't know what URL to pass to my Retry object to test that the server is running. This fails ...

import pytest
import requests
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
...

@pytest.fixture(scope="function")
def wait_for_api(function_scoped_container_getter):
    """Wait for sql server to become responsive"""
    request_session = requests.Session()
    retries = Retry(total=5,
                    backoff_factor=0.1,
                    status_forcelist=[500, 502, 503, 504])
    request_session.mount('http://', HTTPAdapter(max_retries=retries))

    service = function_scoped_container_getter.get("sql-server-db").network_info[0]
    api_url = "http://%s:%s/" % (service.hostname, service.host_port)
    assert request_session.get(api_url)
    return request_session, api_url

with this exception

raise ConnectionError(e, request=request)
E           requests.exceptions.ConnectionError: HTTPConnectionPool(host='0.0.0.0', port=1433): Max retries exceeded with url: / (Caused by ProtocolError('Connection aborted.', RemoteDisconnected('Remote end closed connection without response')))
urllib3 docker-compose pytest python-3.x sql-server
2 Answers
if connection.is_connected():
    db_Info = connection.get_server_info()
    print("Connected to MySQL Server version ", db_Info)
    cursor = connection.cursor()
    cursor.execute("select database();")
    record = cursor.fetchone()
    print("You're connected to database: ", record)

you could use something like this and it would output if it was connected

Admin1

Here is a sample function that will retry to connect to the DB and won't return until it has successfully connected or the defined maxRetries is reached:

def waitDb(server, database, username, password, maxAttempts, waitBetweenAttemptsSeconds):
    """
    Returns True if the connection is successfully established before the maxAttempts number is reached
    Conversely returns False
    pyodbc.connect has a built-in timeout. Use a waitBetweenAttemptsSeconds greater than zero to add a delay on top of this timeout 
    """
    for attemptNumber in range(maxAttempts):
        cnxn = None
        try:
            cnxn = pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER='+server+';DATABASE='+database+';UID='+username+';PWD='+ password)
            cursor = cnxn.cursor()
        except Exception as e:
            print(traceback.format_exc())
        finally:
            if cnxn:
                print("The DB is up and running: ")
                return True
            else:
                print("DB not running yet on attempt numer " + str(attemptNumber))
            time.sleep(waitBetweenAttemptsSeconds)
    print("Max attempts waiting for DB to come online exceeded")
    return False

I wrote a minimal example here: https://github.com/claudiumocanu/flow-pytest-docker-compose-wait-mssql.

I included the three actions that can be executed independently, but you can jump to the last step specifically for what you asked:

1. Connected from python to the mssql launched by the compose-file:
For me it was quite annoying to find and install the appropriate ODBC Driver and its dependencies - ODBC Driver 17 for SQL Server worked the best for me on Ubuntu 18.
To perform only this step, docker-compose up the the docker-compose.yml in my example, then run the example-connect.py

2. Created a function that attempts to connect to the DB with a maxAttemptsNumber and a delay between retries:
Just run this example-waitDb.py. You can play with the maxAttempts and the delayBetweenAttempts values, then bring up the database at randomly, to test it.

3. Put everything together in the test_db.py test suite:

  • the waitDb function described above.
  • same wrapper and annotations that you provided in your example to spin-up the resources defined in the compose-file
  • a dummy integration test that will not be executed before waitDb returns (if you want to block this tests completely, you can throw instead of returning False from the waitDb function)

PS: Keep using ENVs/vault etc rather than storing the real passwords like I did for the dummy example.

user_137365250
Related