Testing Azure SQL Server connection with Python

Testing connectivity from AKS clusters to Azure Micrsoft SQL Server Endpoint can be a cumbersome task as we need to download and install different tools to work with respective Database. In this example, we will use a simple python script that use the pymssql official library to execute a connection on the provided endpoint with the username, password and database name provided as environment variables.

We will use a nginx based Pod/VM where will install the following components:

• Python3
• Python3-pip
• pymssql (pip Library)

# Installation

We can use a standard nginx Pod or a preconfigured one as described in the Source Code section

apt update -y

apt install python3 -y

apt install python3-pip

pip install pymssql

mkdir /app
cd /app

cat << EOF > ./azuresqltest.py
import pymssql
import os
import time
from datetime import datetime

def checkAzureSql(server, user, password, database):
  try:
    now = datetime.now()
    dt_string = now.strftime("%d/%m/%Y %H:%M:%S")
    conn = pymssql.connect(server=server, user=user, password=password, database=database)
    cursor = conn.cursor()
    print("Successfull Connected to Azure SQL")
    conn.close()
    file1= open("azsqltest.log","a")
    content = dt_string + " " + "Successfull Connected to Azure SQL"
    file1.write(content)
    file1.write("\n")
    file1.close()
  except:
    print(dt_string + " " + "A MSSQLDriverException has been caught.")
    file1= open("azsqltest.log","a")
    content = dt_string + " " + "A MSSQLDriverException has been caught."
    file1.write(content)
    file1.write("\n")
    file1.close()
if __name__ == '__main__':
  print("Starting AzureSQL Connectivity Check:...")
  myHostname = os.getenv('SQLSERVER')
  myPGUser = os.getenv("SQLUSER")
  myPassword = os.environ.get('SQLPASS')
  myDB = os.environ.get('SQLDB')
  interval = os.environ.get('TIMEINTERVAL')
  while True:
    checkAzureSql(myHostname, myPGUser, myPassword, myDB)
    time.sleep(int(interval))
EOF

Adding the Environment Variables that holds our SQL endpoint, username, password and database name along with the time interval in second:

export SQLSERVER="yourslqname.database.windows.net"
export SQLUSER="yourskluser@yoursqlname.database.windows.net"
export SQLPASS="yoursqlpassword"
export SQLDB="yousql_db_name"
export TIMEINTERVAL="5"

Run your script with the following command:

python3 ./azuresqltest.py

If successful connection, it will output the following:

Starting AzureSQL Connectivity Check: ... 
Successful Connected to Azure SQL

Otherwise, it will throw an error like this:

Starting AzureSQL Connectivity Check:...
10/11/2022 14:50:53 A MSSQLDriverException has been caught.

It will also log the requests in a local file on Pod, azsqltest.log. For data persistency please mount an emptyDir and share the script’s folder with the underlying Node.

Ready to run Pod environment

apiVersion: v1
kind: Pod
metadata:
  creationTimestamp: null
  labels:
    run: azuresqltest
  name: azuresqltest
spec:
  containers:
  - image: nginx
    name: azuresqltest
    command: ["/bin/bash"]
    args: ["-c", "apt update && apt install python3 python3-pip nano -y && pip install pymssql && sleep 3600"]
    resources:
      requests:
        memory: "1024Mi"
        cpu: "650m"
      limits:
        memory: "2048Mi"
        cpu: "1000m"

  dnsPolicy: ClusterFirst
  restartPolicy: Always
status: {}

Script Content:

import pymssql
import os
import time
from datetime import datetime

def checkAzureSql(server, user, password, database):
  try:
    now = datetime.now()
    dt_string = now.strftime("%d/%m/%Y %H:%M:%S")
    conn = pymssql.connect(server=server, user=user, password=password, database=database)
    cursor = conn.cursor()
    print("Successfull Connected to Azure SQL")
    conn.close()
    file1= open("azsqltest.log","a")
    content = dt_string + " " + "Successfull Connected to Azure SQL"
    file1.write(content)
    file1.write("\n")
    file1.close()
  except:
    print(dt_string + " " + "A MSSQLDriverException has been caught.")
    file1= open("azsqltest.log","a")
    content = dt_string + " " + "A MSSQLDriverException has been caught."
    file1.write(content)
    file1.write("\n")
    file1.close()
if __name__ == '__main__':
  print("Starting AzureSQL Connectivity Check:...")
  myHostname = os.getenv('SQLSERVER')
  myPGUser = os.getenv("SQLUSER")
  myPassword = os.environ.get('SQLPASS')
  myDB = os.environ.get('SQLDB')
  interval = os.environ.get('TIMEINTERVAL')
  while True:
    checkAzureSql(myHostname, myPGUser, myPassword, myDB)
    time.sleep(int(interval))

GitHub Repository: https://github.com/OvidiuBorlean/azsqltest