codekingpro/portable-devtools

1##########################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##########################################################################9 10"""pgagent helper utilities"""11from flask import render_template12 13 14def format_boolean_array(value):15    """16    Converts to proper array data for sql17    Args:18        value: data to be converted19 20    Returns:21        Converted data22    """23    if not isinstance(value, list):24        return value.replace("[", "{").replace("]", "}")25    return value26 27 28def format_schedule_data(data):29    """30    This function is used to format the schedule data. If data is not an31    instance of list then format32    :param data:33    :return:34    """35 36    if 'jscminutes' in data and data['jscminutes'] is not None:37        data['jscminutes'] = format_boolean_array(data['jscminutes'])38    if 'jschours' in data and data['jschours'] is not None:39        data['jschours'] = format_boolean_array(data['jschours'])40    if 'jscweekdays' in data and data['jscweekdays'] is not None:41        data['jscweekdays'] = format_boolean_array(data['jscweekdays'])42    if 'jscmonthdays' in data and data['jscmonthdays'] is not None:43        data['jscmonthdays'] = format_boolean_array(data['jscmonthdays'])44    if 'jscmonths' in data and data['jscmonths'] is not None:45        data['jscmonths'] = format_boolean_array(data['jscmonths'])46 47    return data48 49 50def format_step_data(job_id, data, has_connection_str, conn, template_path):51    """52    This function is used to format the step data. If data is not an53    instance of list then format54    :param job_id: Job ID55    :param data: a step data56    :param has_connection_str: has pgagent connection str57    :param conn: Connection obj58    :param conn: SQL template path59    """60    if 'jstconntype' not in data and \61        ('jstdbname' in data or62         'jstconnstr' in data) and has_connection_str:63        status, rset = conn.execute_dict(64            render_template(65                "/".join([template_path, 'steps.sql']),66                jid=job_id,67                jstid=data['jstid'],68                conn=conn,69                has_connstr=has_connection_str70            )71        )72        if not status:73            return False, rset74 75        row = rset['rows'][0]76        data['jstconntype'] = row['jstconntype']77        if row['jstconntype']:78            data['jstdbname'] = data.get(79                'jstdbname', row['jstdbname'])80        else:81            data['jstconnstr'] = data.get(82                'jstconnstr', row['jstconnstr'])83 84    return True, None85