IntenseWebs/pgadmin4
3
0
Fork 0
mirror of https://github.com/pgadmin-org/pgadmin4.git synced 2025-02-25 18:55:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

Fix syntax error when creating new pgAgent schedules with a start

Browse Source 
date/time and exception. Fixes #3638

- Fixed syntax error.
- Update/Delete exceptions are not working, SQL is wrong.
- Wrong exception time is shown, it should shown in UTC as datatype of
  the column is timestamp with out timezone.
- Added test cases for Schedule and Steps of pgAgent Job.
This commit is contained in:
Akshay Joshi 2018-10-17 12:28:31 +01:00 committed by Dave Page
parent 6fa6939f52
commit 50738db2b8
11 changed files with 703 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
web/pgadmin/browser/server_groups/servers/pgagent/schedules/static/js/pga_schedule.js
View File

@ -151,8 +151,8 @@ define('pgadmin.node.pga_schedule', [
options: {format: 'HH:mm', buttons: {
showToday: false,
}}, displayFormat: 'HH:mm',
modelFormat: 'HH:mm:ss', displayInUTC: false, allowEmpty: true,
cellHeaderClasses:'width_percent_50', modalInUTC: false,
modelFormat: 'HH:mm:ss', displayInUTC: true, allowEmpty: true,
cellHeaderClasses:'width_percent_50', modalInUTC: true,
}],
validate: function() {
var self = this, exceptions = this.collection,
@ -356,7 +356,7 @@ define('pgadmin.node.pga_schedule', [
}),
schema:[{
id: 'jscweekdays', label: gettext('Week Days'), cell: 'select2',
group: gettext('Days'), control: 'select2',
group: gettext('Days'), control: 'select2', type: 'array',
select2: {
first_empty: false,
multiple: true,
@ -372,7 +372,7 @@ define('pgadmin.node.pga_schedule', [
options: BooleanArrayOptions,
},{
id: 'jscmonthdays', label: gettext('Month Days'), cell: 'select2',
group: gettext('Days'), control: 'select2',
group: gettext('Days'), control: 'select2', type: 'array',
select2: {
first_empty: false,
multiple: true,
@ -387,7 +387,7 @@ define('pgadmin.node.pga_schedule', [
selector: monthdays, options: BooleanArrayOptions,
},{
id: 'jscmonths', label: gettext('Months'), cell: 'select2',
group: gettext('Days'), control: 'select2',
group: gettext('Days'), control: 'select2', type: 'array',
select2: {
first_empty: false,
multiple: true,
@ -406,7 +406,7 @@ define('pgadmin.node.pga_schedule', [
group: gettext('Repeat'), mode: ['create', 'edit'],
schema:[{
id: 'jschours', label: gettext('Hours'), cell: 'select2',
group: gettext('Times'), control: 'select2',
group: gettext('Times'), control: 'select2', type: 'array',
select2: {
first_empty: false,
multiple: true,
@ -421,7 +421,7 @@ define('pgadmin.node.pga_schedule', [
selector: hours, options: BooleanArrayOptions,
},{
id: 'jscminutes', label: gettext('Minutes'), cell: 'select2',
group: gettext('Times'), control: 'select2',
group: gettext('Times'), control: 'select2', type: 'array',
select2: {
first_empty: false,
multiple: true,

8
web/pgadmin/browser/server_groups/servers/pgagent/templates/macros/pga_exception.macros
View File

@ -7,18 +7,18 @@
INSERT INTO pgagent.pga_exception (
jexscid, jexdate, jextime
) VALUES (
{% if jscid %}{{ jscid|qtLiteral }}{% else %}scid{% endif %}, {% if data.jexdate %}to_date({{ data.jexdate|qtLiteral }}, 'MM/DD/YYYY')::date{% else %}NULL::date{% endif %}, {% if data.jextime %}{{ data.jextime|qtLiteral }}::time without time zone{% else %}NULL::time without time zone{% endif %}
{% if jscid %}{{ jscid|qtLiteral }}{% else %}scid{% endif %}, {% if data.jexdate %}to_date({{ data.jexdate|qtLiteral }}, 'YYYY-MM-DD')::date{% else %}NULL::date{% endif %}, {% if data.jextime %}{{ data.jextime|qtLiteral }}::time without time zone{% else %}NULL::time without time zone{% endif %}
);
{%- endmacro %}
{% macro UPDATE(jscid, data) -%}
-- Updating an existing schedule exception (id: {{ data.jexid|qtLiteral }}, schedule: {{ jscid|qtLiteral }})
UPDATE pgagent.pga_exception SET
{% if 'jexdate' in data %}jexdate={% if data.jexdate %}to_date({{ data.jexdate|qtLiteral }}, 'MM/DD/YYYY')::date{% else %}NULL::date{% endif %}{% endif %}{% if 'jextime' in data%}{% if 'jexdate' in data %}, {% endif %}jextime={% if data.jextime %}{{ data.jextime|qtLiteral }}::time without time zone{% else %}NULL::time without time zone{% endif %}{% endif %}
{% if 'jexdate' in data %}jexdate={% if data.jexdate %}to_date({{ data.jexdate|qtLiteral }}, 'YYYY-MM-DD')::date{% else %}NULL::date{% endif %}{% endif %}{% if 'jextime' in data%}{% if 'jexdate' in data %}, {% endif %}jextime={% if data.jextime %}{{ data.jextime|qtLiteral }}::time without time zone{% else %}NULL::time without time zone{% endif %}{% endif %}
WHERE jexid={{ data.jexid|qtLiteral }}::integer AND jscid={{ jscid|qtLiteral }}::integer;
WHERE jexid={{ data.jexid|qtLiteral }}::integer AND jexscid={{ jscid|qtLiteral }}::integer;
{%- endmacro %}
{% macro DELETE(jscid, data) -%}
-- Deleting a schedule exception (id: {{ data.jexid|qtLiteral }}, schedule: {{ jscid|qtLiteral }})
DELETE FROM pgagent.pga_exception WHERE jexid={{ data.jexid|qtLiteral }}::integer AND jscid={{ jscid|qtLiteral }}::integer;
DELETE FROM pgagent.pga_exception WHERE jexid={{ data.jexid|qtLiteral }}::integer AND jexscid={{ jscid|qtLiteral }}::integer;
{%- endmacro %}

2
web/pgadmin/browser/server_groups/servers/pgagent/templates/macros/pga_schedule.macros
View File

@ -23,7 +23,7 @@ INSERT INTO pgagent.pga_schedule(
{{ data.jscmonthdays|qtLiteral }}::boolean[],
-- Months
{{ data.jscmonths|qtLiteral }}::boolean[]
) RETURNING jscid {% if not jid %}INTO scid;{% endif %}{% if 'jscexceptions' in data %}
) RETURNING jscid INTO scid;{% if 'jscexceptions' in data %}
{% for exc in data.jscexceptions %}
{{ EXCEPTIONS.INSERT(None, exc) }}{% endfor %}{% endif %}

6
web/pgadmin/browser/server_groups/servers/pgagent/templates/pga_schedule/sql/pre3.4/create.sql
View File

@ -1,2 +1,8 @@
{% import 'macros/pga_schedule.macros' as SCHEDULE %}
DO $$
DECLARE
scid integer;
BEGIN
{{ SCHEDULE.INSERT(jid, data) }}
END
$$;

80
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_add_schedule.py Normal file
View File

@ -0,0 +1,80 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentAddScheduleTestCase(BaseTestGenerator):
"""This class will test the add schedule in the pgAgent job API"""
scenarios = [
('Create schedule with exception in pgAgent job', dict(
url='/browser/pga_schedule/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
# Check and Delete entry for pga_exception table for the above
# date and time as no primary key is defined for pga_exception table
# and there is a unique constraint for date and time. So when we run
# the test cases multiple time then it will fail with unique
# constraint error.
jexdate = '2050-01-01'
jextime = '12:00:00'
pgagent_utils.delete_pgagent_exception(self, jexdate, jextime)
self.pgagent_schedule_name = "test_sch_add%s" % str(uuid.uuid4())[1:8]
data = {
'jscjobid': self.job_id,
'jscenabled': True,
'jscdesc': '',
'jscname': self.pgagent_schedule_name,
'jscexceptions': [{'jexdate': jexdate,
'jextime': jextime}],
'jscstart': '2050-01-01 12:14:21 +05:30',
'jscend': '2050-03-01 12:14:21 +05:30',
'jscminutes': [False] * 60,
'jscweekdays': [True] * 7,
'jscmonthdays': [True] * 32,
'jschours': [False] * 24,
'jscmonths': [True] * 12
}
response = self.tester.post(
'{0}{1}/{2}/{3}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
data=json.dumps(data),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
response_data = json.loads(response.data)
self.schedule_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_schedule(self)
self.assertTrue(
is_present, "pgAgent schedule was not created successfully"
)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

69
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_add_steps.py Normal file
View File

@ -0,0 +1,69 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentAddStepTestCase(BaseTestGenerator):
"""This class will test the add step in the pgAgent job API"""
scenarios = [
('Create step for pgAgent job', dict(
url='/browser/pga_jobstep/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_get%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
def runTest(self):
self.pgagent_step_name = "test_step_add%s" % str(uuid.uuid4())[1:8]
data = {
'jstjobid': self.job_id,
'jstname': self.pgagent_step_name,
'jstdesc': '',
'jstenabled': True,
'jstkind': True,
'jstconntype': True,
'jstcode': 'SELECT 1;',
'jstconnstr': None,
'jstdbname': 'postgres',
'jstonerror': 'f',
'jstnextrun': '',
}
response = self.tester.post(
'{0}{1}/{2}/{3}/'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id)
),
data=json.dumps(data),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
response_data = json.loads(response.data)
self.step_id = response_data['node']['_id']
is_present = pgagent_utils.verify_pgagent_step(self)
self.assertTrue(
is_present, "pgAgent step was not created successfully"
)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

52
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_delete_schedule.py Normal file
View File

@ -0,0 +1,52 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentDeleteScheduleTestCase(BaseTestGenerator):
"""This class will test the delete pgAgent job schedule API"""
scenarios = [
('Delete pgAgent Schedule', dict(url='/browser/pga_schedule/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_delete%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_delete%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
def runTest(self):
"""This function will deletes pgAgent job schedule"""
response = self.tester.delete(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.schedule_id)
),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
is_present = pgagent_utils.verify_pgagent_schedule(self)
self.assertFalse(
is_present, "pgAgent schedule was not deleted successfully"
)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

52
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_delete_steps.py Normal file
View File

@ -0,0 +1,52 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentDeleteStepTestCase(BaseTestGenerator):
"""This class will test the delete pgAgent job step API"""
scenarios = [
('Delete pgAgent Step', dict(url='/browser/pga_jobstep/obj/'))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_delete%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name = "test_step_delete%s" % str(uuid.uuid4())[1:8]
self.step_id = pgagent_utils.create_pgagent_step(
self, step_name, self.job_id)
def runTest(self):
"""This function will deletes pgAgent job step"""
response = self.tester.delete(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.step_id)
),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
is_present = pgagent_utils.verify_pgagent_step(self)
self.assertFalse(
is_present, "pgAgent step was not deleted successfully"
)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_job(self)

131
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_put_schedule.py Normal file
View File

@ -0,0 +1,131 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentPutScheduleTestCase(BaseTestGenerator):
"""This class will test the update pgAgent schedule API"""
scenarios = [
('Update schedule with start and end time', dict(
url='/browser/pga_schedule/obj/',
data={
'jscdesc': 'Test Schedule',
'jscstart': '2050-01-01 12:00:00 +05:30',
'jscend': '2050-01-20 12:00:00 +05:30',
})),
('Update schedule with repeat', dict(
url='/browser/pga_schedule/obj/',
data={
'jscmonthdays': '[true,false,true,false,true,false,false,'
'false,false,false,false,false,false,false,'
'false,false,false,false,false,false,false,'
'false,false,false,false,false,false,false,'
'false,false,false,false]',
'jscweekdays': '[true,false,false,true,false,false,false]',
'jscmonths': '[true,false,false,true,false,false,false,false,'
'false,false,false,false]',
'jschours': '[false,false,false,false,true,false,false,false,'
'false,false,false,false,false,false,false,false,'
'false,false,false,false,false,false,false,false]'
})),
('Update schedule add exception', dict(
url='/browser/pga_schedule/obj/',
data={
'jscexceptions': {
'added': [{'jexdate': '2050-01-01',
'jextime': '12:00:00'}]
}},
delete_existing_exception=True)),
('Update schedule change exception date and time', dict(
url='/browser/pga_schedule/obj/',
data={
'jscexceptions': {
'changed': [{'jexdate': '2050-01-31',
'jextime': '20:00:00'}]
}},
create_exception=True)),
('Update schedule delete exception', dict(
url='/browser/pga_schedule/obj/',
data={
'jscexceptions': {
'deleted': [{'jexdate': '2050-01-01',
'jextime': '12:00:00'}]
}},
create_exception=True,
is_delete=True)),
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_update%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
sch_name = "test_schedule_update%s" % str(uuid.uuid4())[1:8]
self.schedule_id = pgagent_utils.create_pgagent_schedule(
self, sch_name, self.job_id)
def runTest(self):
"""This function will update pgAgent schedule"""
# Check and Delete entry for pga_exception table for the specified
# date and time as no primary key is defined for pga_exception table
# and there is a unique constraint for date and time. So when we run
# the test cases multiple time then it will fail with unique
# constraint error.
if hasattr(self, 'delete_existing_exception'):
pgagent_utils.delete_pgagent_exception(
self, self.data['jscexceptions']['added'][0]['jexdate'],
self.data['jscexceptions']['added'][0]['jextime'])
# Create exception for update and delete
if hasattr(self, 'create_exception'):
date = None
time = None
if hasattr(self, 'is_delete'):
date = self.data['jscexceptions']['deleted'][0]['jexdate']
time = self.data['jscexceptions']['deleted'][0]['jextime']
else:
date = self.data['jscexceptions']['changed'][0]['jexdate']
time = self.data['jscexceptions']['changed'][0]['jextime']
self.excp_id = pgagent_utils.create_pgagent_exception(
self, self.schedule_id, date, time)
# Add created exception id in data
if hasattr(self, 'is_delete'):
self.data['jscexceptions']['deleted'][0]['jexid'] = \
self.excp_id
else:
self.data['jscexceptions']['changed'][0]['jexid'] = \
self.excp_id
self.data['jscid'] = str(self.schedule_id)
response = self.tester.put(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.schedule_id)
),
data=json.dumps(self.data),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_schedule(self)
pgagent_utils.delete_pgagent_job(self)

67
web/pgadmin/browser/server_groups/servers/pgagent/tests/tests_pgagent_put_step.py Normal file
View File

@ -0,0 +1,67 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
import simplejson as json
import uuid
from pgadmin.utils.route import BaseTestGenerator
from regression.python_test_utils import test_utils as utils
from . import utils as pgagent_utils
class PgAgentPutStepTestCase(BaseTestGenerator):
"""This class will test the update pgAgent steps API"""
scenarios = [
('Update step with kind, description, code and error', dict(
url='/browser/pga_jobstep/obj/',
data={
'jstdesc': 'Test Steps',
'jstkind': False,
'jstcode': 'SELECT 12345',
'jstonerror': 'i'
})),
('Update step with connection type and string', dict(
url='/browser/pga_jobstep/obj/',
data={
'jstconntype': False,
'jstconnstr':
'host=localhost port=5432 dbname=mydb connect_timeout=10'
}))
]
def setUp(self):
flag, msg = pgagent_utils.is_valid_server_to_run_pgagent(self)
if not flag:
self.skipTest(msg)
flag, msg = pgagent_utils.is_pgagent_installed_on_server(self)
if not flag:
self.skipTest(msg)
name = "test_job_update%s" % str(uuid.uuid4())[1:8]
self.job_id = pgagent_utils.create_pgagent_job(self, name)
step_name = "test_step_update%s" % str(uuid.uuid4())[1:8]
self.step_id = pgagent_utils.create_pgagent_step(
self, step_name, self.job_id)
def runTest(self):
"""This function will update pgAgent steps"""
self.data['jstid'] = str(self.step_id)
response = self.tester.put(
'{0}{1}/{2}/{3}/{4}'.format(
self.url, str(utils.SERVER_GROUP), str(self.server_id),
str(self.job_id), str(self.step_id)
),
data=json.dumps(self.data),
content_type='html/json'
)
self.assertEquals(response.status_code, 200)
def tearDown(self):
"""Clean up code"""
pgagent_utils.delete_pgagent_step(self)
pgagent_utils.delete_pgagent_job(self)

235
web/pgadmin/browser/server_groups/servers/pgagent/tests/utils.py
View File

@ -151,7 +151,7 @@ def delete_pgagent_job(self):
def verify_pgagent_job(self):
"""
This function deletes the pgAgent job.
This function verifies the pgAgent job.
"""
try:
connection = utils.get_db_connection(
@ -173,3 +173,236 @@ def verify_pgagent_job(self):
return count is not None and int(count) != 0
except Exception:
traceback.print_exc(file=sys.stderr)
def create_pgagent_schedule(self, sch_name, jobid):
"""
This function create the pgAgent schedule.
"""
try:
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
query = """
INSERT INTO pgagent.pga_schedule(
jscname, jscjobid, jscenabled, jscdesc, jscstart, jscend
) VALUES (
'{0}'::text, {1}::int, true, '',
'2050-01-01 12:14:21 +05:30'::timestamp with time zone,
'2050-01-30 12:14:21 +05:30'::timestamp with time zone
) RETURNING jscid;
""".format(sch_name, jobid)
pg_cursor.execute(query)
sch_id = pg_cursor.fetchone()
connection.set_isolation_level(old_isolation_level)
connection.commit()
connection.close()
return sch_id[0]
except Exception:
traceback.print_exc(file=sys.stderr)
def delete_pgagent_schedule(self):
"""
This function deletes the pgAgent schedule.
"""
try:
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
pg_cursor.execute(
"DELETE FROM pgagent.pga_schedule "
"WHERE jscid = '%s'::integer;" % self.schedule_id
)
connection.set_isolation_level(old_isolation_level)
connection.commit()
connection.close()
except Exception:
traceback.print_exc(file=sys.stderr)
def verify_pgagent_schedule(self):
"""
This function verifies the pgAgent schedule.
"""
try:
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
pg_cursor = connection.cursor()
pg_cursor.execute(
"SELECT COUNT(*) FROM pgagent.pga_schedule "
"WHERE jscid = '%s'::integer;" % self.schedule_id
)
result = pg_cursor.fetchone()
count = result[0]
connection.close()
return count is not None and int(count) != 0
except Exception:
traceback.print_exc(file=sys.stderr)
def delete_pgagent_exception(self, date, time):
"""
This function deletes the pgAgent exception.
"""
try:
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
pg_cursor = connection.cursor()
query = "DELETE FROM pgagent.pga_exception " \
"WHERE jexdate = to_date('{0}', 'YYYY-MM-DD') AND " \
"jextime = '{1}'::time without time zone;".format(date, time)
pg_cursor.execute(query)
connection.close()
except Exception:
traceback.print_exc(file=sys.stderr)
def create_pgagent_exception(self, schid, date, time):
"""
This function create the pgAgent exception.
"""
try:
# Delete existing exception if already exists
delete_pgagent_exception(self, date, time)
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
query = """
INSERT INTO pgagent.pga_exception(jexscid, jexdate, jextime
) VALUES ({0},
to_date('{1}', 'YYYY-MM-DD'), '{2}'::time without time zone
) RETURNING jexid;
""".format(schid, date, time)
pg_cursor.execute(query)
excep_id = pg_cursor.fetchone()
connection.set_isolation_level(old_isolation_level)
connection.commit()
connection.close()
return excep_id[0]
except Exception:
traceback.print_exc(file=sys.stderr)
def create_pgagent_step(self, step_name, jobid):
"""
This function create the pgAgent step.
"""
try:
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
query = """
INSERT INTO pgagent.pga_jobstep(
jstname, jstjobid, jstenabled, jstkind,
jstcode, jstdbname
) VALUES (
'{0}'::text, {1}::int, true, 's', 'SELECT 1', 'postgres'
) RETURNING jstid;
""".format(step_name, jobid)
pg_cursor.execute(query)
step_id = pg_cursor.fetchone()
connection.set_isolation_level(old_isolation_level)
connection.commit()
connection.close()
return step_id[0]
except Exception:
traceback.print_exc(file=sys.stderr)
def delete_pgagent_step(self):
"""
This function deletes the pgAgent step.
"""
try:
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
old_isolation_level = connection.isolation_level
connection.set_isolation_level(0)
pg_cursor = connection.cursor()
pg_cursor.execute(
"DELETE FROM pgagent.pga_jobstep "
"WHERE jstid = '%s'::integer;" % self.step_id
)
connection.set_isolation_level(old_isolation_level)
connection.commit()
connection.close()
except Exception:
traceback.print_exc(file=sys.stderr)
def verify_pgagent_step(self):
"""
This function verifies the pgAgent step .
"""
try:
connection = utils.get_db_connection(
self.server['db'],
self.server['username'],
self.server['db_password'],
self.server['host'],
self.server['port'],
self.server['sslmode']
)
pg_cursor = connection.cursor()
pg_cursor.execute(
"SELECT COUNT(*) FROM pgagent.pga_jobstep "
"WHERE jstid = '%s'::integer;" % self.step_id
)
result = pg_cursor.fetchone()
count = result[0]
connection.close()
return count is not None and int(count) != 0
except Exception:
traceback.print_exc(file=sys.stderr)