PySnowHelper.py

import os
import pysnow
from typing import Dict, List, Optional, Any
from robot.libraries.BuiltIn import BuiltIn
from robot.api import logger
from robot.api.deco import keyword
from urllib.parse import urlparse

class PySnowHelper:
    """
    A helper library for ServiceNow operations using pysnow.

    Environment Variables Required:
    - SNOW_HOST: ServiceNow instance URL
    - SNOW_USERNAME: ServiceNow username
    - SNOW_PASSWORD: ServiceNow password
    """

    def __init__(self):
        self.environment = BuiltIn().get_variable_value('${ENVIRONMENT}')
        parsed_url = urlparse(self.environment)
        self.host = parsed_url.hostname
        self.username = BuiltIn().get_variable_value('${SNOW_USER_NAME}')
        self.password = BuiltIn().get_variable_value('${SNOW_PASSWORD}')

        if not all([self.host, self.username, self.password]):
            raise ValueError("Missing required environment variables: SNOW_HOST, SNOW_USER_NAME, SNOW_PASSWORD")

        self.client = pysnow.Client(
            host=self.host,
            user=self.username,
            password=self.password
        )

    @keyword
    def get_record_by_sys_id(self, table: str, sys_id: str, fields: Optional[List[str]] = None) -> Optional[Dict]:
        """
        Get a single record by sys_id.

        Args:
            table: ServiceNow table name
            sys_id: System ID of the record
            fields: Optional list of field names to return. If None, returns all fields.

        Returns:
            Dictionary containing record data or None if not found
        """
        try:
            resource = self.client.resource(api_path=f'/table/{table}')

            query_params = {
                'sys_id': sys_id,
            }

            if fields:
                response = resource.get(query=query_params, fields=fields, display_value=True, exclude_reference_link=True)
            else:
                response = resource.get(query=query_params, display_value=True, exclude_reference_link=True)

            records = response.all()
            data = records[0] if records else None
            logger.info(data)
            return data

        except Exception as e:
            logger.error(f"Error getting record by sys_id: {str(e)}")
            return None

    @keyword
    def query_records(self, table: str, query: Dict[str, Any], limit: Optional[int] = 1, fields: Optional[List[str]] = None) -> List[Dict]:
        """
        Query multiple records from a table.

        Args:
            table: ServiceNow table name
            query: Dictionary of field-value pairs for query
            limit: Maximum number of records to return
            fields: Optional list of field names to return. If None, returns all fields.

        Returns:
            List of dictionaries containing record data
        """
        try:
            resource = self.client.resource(api_path=f'/table/{table}')


            if fields:
                response = resource.get(query=query, fields=fields, display_value=True, exclude_reference_link=True, limit=limit)
            else:
                response = resource.get(query=query, display_value=True, exclude_reference_link=True, limit=limit)
            data = response.all()
            logger.info(data)
            return data

        except Exception as e:
            logger.error(f"Error querying records: {str(e)}")
            return []

    @keyword
    def query_with_encoded_query(self, table: str, encoded_query: str, limit: Optional[int] = 1, fields: Optional[List[str]] = None) -> List[Dict]:
        """
        Query records using ServiceNow encoded query string.

        Args:
            table: ServiceNow table name
            encoded_query: ServiceNow encoded query string
            limit: Maximum number of records to return
            fields: Optional list of field names to return. If None, returns all fields.

        Returns:
            List of dictionaries containing record data
        """
        try:
            resource = self.client.resource(api_path=f'/table/{table}')

            query_params = {
                'sysparm_query': encoded_query,
            }

            if fields:
                response = resource.get(query=query_params, fields=fields, display_value=True, exclude_reference_link=True, limit=limit)
            else:
                response = resource.get(query=query_params, display_value=True, exclude_reference_link=True, limit=limit)
            data = response.all()
            logger.info(data)
            return data

        except Exception as e:
            logger.error(f"Error querying with encoded query: {str(e)}")
            return []

    @keyword
    def create_record(self, table: str, data: Dict[str, Any]) -> Optional[Dict]:
        """
        Create a new record in a table.

        Args:
            table: ServiceNow table name
            data: Dictionary of field-value pairs for the new record

        Returns:
            Dictionary containing created record data or None if failed
        """
        try:
            resource = self.client.resource(api_path=f'/table/{table}')
            response = resource.create(payload=data)
            logger.info(f"Successfully created record in {table}")
            data = dict(response)
            logger.info(data)
            return data

        except Exception as e:
            logger.error(f"Error creating record: {str(e)}")
            return None

    @keyword
    def update_record(self, table: str, sys_id: str, data: Dict[str, Any]) -> Optional[Dict]:
        """
        Update an existing record.

        Args:
            table: ServiceNow table name
            sys_id: System ID of the record to update
            data: Dictionary of field-value pairs to update

        Returns:
            Dictionary containing updated record data or None if failed
        """
        try:
            resource = self.client.resource(api_path=f'/table/{table}/{sys_id}')
            response = resource.update(payload=data)

            logger.info(f"Successfully updated record {sys_id} in {table}")
            data = dict(response)
            logger.info(data)
            return data

        except Exception as e:
            logger.error(f"Error updating record: {str(e)}")
            return None

    @keyword
    def delete_record(self, table: str, sys_id: str) -> bool:
        """
        Delete a record by sys_id.

        Args:
            table: ServiceNow table name
            sys_id: System ID of the record to delete

        Returns:
            True if successful, False otherwise
        """
        try:
            resource = self.client.resource(api_path=f'/table/{table}/{sys_id}')
            resource.delete()

            logger.info(f"Successfully deleted record {sys_id} from {table}")
            return True

        except Exception as e:
            logger.error(f"Error deleting record: {str(e)}")
            return False


    @keyword
    def get_table_schema(self, table: str) -> Optional[Dict]:
        """
        Get table schema information.

        Args:
            table: ServiceNow table name

        Returns:
            Dictionary containing table schema or None if failed
        """
        try:
            resource = self.client.resource(api_path=f'/table/sys_dictionary')
            response = resource.get(query={'name': table}, display_value=True, exclude_reference_link=True)
            data = response.all()
            logger.info(data)
            return data

        except Exception as e:
            logger.error(f"Error getting table schema: {str(e)}")
            return None

    @keyword
    def count_records(self, table: str, query: Optional[Dict[str, Any]] = None) -> int:
        """
        Count records in a table matching the query.

        Args:
            table: ServiceNow table name
            query: Optional dictionary of field-value pairs for query

        Returns:
            Number of records matching the query
        """
        try:
            resource = self.client.resource(api_path=f'/table/{table}')

            if query:
                response = resource.get(query=query, fields=['sys_id'])
            else:
                response = resource.get(fields=['sys_id'])

            data = len(response.all())
            logger.info(f"Counted {data} records in {table}")
            return data

        except Exception as e:
            logger.error(f"Error counting records: {str(e)}")
            return 0

    @keyword
    def test_connection(self) -> bool:
        """
        Test the connection to ServiceNow.

        Returns:
            True if connection is successful, False otherwise
        """
        try:
            resource = self.client.resource(api_path='/table/sys_user')
            response = resource.get(query={'sys_id': 'dummy'}, display_value=True, exclude_reference_link=True)
            # For pysnow, if we can call .all() without exception, connection is good
            response.all()
            logger.info("Connection test successful")
            return True
        except Exception as e:
            logger.error(f"Connection test failed: {str(e)}")
            return False