Configure Snowflake UDF for Precisely Gateway API access - Precisely Data Integrity Suite

Configure network rules, external access integration, and a user-defined function (UDF) so Snowflake Pipeline Engines can access the Precisely Gateway API.

1. Create a network rule and external access integration at the account level to allow outbound connections to Precisely Gateway API endpoints.

   Run the following SQL commands:

   CREATE OR REPLACE NETWORK RULE precisely_gateway_api_network_rule
    MODE = EGRESS
    TYPE = HOST_PORT
    VALUE_LIST = ('gateway.dqcore.cloud.precisely.com','gateway.dqcore.au1.cloud.precisely.com','gateway.dqcore.eu1.cloud.precisely.com','gateway.dqcore.gb1.cloud.precisely.com');
   CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION precisely_gateway_api_integration
    ALLOWED_NETWORK_RULES = (precisely_gateway_api_network_rule)
    ENABLED = TRUE;

   This rule lets Snowflake connect to the specified Precisely Gateway API hosts.

2. Create the precisely_api_execute UDF for Pipeline Engines.

   Run the following SQL command:

   CREATE OR REPLACE FUNCTION DataBaseName.SchemaName.precisely_api_execute(input VARIANT, url VARCHAR, token VARCHAR)
     RETURNS VARIANT
     LANGUAGE PYTHON
     EXTERNAL_ACCESS_INTEGRATIONS = (precisely_gateway_api_integration)
     RUNTIME_VERSION = 3.9
     HANDLER = 'handle_calls'
     PACKAGES = ('pandas', 'requests')
   AS $$
   import pandas
   import numpy as np
   import json
   import requests
   import _snowflake
   from _snowflake import vectorized
   from requests.adapters import HTTPAdapter
   from requests.packages.urllib3.util.retry import Retry
   
   session = requests.Session()
   retries = Retry(total=10, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], allowed_methods = None)
   
   session.mount('https://', HTTPAdapter(max_retries=retries))
   
   @vectorized(input=pandas.DataFrame, max_batch_size=20)
   def handle_calls(df):
     try:
         url = df.iloc[0][1]
         token_value = df.iloc[0][2]
         session.headers.update({'Authorization': 'Bearer ' + str(token_value)})
   
         request_rows = df.iloc[:, 0].tolist()
         request_payload = {'requests': request_rows}
   
         response = session.post(url, json=request_payload)
   
         response.raise_for_status()
         response_payload = json.loads(response.text)
       
         response_rows = response_payload['responses']
       
         return response_rows
     except Exception as e:
         error_obj = {"__call_error": str(e)}
         return [error_obj for _ in range(len(df))]
   $$;

   This UDF sends batched requests to the Precisely Gateway API, returns the responses, and handles retries and errors automatically.

3. Optional: Grant a role permission to use the UDF.

   Run the following command:

   grant usage on function precisely_api_execute(VARIANT, VARCHAR, VARCHAR) to role <RoleName>;

   Ensure only authorized users and roles can access the UDF and external access integration.