initialize_target(debug_type, sid, did, scid, func_id, tri_id) This method is responsible for creating an asynchronous connection. Parameters: debug_type - Type of debugging (Direct or Indirect) sid - Server Id did - Database Id
(debug_type, trans_id, sid, did,
scid, func_id, tri_id=None)
| 835 | ) |
| 836 | @pga_login_required |
| 837 | def initialize_target(debug_type, trans_id, sid, did, |
| 838 | scid, func_id, tri_id=None): |
| 839 | """ |
| 840 | initialize_target(debug_type, sid, did, scid, func_id, tri_id) |
| 841 | |
| 842 | This method is responsible for creating an asynchronous connection. |
| 843 | |
| 844 | Parameters: |
| 845 | debug_type |
| 846 | - Type of debugging (Direct or Indirect) |
| 847 | sid |
| 848 | - Server Id |
| 849 | did |
| 850 | - Database Id |
| 851 | scid |
| 852 | - Schema Id |
| 853 | func_id |
| 854 | - Function Id |
| 855 | tri_id |
| 856 | - Trigger Function Id |
| 857 | """ |
| 858 | |
| 859 | # Create asynchronous connection using random connection id. |
| 860 | conn_id = str(secrets.choice(range(1, 9999999))) |
| 861 | manager = get_driver(PG_DEFAULT_DRIVER).connection_manager(sid) |
| 862 | conn = manager.connection(did=did, conn_id=conn_id) |
| 863 | data_obj = {} |
| 864 | |
| 865 | # Connect the Server |
| 866 | status, msg = conn.connect() |
| 867 | if not status: |
| 868 | return internal_server_error(errormsg=str(msg)) |
| 869 | |
| 870 | user = manager.user_info |
| 871 | status, error = validate_debug(conn, debug_type, user['is_superuser']) |
| 872 | if not status: |
| 873 | return error |
| 874 | |
| 875 | if tri_id is not None: |
| 876 | # Find trigger function id from trigger id |
| 877 | sql = render_template( |
| 878 | "/".join([DEBUGGER_SQL_PATH, 'get_trigger_function_info.sql']), |
| 879 | table_id=func_id, trigger_id=tri_id |
| 880 | ) |
| 881 | |
| 882 | status, tr_set = conn.execute_dict(sql) |
| 883 | if not status: |
| 884 | current_app.logger.debug( |
| 885 | "Error retrieving trigger function information from database") |
| 886 | return internal_server_error(errormsg=tr_set) |
| 887 | |
| 888 | func_id = tr_set['rows'][0]['tgfoid'] |
| 889 | |
| 890 | status, search_path = get_search_path(conn) |
| 891 | if not status: |
| 892 | return search_path |
| 893 | |
| 894 | # Find out the debugger version and store it in session variables |
nothing calls this directly
no test coverage detected