Skip to content

Instantly share code, notes, and snippets.

View kuharan's full-sized avatar
🏠
Working from home

KUHARAN BHOWMIK kuharan

🏠
Working from home
View GitHub Profile
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")
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'
}
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;">&lt; 1 Day&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;&nbsp;</span></td>
</tr>
<tr style="height: 18px;">
<td style="width: 100%; height: 18px;">
<div>
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)
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 = []
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)
@kuharan
kuharan / get_all_specifications.py
Created July 11, 2021 12:11
get all specification configured in data protector
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']
@kuharan
kuharan / data_protector_login.py
Created July 11, 2021 12:07
login to data protector
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
@kuharan
kuharan / rest_example.ps1
Created May 22, 2021 19:33
This powershell snippet makes a api call to url with a json body converted from csv.
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;
}
}
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",