This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import signal | |
def timeout_handler(_signal, _frame): | |
global unprocess_bucket | |
global unprocess_file | |
logger.info("Time exceeded! Creating Unprocessed File.") | |
session = boto3.Session() | |
s3_client = session.client(service_name="s3") |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def send_email(html_payload): | |
url = "https://xxxx.execute-api.us-east-1.amazonaws.com/default/lambda-function-sendEmail" | |
payload = json.dumps({ | |
"html": html_payload | |
}) | |
headers = { | |
'Content-Type': 'application/json' | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def create_html(json_data): | |
html_body = """<html><body><p style="text-align: center;"><span style="text-decoration: underline;"><strong>Data Protector XXXX Cell Manager No Backup Report</strong></span></p> | |
<table style="border-collapse: collapse; width: 9%; height: 85px; margin-left: auto; margin-right: auto;" border="1"> | |
<tbody> | |
<tr style="height: 18px;"> | |
<td style="width: 100%; height: 18px; text-align: left;"><span style="background-color: #258721;">< 1 Day </span></td> | |
</tr> | |
<tr style="height: 18px;"> | |
<td style="width: 100%; height: 18px;"> | |
<div> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def calculate_days_since_last_run(all_completed_runs): | |
for run in all_completed_runs: | |
dt1 = datetime.datetime.fromtimestamp(int(time.time())) | |
dt2 = datetime.datetime.fromtimestamp(run['endTime']) | |
rd = dateutil.relativedelta.relativedelta (dt1, dt2) | |
run['time_since_last_sucess_run'] = str(rd.years) + '-' + str(rd.months) + '-' + str(rd.days) + '-' + str(rd.hours) + '-' + str(rd.minutes) | |
return all_completed_runs | |
all_completed_runs = calculate_days_since_last_run(all_completed_runs) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def find_last_success_run(access_token, session, all_backup_specification): | |
url = "https://xxxx.com:7116/idb/sessions/filter/" | |
headers = { | |
'Authorization': 'Bearer {}'.format(access_token), | |
'Content-Type': 'application/json' | |
} | |
all_completed_runs = [] | |
sessions_not_found = [] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def filter_specification(all_specifications): | |
filtered = [item for item in all_specifications if item['jobType'].lower()=='backup'] | |
return filtered | |
all_backup_specification = filter_specification(all_specifications) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def get_all_specifications(access_token, session): | |
url = 'https://xxxx.com:7116/dp-gui/dp-scheduler-gui/restws/specification' | |
headers = { | |
'Authorization': 'Bearer {}'.format(access_token), | |
'Content-Type': 'application/json' | |
} | |
response = json.loads(session.get(url, headers=headers).text) | |
return response['specifications'] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
def login(): | |
session = requests.session() | |
url = 'https://xxxx.com:7116/auth/realms/DataProtector/protocol/openid-connect/token' | |
payload = {'username':'username|*|xxxx.com', 'password':'password.', 'client_id':'dp-gui', 'grant_type':'password'} | |
response = json.loads(session.post(url, data=payload,verify='xxxx.com_cacert.pem').text) | |
access_token = response['access_token'] | |
return access_token, session |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
add-type @" | |
using System.Net; | |
using System.Security.Cryptography.X509Certificates; | |
public class TrustAllCertsPolicy : ICertificatePolicy { | |
public bool CheckValidationResult( | |
ServicePoint srvPoint, X509Certificate certificate, | |
WebRequest request, int certificateProblem) { | |
return true; | |
} | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
import json | |
def open_incidents_test(incident): | |
url = "https:/xxxx/api/koi/servicenow_incident_api/update" | |
payload = json.dumps({ | |
"Incident_Update": { | |
"number": incident, | |
"state": "New", |
NewerOlder