[Python] Setting up alerts on SQL Server Job Agent failures

Level of Difficulty: Beginner to Intermediate.

So you’ve found yourself in a semi-ops situation where you need to do daily checks on SQL Server Agent Jobs because the client, Steve, has concerns that his data is not updating.

Now you have to go through the manual process of opening SQL Server Management Studio then logging into the server, clicking on the SQL Server Agent and actually checking the status of the jobs every so often when, in reality, you could just get notified when something fails. Sounds like a good idea right? Here’s how to do it!

What are the steps?

The steps that we will be following are:

1. Create the Python script
   1. Import Libraries
   2. Enter server details
   3. Select and configure a database engine
   4. Get jobs with their statuses
   5. Filter for failures that occurred today
2. Set up alerting
3. Schedule Python

Deep Dive

Let’s dive deeper into the steps listed above.

Create The Python Script

Create a Python script named JobMonitoring.py.

Ensure that the pandas and either pyodbc (ensure sqlalchemy is also installed if you are using pyodbc) or pymssql libraries are installed. You can install these libraries by executing the following commands:

Using pymssql

pip install pandas
pip install pymssql

Using pyodbc

pip install pandas
pip install pyodbc
pip install sqlalchemy

Import libraries

Use the code below to import the correct libraries if you are using pyodbc:

import pandas as pd
import pymssql
import pyodbc
from sqlalchemy import create_engine

Use the code below to import the correct libraries if you are using pymssql:

import pandas as pd
import pymssql

Enter Server Details

Add the necessary information to the JobMonitoring.py script or read from a config file (recommended). Ensure that the database is set to read from msdb, unless there’s a stored procedure that lives elsewhere:

server = ''
user = ''
password = ''
database = 'msdb'

Configure Database Engine

If you are using pyodbc, configure the server using the following code:

params = urllib.parse.quote_plus(r'DRIVER={};SERVER={};DATABASE={};UID={};PWD={}'.format(driver, server, database, user, password))
conn_str = 'mssql+pyodbc:///?odbc_connect={}'.format(params)
conn = create_engine(conn_str)

If you are using pyodbc, configure the server using the following code:

conn = pymssql.connect(server, user, password, database)

If you are using Windows authentication to get into the SQL Server, remove user and password from either of the above methods.

Get Jobs with Statuses

For this portion of the code, we will be relying on one of the functions built into the msdb, sp_help_jobactivity. The following code needs to be added to read the results into a pandas dataframe:

df = pd.read_sql('EXEC msdb.dbo.sp_help_jobactivity;', conn)

Are you facing permission challenges?

Ask the admin of the database to create a stored procedure for you to run and replace the msdb.dbo.sp_help_jobactivity in the code above with SQLAgentJobHistory. The stored procedure should look as follows:

CREATE PROC SQLAgentJobHistory
AS
BEGIN

	EXEC msdb.dbo.sp_help_jobactivity; 

END;

Filter for Failures that Occurred Today

Back in JobMonitoring.py, add the following code to apply the filters:

# order by last_executed_step_date
df['last_executed_step_date'] = pd.to_datetime(df['last_executed_step_date']).apply(lambda x: x.date())
df['last_executed_step_date'] = df['last_executed_step_date'].astype(str)

# create a dataframe that contains jobs executed today
df_today = df[df['last_executed_step_date'] == datetime.today().strftime("%Y-%m-%d")]

# create a dataframe that contains the jobs that have failed
df_failed = df_today[df_today['run_status'] == 0]

Setup Alerting

Inside of JobMonitoring.py, create a method named alert:

def alert(job_name, last_run):
    
    # add method of alert, ie. MS Teams, Slack, API call, etc.
    print('{} failed at {}'.format(job_name, last_run))

Check out this repo for ways of incorporating alerting via certain instant messaging platforms.

Add the following code to call the alert method that was just created:

if len(df_failed) > 0:
    
    for index, element in df_failed.iterrows():
        
        alert(element['job_name'], element['last_executed_step_date'])

Schedule Python Script

Let’s schedule this script to run on your local PC or server, using Task Scheduler.

1. Hit Start and search for Task Scheduler
2. Under action items, hit Create Task…
3. Give your task a name and description
4. Navigate to the Triggers tab and hit New…
5. Customise the schedule as necessary and make sure that Enabled is checked
6. Navigate to the Actions tab and click on New…
7. Assign the Program/Script to your default interpreter, here’s how to find your interpreter
8. Grab the path of the JobMonitoring.py script and paste it in Arguments
9. Run through the settings in the rest of the tabs to see if there’s anything else you’d like to customise to your need.
10. Select OK
11. Now right click on the job in Task Scheduler and select Run

The above code is available in this Github repo.

Consider your job scheduled! Did you get stuck somewhere or is there anything you want to add? Drop a comment or send your question/ suggestion through to jacqui.jm77@gmail.com