# -*- coding: utf-8 -*-
"""
:Module: salespyforce.core
:Synopsis: This module performs the core Salesforce-related operations
:Usage: ``from salespyforce import Salesforce``
:Example: ``sfdc = Salesforce(helper=helper_file_path)``
:Created By: Jeff Shurtliff
:Last Modified: Jeff Shurtliff
:Modified Date: 03 Feb 2026
"""
from __future__ import annotations
import re
from typing import Optional
import requests
from . import api, errors
from . import chatter as chatter_module
from . import knowledge as knowledge_module
from .utils import core_utils, log_utils
from .utils.helper import get_helper_settings
# Define constants
FALLBACK_SFDC_API_VERSION = '65.0' # Used if querying the org for the version fails
VALID_ACCESS_CONTROL_FIELDS = {'HasReadAccess', 'HasEditAccess', 'HasDeleteAccess'}
# Initialize logging
logger = log_utils.initialize_logging(__name__)
[docs]
class Salesforce(object):
"""This is the class for the core object leveraged in this module."""
# Define the function that initializes the object instance (i.e. instantiates the object)
[docs]
def __init__(self, connection_info=None, version=None, base_url=None, org_id=None, username=None,
password=None, endpoint_url=None, client_id=None, client_secret=None, security_token=None, helper=None):
"""This method instantiates the core Salesforce object.
.. versionchanged:: 1.4.0
The authorized Salesforce org is now queried to determine the latest API version to leverage unless
explicitly defined with the ``version`` parameter when instantiating the object.
:param connection_info: The information for connecting to the Salesforce instance
:type connection_info: dict, None
:param version: The Salesforce API version to utilize (uses latest version from org if not explicitly defined)
:type version: str, None
:param base_url: The base URL of the Salesforce instance
:type base_url: str, None
:param org_id: The Org ID of the Salesforce instance
:type org_id: str, None
:param username: The username of the API user
:type username: str, None
:param password: The password of the API user
:type password: str, None
:param endpoint_url: The endpoint URL for the Salesforce instance
:type endpoint_url: str, None
:param client_id: The Client ID for the Salesforce instance
:type client_id: str, None
:param client_secret: The Client Secret for the Salesforce instance
:type client_secret: str, None
:param security_token: The Security Token for the Salesforce instance
:type security_token: str, None
:param helper: The file path of a helper file
:type helper: str, None
:returns: The instantiated object
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`
"""
# Define the default settings
self._helper_settings = {}
# Check for provided connection info
if connection_info is None:
# Check for a supplied helper file
if helper:
# Parse the helper file contents
self.helper_path = helper
if any((isinstance(helper, tuple), isinstance(helper, list), isinstance(helper, set))):
helper_file_path, helper_file_type = helper
elif isinstance(helper, str):
helper_file_path, helper_file_type = (helper, 'yaml')
elif isinstance(helper, dict):
helper_file_path, helper_file_type = helper.values()
else:
error_msg = "The 'helper' argument can only be supplied as tuple, string, list, set or dict."
logger.error(error_msg)
raise TypeError(error_msg)
self._helper_settings = get_helper_settings(helper_file_path, helper_file_type)
connection_info = self._parse_helper_connection_info()
elif not any((base_url, org_id, username, password, endpoint_url, client_id, client_secret, security_token)):
# Prompt for the connection info if not defined
connection_info = define_connection_info()
else:
# Compile the connection info from the provided parameters
connection_info = compile_connection_info(base_url, org_id, username, password, endpoint_url,
client_id, client_secret, security_token)
# Get the connection information used to connect to the instance
self.connection_info = connection_info if connection_info is not None else self._get_empty_connection_info()
# Define the base URL and Org ID
self.base_url = self.connection_info.get('base_url', '')
self.org_id = self.connection_info.get('org_id', '')
# Define the connection response data variables
auth_response = self.connect()
self.access_token = auth_response.get('access_token')
self.instance_url = auth_response.get('instance_url')
self.signature = auth_response.get('signature')
# Define the version with explicitly provided version or by querying the Salesforce org
self.version = f'v{version}' if version else f'v{self.get_latest_api_version()}'
# Retrieve info about current user
self.current_user_info = self.retrieve_current_user_info(on_init=True, raise_exc_on_error=False)
# Import inner object classes so their methods can be called from the primary object
self.chatter = self._import_chatter_class()
self.knowledge = self._import_knowledge_class()
def _import_chatter_class(self):
"""This method allows the :py:class:`salespyforce.core.Salesforce.Chatter` class to be utilized in the core object."""
return Salesforce.Chatter(self)
def _import_knowledge_class(self):
"""This method allows the :py:class:`salespyforce.core.Salesforce.Knowledge` class to be utilized in the core object."""
return Salesforce.Knowledge(self)
@staticmethod
def _get_empty_connection_info():
"""This method returns an empty connection_info dictionary with all blank values."""
_connection_info = {}
_fields = ['username', 'password', 'base_url', 'endpoint_url',
'client_key', 'client_secret', 'org_id', 'security_token']
for _field in _fields:
_connection_info[_field] = ''
return _connection_info
def _parse_helper_connection_info(self):
"""This method parses the helper content to populate the connection info."""
_connection_info = {}
_fields = ['username', 'password', 'base_url', 'endpoint_url',
'client_key', 'client_secret', 'org_id', 'security_token']
for _field in _fields:
if _field in self._helper_settings['connection']:
_connection_info[_field] = self._helper_settings['connection'][_field]
return _connection_info
def _get_headers(self, _header_type='default'):
"""This method returns the appropriate HTTP headers to use for different types of API calls."""
return api._get_headers(_access_token=self.access_token, _header_type=_header_type)
def _get_cached_user_info(self, _field: str, _retrieve_if_missing: bool = False):
"""This method attempts to retrieve a value for a given field in the cached ``userinfo`` data and
optionally queries the API as needed to retrieve the data when not found.
.. versionadded:: 1.4.0
:param _field: The name of the field for which the value is needed
:type _field: str
:param _retrieve_if_missing: Will query the Salesforce REST API for the data when missing if True
(``False`` by default)
:type _retrieve_if_missing: bool
:returns: The field value when found (or retrieved), or a None value if the field value could not be obtained
:raises: :py:exc:`salespyforce.errors.exceptions.APIRequestError`
"""
_field_value = None
_not_present_msg = f"The '{_field}' field is not present in the current user info data"
if self.current_user_info and _field in self.current_user_info:
_field_value = self.current_user_info[_field]
else:
logger.warning(_not_present_msg)
if _retrieve_if_missing:
self.current_user_info = self.retrieve_current_user_info(raise_exc_on_error=True)
if self.current_user_info and _field in self.current_user_info:
_field_value = self.current_user_info[_field]
else:
logger.error(f'{_not_present_msg} even after refreshing cached current user info')
return _field_value
[docs]
def connect(self):
"""This method connects to the Salesforce instance to obtain the access token.
(`Reference <https://jereze.com/code/authentification-salesforce-rest-api-python/>`_)
:returns: The API call response with the authorization information
:raises: :py:exc:`RuntimeError`
"""
params = {
'grant_type': 'password',
'client_id': self.connection_info.get('client_key'),
'client_secret': self.connection_info.get('client_secret'),
'username': self.connection_info.get('username'),
'password': f'{self.connection_info.get("password")}{self.connection_info.get("security_token")}'
}
response = requests.post(self.connection_info.get('endpoint_url'), params=params)
if response.status_code != 200:
raise RuntimeError(f'Failed to connect to the Salesforce instance.\n{response.text}')
return response.json()
[docs]
def retrieve_current_user_info(self, all_data=False, raise_exc_on_error=False, on_init=False) -> dict:
"""This method retrieves the ``userinfo`` data for the current/running user.
.. versionadded:: 1.4.0
:param all_data: Returns all ``userinfo`` data from the API when True instead of only the relevant fields/values
(``False`` by default)
:type all_data: bool
:param raise_exc_on_error: Raises an exception if the API retrieval attempt fails when True (``False`` by default)
:type raise_exc_on_error: bool
:param on_init: Indicates if the method is being called during the core object instantiation (``False`` by default)
:type on_init: bool
:returns: The user info data within a dictionary
:raises: :py:exc:`RuntimeError`,
:py:exc:`salespyforce.errors.exceptions.APIRequestError`
"""
user_info = {'user_id': '', 'nickname': '', 'name': '', 'email': '', 'user_type': '',
'language': '', 'locale': '', 'utcOffset': '', 'is_salesforce_integration_user': None}
bool_fields = ['is_salesforce_integration_user']
endpoint = '/services/oauth2/userinfo'
base_error_msg = 'Failed to retrieve current user info'
msg_init_segment = 'on core object instantiation'
if on_init:
base_error_msg = f'{base_error_msg} {msg_init_segment}'
try:
response = self.get(endpoint)
if isinstance(response, dict) and all_data:
user_info = response
elif isinstance(response, dict):
for field in user_info.keys():
if field in response:
default_val = None if field in bool_fields else ''
user_info[field] = response.get(field, default_val)
else:
logger.error(f'{base_error_msg} with a usable format')
except Exception as exc:
exc_type = errors.handlers.get_exception_type(exc)
exc_msg = f'{base_error_msg} due to {exc_type} exception: {exc}'
logger.error(exc_msg)
if raise_exc_on_error:
raise errors.exceptions.APIRequestError(f'{exc_type}: {exc}')
# Return the populated user info
return user_info
[docs]
def get(self, endpoint, params=None, headers=None, timeout=None, show_full_error=True, return_json=True):
"""This method performs a GET request against the Salesforce instance.
(`Reference <https://jereze.com/code/authentification-salesforce-rest-api-python/>`_)
.. versionchanged:: 1.4.0
A global constant is now leveraged for the API timeout value instead of hardcoding the value.
(Timeout is still **30** seconds in this version)
:param endpoint: The API endpoint to query
:type endpoint: str
:param params: The query parameters (where applicable)
:type params: dict, None
:param headers: Specific API headers to use when performing the API call
:type headers: dict, None
:param timeout: The timeout period in seconds (defaults to ``30``)
:type timeout: int, None
:param show_full_error: Determines if the full error message should be displayed (defaults to ``True``)
:type show_full_error: bool
:param return_json: Determines if the response should be returned in JSON format (defaults to ``True``)
:returns: The API response in JSON format or as a ``requests`` object
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`,
:py:exc:`salespyforce.errors.exceptions.InvalidURLError`
"""
return api.get(self, endpoint=endpoint, params=params, headers=headers, timeout=timeout,
show_full_error=show_full_error, return_json=return_json)
[docs]
def api_call_with_payload(self, method, endpoint, payload, params=None, headers=None, timeout=None,
show_full_error=True, return_json=True):
"""This method performs a POST call against the Salesforce instance.
(`Reference <https://jereze.com/code/authentification-salesforce-rest-api-python/>`_)
.. versionchanged:: 1.4.0
A global constant is now leveraged for the API timeout value instead of hardcoding the value.
(Timeout is still **30** seconds in this version)
:param method: The API method (``post``, ``put``, or ``patch``)
:type method: str
:param endpoint: The API endpoint to query
:type endpoint: str
:param payload: The payload to leverage in the API call
:type payload: dict
:param params: The query parameters (where applicable)
:type params: dict, None
:param headers: Specific API headers to use when performing the API call
:type headers: dict, None
:param timeout: The timeout period in seconds (defaults to ``30``)
:type timeout: int, None
:param show_full_error: Determines if the full error message should be displayed (defaults to ``True``)
:type show_full_error: bool
:param return_json: Determines if the response should be returned in JSON format (defaults to ``True``)
:returns: The API response in JSON format or as a ``requests`` object
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`,
:py:exc:`ValueError`,
:py:exc:`salespyforce.errors.exceptions.InvalidURLError`
"""
return api.api_call_with_payload(self, method=method, endpoint=endpoint, payload=payload, params=params,
headers=headers, timeout=timeout, show_full_error=show_full_error,
return_json=return_json)
[docs]
def post(self, endpoint, payload, params=None, headers=None, timeout=None, show_full_error=True, return_json=True):
"""This method performs a POST call against the Salesforce instance.
(`Reference <https://jereze.com/code/authentification-salesforce-rest-api-python/>`_)
.. versionchanged:: 1.4.0
A global constant is now leveraged for the API timeout value instead of hardcoding the value.
(Timeout is still **30** seconds in this version)
:param endpoint: The API endpoint to query
:type endpoint: str
:param payload: The payload to leverage in the API call
:type payload: dict
:param params: The query parameters (where applicable)
:type params: dict, None
:param headers: Specific API headers to use when performing the API call
:type headers: dict, None
:param timeout: The timeout period in seconds (defaults to ``30``)
:type timeout: int, None
:param show_full_error: Determines if the full error message should be displayed (defaults to ``True``)
:type show_full_error: bool
:param return_json: Determines if the response should be returned in JSON format (defaults to ``True``)
:returns: The API response in JSON format or as a ``requests`` object
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`,
:py:exc:`ValueError`,
:py:exc:`salespyforce.errors.exceptions.InvalidURLError`
"""
return api.api_call_with_payload(self, 'post', endpoint=endpoint, payload=payload, params=params,
headers=headers, timeout=timeout, show_full_error=show_full_error,
return_json=return_json)
[docs]
def patch(self, endpoint, payload, params=None, headers=None, timeout=None, show_full_error=True, return_json=False):
"""This method performs a PATCH call against the Salesforce instance.
(`Reference <https://jereze.com/code/authentification-salesforce-rest-api-python/>`_)
.. versionchanged:: 1.4.0
A global constant is now leveraged for the API timeout value instead of hardcoding the value.
(Timeout is still **30** seconds in this version)
:param endpoint: The API endpoint to query
:type endpoint: str
:param payload: The payload to leverage in the API call
:type payload: dict
:param params: The query parameters (where applicable)
:type params: dict, None
:param headers: Specific API headers to use when performing the API call
:type headers: dict, None
:param timeout: The timeout period in seconds (defaults to ``30``)
:type timeout: int, None
:param show_full_error: Determines if the full error message should be displayed (defaults to ``True``)
:type show_full_error: bool
:param return_json: Determines if the response should be returned in JSON format (defaults to ``True``)
:returns: The API response in JSON format or as a ``requests`` object
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`,
:py:exc:`ValueError`,
:py:exc:`salespyforce.errors.exceptions.InvalidURLError`
"""
return api.api_call_with_payload(self, 'patch', endpoint=endpoint, payload=payload, params=params,
headers=headers, timeout=timeout, show_full_error=show_full_error,
return_json=return_json)
[docs]
def put(self, endpoint, payload, params=None, headers=None, timeout=None, show_full_error=True, return_json=True):
"""This method performs a PUT call against the Salesforce instance.
(`Reference <https://jereze.com/code/authentification-salesforce-rest-api-python/>`_)
.. versionchanged:: 1.4.0
A global constant is now leveraged for the API timeout value instead of hardcoding the value.
(Timeout is still **30** seconds in this version)
:param endpoint: The API endpoint to query
:type endpoint: str
:param payload: The payload to leverage in the API call
:type payload: dict
:param params: The query parameters (where applicable)
:type params: dict, None
:param headers: Specific API headers to use when performing the API call
:type headers: dict, None
:param timeout: The timeout period in seconds (defaults to ``30``)
:type timeout: int, None
:param show_full_error: Determines if the full error message should be displayed (defaults to ``True``)
:type show_full_error: bool
:param return_json: Determines if the response should be returned in JSON format (defaults to ``True``)
:returns: The API response in JSON format or as a ``requests`` object
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`,
:py:exc:`ValueError`,
:py:exc:`salespyforce.errors.exceptions.InvalidURLError`
"""
return api.api_call_with_payload(self, 'put', endpoint=endpoint, payload=payload, params=params,
headers=headers, timeout=timeout, show_full_error=show_full_error,
return_json=return_json)
[docs]
def delete(self, endpoint, params=None, headers=None, timeout=None, show_full_error=True, return_json=True):
"""This method performs a DELETE request against the Salesforce instance.
(`Reference <https://jereze.com/code/authentification-salesforce-rest-api-python/>`_)
.. versionadded:: 1.4.0
:param endpoint: The API endpoint to query
:type endpoint: str
:param params: The query parameters (where applicable)
:type params: dict, None
:param headers: Specific API headers to use when performing the API call
:type headers: dict, None
:param timeout: The timeout period in seconds (defaults to ``30``)
:type timeout: int, None
:param show_full_error: Determines if the full error message should be displayed (defaults to ``True``)
:type show_full_error: bool
:param return_json: Determines if the response should be returned in JSON format (defaults to ``True``)
:returns: The API response in JSON format or as a ``requests`` object
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`,
:py:exc:`salespyforce.errors.exceptions.InvalidURLError`
"""
return api.delete(self, endpoint=endpoint, params=params, headers=headers, timeout=timeout,
show_full_error=show_full_error, return_json=return_json)
[docs]
def get_api_versions(self) -> list:
"""This method returns the API versions for the Salesforce releases.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_versions.htm>`_)
:returns: A list containing the API metadata from the ``/services/data`` endpoint.
:raises: :py:exc:`RuntimeError`
"""
return self.get('/services/data')
[docs]
def get_latest_api_version(self) -> str:
"""This method returns the latest Salesforce API version by querying the authorized org.
.. versionadded:: 1.4.0
:returns: The latest Salesforce API version for the authorized org as a string (e.g. ``65.0``)
"""
versions = self.get_api_versions()
try:
latest_version = versions[-1]['version']
except Exception as exc:
exc_type = type(exc).__name__
logger.warning(
f"Failed to retrieve API version due to a(n) {exc_type} exception; defaulting to "
f"the fallback version {FALLBACK_SFDC_API_VERSION}"
)
latest_version = FALLBACK_SFDC_API_VERSION
return latest_version
[docs]
def get_org_limits(self):
"""This method returns a list of all org limits.
.. versionadded:: 1.1.0
:returns: The Salesforce org governor limits data
:raises: :py:exc:`RuntimeError`
"""
return self.get(f'/services/data/{self.version}/limits')
[docs]
def get_all_sobjects(self):
"""This method returns a list of all Salesforce objects. (i.e. sObjects)
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_describeGlobal.htm>`_)
:returns: The list of all Salesforce objects
:raises: :py:exc:`RuntimeError`
"""
return self.get(f'/services/data/{self.version}/sobjects')
[docs]
def get_sobject(self, object_name, describe=False):
"""This method returns basic information or the full (describe) information for a specific sObject.
(`Reference 1 <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_sobject_basic_info_get.htm>`_,
`Reference 2 <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_sobject_describe.htm>`_)
:param object_name: The name of the Salesforce object
:type object_name: str
:param describe: Determines if the full (i.e. ``describe``) data should be returned (defaults to ``False``)
:type describe: bool
:returns: The Salesforce object data
:raises: :py:exc:`RuntimeError`
"""
uri = f'/services/data/{self.version}/sobjects/{object_name}'
uri = f'{uri}/describe' if describe else uri
return self.get(uri)
[docs]
def describe_object(self, object_name):
"""This method returns the full (describe) information for a specific sObject.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_sobject_describe.htm>`_)
:param object_name: The name of the Salesforce object
:type object_name: str
:returns: The Salesforce object data
:raises: :py:exc:`RuntimeError`
"""
return self.get_sobject(object_name, describe=True)
[docs]
def get_rest_resources(self):
"""This method returns a list of all available REST resources.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_discoveryresource.htm>`_)
:returns: The list of all available REST resources for the Salesforce org
:raises: :py:exc:`RuntimeError`
"""
return self.get(f'/services/data/{self.version}')
[docs]
@staticmethod
def get_18_char_id(record_id: str) -> str:
"""This method converts a 15-character Salesforce record ID to its 18-character case-insensitive form.
.. versionadded:: 1.4.0
:param record_id: The Salesforce record ID to convert (or return unchanged if already 18 characters)
:type record_id: str
:returns: The 18-character Salesforce record ID
:raises: :py:exc:`ValueError`
"""
return core_utils.get_18_char_id(record_id=record_id)
[docs]
def soql_query(self, query, replace_quotes=True, next_records_url=False):
"""This method performs a SOQL query and returns the results in JSON format.
(`Reference 1 <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_query.htm>`_,
`Reference 2 <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/knowledge_development_soql_sosl_intro.htm>`_)
:param query: The SOQL query to perform
:type query: str
:param replace_quotes: Determines if double-quotes should be replaced with single-quotes (``True`` by default)
:type replace_quotes: bool
:param next_records_url: Indicates that the ``query`` parameter is a ``nextRecordsUrl`` value.
:type next_records_url: bool
:returns: The result of the SOQL query
:raises: :py:exc:`RuntimeError`
"""
if next_records_url:
query = re.sub(r'^.*/', '', query) if '/' in query else query
else:
if replace_quotes:
query = query.replace('"', "'")
query = core_utils.url_encode(query)
query = f'?q={query}'
return self.get(f'/services/data/{self.version}/query/{query}')
[docs]
def search_string(self, string_to_search):
"""This method performs a SOSL query to search for a given string.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_search.htm>`_)
.. versionadded:: 1.1.0
:param string_to_search: The string for which to search
:type string_to_search: str
:returns: The SOSL response data in JSON format
:raises: :py:exc:`RuntimeError`
"""
query = 'FIND {' + string_to_search + '}'
query = core_utils.url_encode(query)
return self.get(f'/services/data/{self.version}/search/?q={query}')
[docs]
def check_user_record_access(self, record_id: str, user_id=None) -> dict:
"""This method checks the Read, Edit, and Delete access for a given record and user.
.. versionadded:: 1.4.0
:param record_id: The ``Id`` value of the record against which to check the user access
:type record_id: str
:param user_id: The ``Id`` of the user to evaluate (or the current user's ID if not explicitly defined)
:type user_id: str, None
:returns: Dictionary with Boolean values for ``HasReadAccess``, ``HasEditAccess``, and ``HasDeleteAccess``
:raises: :py:exc:`RuntimeError`,
:py:exc:`salespyforce.errors.exceptions.APIRequestError`
"""
record_access = {'HasReadAccess': None, 'HasEditAccess': None, 'HasDeleteAccess': None}
# Use the current/running user's ID if an ID wasn't explicitly provided
if not user_id:
user_id = self._get_cached_user_info(_field='user_id', _retrieve_if_missing=True)
if user_id:
logger.debug(f'Using the User Id {user_id} for the running user as an Id was not specified')
# Raise an exception if the User ID is still undefined
if not user_id:
error_msg = f'The user access for record Id {record_id} cannot be checked as the User Id is undefined'
logger.error(error_msg)
raise errors.exceptions.MissingRequiredDataError(error_msg)
# Perform SOQL query for the access data
query = f"""
SELECT RecordId, HasReadAccess, HasEditAccess, HasDeleteAccess
FROM UserRecordAccess
WHERE UserId = '{user_id}' AND RecordId = '{record_id}'
"""
response = self.soql_query(query=query)
# Parse the response to extract the relevant field values
if 'records' in response and response['records']:
response = response['records'][0]
for field in record_access.keys():
record_access[field] = response.get(field, None)
# Return the record access data
return record_access
@staticmethod
def _eval_user_record_access(_field: str, _record_id: str,
_record_access_data: dict, _raise_exc_on_failure: bool = True) -> bool:
"""This private method checks for an access level given the field and the record access data.
.. versionadded:: 1.4.0
:param _field: The access level field to evaluate (``HasReadAccess``, ``HasEditAccess``, ``HasDeleteAccess``)
:type _field: str
:param _record_id: The ID value for the record whose access is being checked
:type _record_id: str
:param _record_access_data: The user record access data that has already been retrieved
:type _record_access_data: dict
:param _raise_exc_on_failure: Raises an exception rather than returning a ``None`` value (``True`` by default)
:type _raise_exc_on_failure: bool
:returns: Boolean value indicating the access level for the given field
:raises: :py:exc:`salespyforce.errors.exceptions.InvalidFieldError`
"""
# Raise an exception if a valid access control field is not provided
if _field not in VALID_ACCESS_CONTROL_FIELDS:
_error_msg = f"The field '{_field}' is not a valid record access level field"
raise errors.exceptions.InvalidFieldError(_error_msg)
# Identify the access level value if possible (API retrievals should be handled in a parent method)
_has_access = _record_access_data.get(_field) if _field in _record_access_data else None
if _has_access is None:
_error_msg = f"The value for the '{_field}' is undefined for the given Record Id '{_record_id}'"
logger.error(_error_msg)
if _raise_exc_on_failure:
raise errors.exceptions.MissingRequiredDataError(_error_msg)
# Return the identified access level value
return _has_access
[docs]
def can_access_record(self, access_type, record_id, user_id=None, record_access_data=None, raise_exc_on_failure=True):
"""This method evaluates if a user can access a specific record given the access type.
.. versionadded:: 1.4.0
:param access_type: The type of access to evaluate (``read``, ``edit``, or ``delete``)
:type access_type: str
:param record_id: The ID of the record
:type record_id: str
:param user_id: The ID of the user to evaluate (defaults to the current/running user if not defined)
:type user_id: str, None
:param record_access_data: The user record access data that has already been retrieved (optional)
:type record_access_data: dict, None
:param raise_exc_on_failure: Raises an exception rather than returning a ``None`` value (``True`` by default)
:type raise_exc_on_failure: bool
:returns: Boolean value indicating the access level for the given field
:raises: :py:exc:`salespyforce.errors.exceptions.InvalidFieldError`
"""
# Define the initial value for the result
can_access = None
# Identify the correct field to query based on access type
access_type_field_mapping = {
'read': 'HasReadAccess',
'edit': 'HasEditAccess',
'delete': 'HasDeleteAccess',
}
if access_type.lower() not in access_type_field_mapping:
error_msg = f"The access_type '{access_type}' is invalid (must use 'read', 'edit', or 'delete')"
logger.error(error_msg)
if raise_exc_on_failure:
raise errors.exceptions.InvalidParameterError(error_msg)
else:
# Retrieve the field name to check
read_access_field = access_type_field_mapping.get(access_type.lower())
# Check to see if record access data was provided and validate that it is a dictionary
if record_access_data and not isinstance(record_access_data, dict):
error_msg = f"The record_access_data provided is Type {type(read_access_field)} but must be a dict"
logger.error(error_msg)
if raise_exc_on_failure:
raise errors.exceptions.DataMismatchError(error_msg)
# Perform the API all to check the record access for the user if data not provided
if not record_access_data:
record_access_data = self.check_user_record_access(record_id=record_id, user_id=user_id)
# Return the access level value
can_access = self._eval_user_record_access(_field=read_access_field, _record_id=record_id,
_record_access_data=record_access_data,
_raise_exc_on_failure=raise_exc_on_failure)
# Emit a warning if the value is None rather than a boolean
if can_access is None:
warn_msg = 'The record access check could not be completed and the function will return a None value'
errors.handlers.display_warning(warn_msg)
# Return the result
return can_access
[docs]
def can_read_record(self, record_id, user_id=None, record_access_data=None, raise_exc_on_failure=True):
"""This method evaluates if a user has access to read a specific record.
.. versionadded:: 1.4.0
:param record_id: The ID of the record
:type record_id: str
:param user_id: The ID of the user to evaluate (defaults to the current/running user if not defined)
:type user_id: str, None
:param record_access_data: The user record access data that has already been retrieved (optional)
:type record_access_data: dict, None
:param raise_exc_on_failure: Raises an exception rather than returning a ``None`` value (``True`` by default)
:type raise_exc_on_failure: bool
:returns: Boolean value indicating the access level for the given field
:raises: :py:exc:`salespyforce.errors.exceptions.InvalidFieldError`
"""
return self.can_access_record(access_type='read', record_id=record_id, user_id=user_id,
record_access_data=record_access_data,
raise_exc_on_failure=raise_exc_on_failure)
[docs]
def can_edit_record(self, record_id, user_id=None, record_access_data=None, raise_exc_on_failure=True):
"""This method evaluates if a user has access to edit a specific record.
.. versionadded:: 1.4.0
:param record_id: The ID of the record
:type record_id: str
:param user_id: The ID of the user to evaluate (defaults to the current/running user if not defined)
:type user_id: str, None
:param record_access_data: The user record access data that has already been retrieved (optional)
:type record_access_data: dict, None
:param raise_exc_on_failure: Raises an exception rather than returning a ``None`` value (``True`` by default)
:type raise_exc_on_failure: bool
:returns: Boolean value indicating the access level for the given field
:raises: :py:exc:`salespyforce.errors.exceptions.InvalidFieldError`
"""
return self.can_access_record(access_type='edit', record_id=record_id, user_id=user_id,
record_access_data=record_access_data,
raise_exc_on_failure=raise_exc_on_failure)
[docs]
def can_delete_record(self, record_id, user_id=None, record_access_data=None, raise_exc_on_failure=True):
"""This method evaluates if a user has access to delete a specific record.
.. versionadded:: 1.4.0
:param record_id: The ID of the record
:type record_id: str
:param user_id: The ID of the user to evaluate (defaults to the current/running user if not defined)
:type user_id: str, None
:param record_access_data: The user record access data that has already been retrieved (optional)
:type record_access_data: dict, None
:param raise_exc_on_failure: Raises an exception rather than returning a ``None`` value (``True`` by default)
:type raise_exc_on_failure: bool
:returns: Boolean value indicating the access level for the given field
:raises: :py:exc:`salespyforce.errors.exceptions.InvalidFieldError`
"""
return self.can_access_record(access_type='edit', record_id=record_id, user_id=user_id,
record_access_data=record_access_data,
raise_exc_on_failure=raise_exc_on_failure)
[docs]
def create_sobject_record(self, sobject, payload):
"""This method creates a new record for a specific sObject.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_sobject_create.htm>`_)
:param sobject: The sObject under which to create the new record
:type sobject: str
:param payload: The JSON payload with the record details
:type payload: dict
:returns: The API response from the POST request
:raises: :py:exc:`RuntimeError`,
:py:exc:`TypeError`
"""
# Ensure the payload is in the appropriate format
if not isinstance(payload, dict):
raise TypeError('The sObject payload must be provided as a dictionary.')
# Perform the API call and return the response
response = self.post(f'/services/data/{self.version}/sobjects/{sobject}', payload=payload)
return response
[docs]
def update_sobject_record(self, sobject, record_id, payload):
"""This method updates an existing sObject record.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_update_fields.htm>`_)
:param sobject: The sObject under which to update the record
:type sobject: str
:param record_id: The ID of the record to be updated
:type record_id: str
:param payload: The JSON payload with the record details to be updated
:type payload: dict
:returns: The API response from the PATCH request
:raises: :py:exc:`RuntimeError`,
:py:exc:`TypeError`
"""
# Ensure the payload is in the appropriate format
if not isinstance(payload, dict):
raise TypeError('The sObject payload must be provided as a dictionary.')
# Perform the API call and return the response
response = self.patch(f'/services/data/{self.version}/sobjects/{sobject}/{record_id}', payload=payload)
return response
[docs]
def download_image(self, image_url, record_id, field_name, file_path=None, sobject=None):
"""This method downloads an image using the sObject Rich Text Image Retrieve functionality.
(`Reference 1 <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_sobject_rich_text_image_retrieve.htm>`_,
`Reference 2 <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_sobject_rich_text_image_retrieve.htm>`_)
:param image_url: The URL for the image to be downloaded
:type image_url: str
:param record_id: The Record ID where the image is found
:type record_id: str
:param field_name: The field name within the record where the image is found
:type field_name: str
:param file_path: The path to the directory where the image should be saved (current directory if not defined)
:type file_path: str, None
:param sobject: The sObject for the record where the image is found (``Knowledge__kav`` by default)
:type sobject: str
:returns: The full path to the downloaded image
:raises: :py:exc:`RuntimeError`
"""
# Ensure a valid sObject is defined (SFDC Knowledge unless otherwise specified)
sobject = 'Knowledge__kav' if not sobject else sobject
# Retrieve the reference ID for the image
ref_id = core_utils.get_image_ref_id(image_url)
# Define the URI and perform the API call
image_path = None
try:
uri = f'/services/data/{self.version}/sobjects/{sobject}/{record_id}/richTextImageFields/{field_name}/{ref_id}'
response = self.get(uri, return_json=False)
# Save the image as an image file
try:
image_path = core_utils.download_image(file_name=f'{ref_id}.jpeg', file_path=file_path,
response=response)
except RuntimeError:
errors.handlers.eprint(f'Failed to download the image with refid {ref_id}.')
except RuntimeError as exc:
errors.handlers.eprint(exc)
return image_path
[docs]
class Chatter(object):
"""This class includes methods associated with Salesforce Chatter."""
[docs]
def __init__(self, sfdc_object):
"""This method initializes the :py:class:`salespyforce.core.Salesforce.Chatter` inner class object.
:param sfdc_object: The core :py:class:`salespyforce.Salesforce` object
:type sfdc_object: class[salespyforce.Salesforce]
"""
self.sfdc_object = sfdc_object
[docs]
def get_my_news_feed(self, site_id=None):
"""This method retrieves the news feed for the user calling the function.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.chatterapi.meta/chatterapi/quickreference_get_news_feed.htm>`_)
:param site_id: The ID of an Experience Cloud site against which to query (optional)
:type site_id: str, None
:returns: The news feed data
:raises: :py:exc:`RuntimeError`
"""
return chatter_module.get_my_news_feed(self.sfdc_object, site_id=site_id)
[docs]
def get_user_news_feed(self, user_id, site_id=None):
"""This method retrieves another user's news feed.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.chatterapi.meta/chatterapi/quickreference_get_user_profile_feed.htm>`_)
:param user_id: The ID of the user whose feed you wish to return
:type user_id: str
:param site_id: The ID of an Experience Cloud site against which to query (optional)
:type site_id: str, None
:returns: The news feed data
:raises: :py:exc:`RuntimeError`
"""
return chatter_module.get_user_news_feed(self.sfdc_object, user_id=user_id, site_id=site_id)
[docs]
def get_group_feed(self, group_id, site_id=None):
"""This method retrieves a group's news feed.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.chatterapi.meta/chatterapi/quickreference_get_group_feed.htm>`_)
:param group_id: The ID of the group whose feed you wish to return
:type group_id: str
:param site_id: The ID of an Experience Cloud site against which to query (optional)
:type site_id: str, None
:returns: The news feed data
:raises: :py:exc:`RuntimeError`
"""
return chatter_module.get_group_feed(self.sfdc_object, group_id=group_id, site_id=site_id)
[docs]
def post_feed_item(self, subject_id, message_text=None, message_segments=None, site_id=None, created_by_id=None):
"""This method publishes a new Chatter feed item.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.chatterapi.meta/chatterapi/quickreference_post_feed_item.htm>`_)
:param subject_id: The Subject ID against which to publish the feed item (e.g. ``0F9B000000000W2``)
:type subject_id: str
:param message_text: Plaintext to be used as the message body
:type message_segments: str, None
:param message_segments: Collection of message segments to use instead of a plaintext message
:type message_segments: list, None
:param site_id: The ID of an Experience Cloud site against which to query (optional)
:type site_id: str, None
:param created_by_id: The ID of the user to impersonate (**Experimental**)
:type created_by_id: str, None
:returns: The response of the POST request
:raises: :py:exc:`RuntimeError`
"""
return chatter_module.post_feed_item(self.sfdc_object, subject_id=subject_id, message_text=message_text,
message_segments=message_segments, site_id=site_id,
created_by_id=created_by_id)
[docs]
class Knowledge(object):
"""This class includes methods associated with Salesforce Knowledge."""
[docs]
def __init__(self, sfdc_object):
"""This method initializes the :py:class:`salespyforce.core.Salesforce.Knowledge` inner class object.
:param sfdc_object: The core :py:class:`salespyforce.Salesforce` object
:type sfdc_object: class[salespyforce.Salesforce]
"""
self.sfdc_object = sfdc_object
[docs]
def check_for_existing_article(
self,
title: str,
sobject: Optional[str] = None,
return_id: bool = False,
return_id_and_number: bool = False,
include_archived: bool = False,
):
"""This method checks to see if an article already exists with a given title and returns its article number.
(`Reference 1 <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_query.htm>`_.
`Reference 2 <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/knowledge_development_soql_sosl_intro.htm>`_)
:param title: The title of the knowledge article for which to check
:type title: str
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:param return_id: Determines if the Article ID should be returned (``False`` by default)
:type return_id: bool
:param return_id_and_number: Determines if Article ID and Article Number should be returned (``False`` by default)
:type return_id_and_number: bool
:param include_archived: Determines if archived articles should be included (``False`` by default)
:type include_archived: bool
:returns: The Article Number, Article ID, or both (if found), or a blank string if not found
"""
return knowledge_module.check_for_existing_article(self.sfdc_object, title=title,
sobject=sobject, return_id=return_id,
return_id_and_number=return_id_and_number,
include_archived=include_archived)
[docs]
def get_article_id_from_number(
self,
article_number,
sobject: Optional[str] = None,
return_uri: bool = False,
):
"""This method returns the Article ID when an article number is provided.
(`Reference 1 <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_query.htm>`_,
`Reference 2 <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/knowledge_development_soql_sosl_intro.htm>`_)
.. warning::
The ability to retrieve the article URI/URL rather than the ID will be moved to a separate function in
a future release.
:param article_number: The Article Number to query
:type article_number: str, int
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:param return_uri: Determines if the URI of the article should be returned rather than the ID (``False`` by default)
:type return_uri: bool
:returns: The Article ID or Article URI, or a blank string if no article is found
:raises: :py:exc:`TypeError`,
:py:exc:`RuntimeError`
"""
# TODO: Move return_uri functionality (here and in underlying function) to a separate method/function
return knowledge_module.get_article_id_from_number(self.sfdc_object, article_number=article_number,
sobject=sobject, return_uri=return_uri)
[docs]
def get_articles_list(
self,
query: Optional[str] = None,
sort: Optional[str] = None,
order: Optional[str] = None,
page_size: int = 20,
page_num: int = 1,
) -> list:
"""This method retrieves a list of knowledge articles.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_knowledge_support_artlist.htm>`_)
:param query: A SOQL query with which to filter the results (optional)
:type query: str, None
:param sort: Optionally sort the results with one of the following values: ``LastPublishedDate``,
``CreatedDate``, ``Title``, or ``ViewScore``
:type sort: str, None
:param order: Optionally define the ORDER BY as ``ASC`` or ``DESC``
:type order: str, None
:param page_size: The number of results per page (``20`` by default)
:type page_size: int
:param page_num: The starting page number (``1`` by default)
:type page_num: int
:returns: The list of retrieved knowledge articles
:raises: :py:exc:`RuntimeError`
"""
# TODO: Update to use constants for page_size and page_num
return knowledge_module.get_articles_list(self.sfdc_object, query=query, sort=sort, order=order,
page_size=page_size, page_num=page_num)
[docs]
def get_article_details(
self,
article_id: str,
sobject: Optional[str] = None,
use_knowledge_articles_endpoint: Optional[bool] = None,
):
"""This method retrieves details for a single knowledge article.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_knowledge_support_artdetails.htm>`_)
.. versionchanged:: 1.4.0
A logic issue was resolved and the new optional ``use_knowledge_articles_endpoint`` parameter can
now be set to force the ``knowledgeArticles`` endpoint to be used for the GET request rather than
the ``sobjects`` endpoint.
:param article_id: The Article ID for which to retrieve details
:type article_id: str
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:param use_knowledge_articles_endpoint: Optionally use the ``knowledgeArticles`` endpoint rather than
``sobjects`` to retrieve the article details (``False`` by default)
:type use_knowledge_articles_endpoint: bool, None
:returns: The details for the knowledge article
:raises: :py:exc:`RuntimeError`,
:py:exc:`salespyforce.errors.exceptions.DataMismatchError`
"""
return knowledge_module.get_article_details(self.sfdc_object, article_id=article_id, sobject=sobject,
use_knowledge_articles_endpoint=use_knowledge_articles_endpoint)
[docs]
def get_validation_status(
self,
article_id: Optional[str] = None,
article_details: Optional[dict] = None,
sobject: Optional[str] = None,
) -> str:
"""This method retrieves the Validation Status for a given Article ID.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/resources_knowledge_support_artdetails.htm>`_)
.. versionchanged:: 1.4.0
This method now returns an empty string rather than a ``None`` value if the ``ValidationStatus`` field
is not found in the article details data, and a more specific exception class is used when input
data is missing instead of the generic :py:exc:`RuntimeError` exception class.
:param article_id: The Article ID for which to retrieve details
:type article_id: str, None
:param article_details: The dictionary of article details for the given article
:type article_details: dict, None
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:returns: The validation status as a text string
:raises: :py:exc:`RuntimeError`,
:py:exc:`salespyforce.errors.exceptions.MissingRequiredDataError`
"""
return knowledge_module.get_validation_status(self.sfdc_object, article_id=article_id,
article_details=article_details, sobject=sobject)
[docs]
def get_article_metadata(self, article_id: str) -> dict:
"""This method retrieves metadata for a specific knowledge article.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/knowledge_REST_retrieve_article_metadata.htm>`_)
:param article_id: The Article ID for which to retrieve details
:type article_id: str
:returns: The article metadata as a dictionary
:raises: :py:exc:`RuntimeError`
"""
return knowledge_module.get_article_metadata(self.sfdc_object, article_id=article_id)
[docs]
def get_article_version(self, article_id: str):
"""This method retrieves the version ID for a given master article ID.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/knowledge_REST_retrieve_article_version.htm>`_)
:param article_id: The Article ID for which to retrieve details
:type article_id: str
:returns: The version ID for the given master article ID
:raises: :py:exc:`RuntimeError`
"""
# TODO: Determine how the data is returned and if it needs to be pruned to just the article version
return knowledge_module.get_article_version(self.sfdc_object, article_id=article_id)
[docs]
def get_article_url(
self,
article_id: Optional[str] = None,
article_number=None, # Needs type hint
sobject: Optional[str] = None,
):
"""This function constructs the URL to view a knowledge article in Lightning or Classic.
:param article_id: The Article ID for which to retrieve details
:type article_id: str, None
:param article_number: The article number for which to retrieve details
:type article_number: str, int, None
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:returns: The article URL as a string
:raises: :py:exc:`ValueError`,
:py:exc:`RuntimeError`
"""
return knowledge_module.get_article_url(self.sfdc_object, article_id=article_id,
article_number=article_number, sobject=sobject)
[docs]
def create_article(
self,
article_data: dict,
sobject: Optional[str] = None,
full_response: bool = False,
):
"""This method creates a new knowledge article draft.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_sobject_create.htm>`_)
:param article_data: The article data used to populate the article
:type article_data: dict
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:param full_response: Determines if the full API response should be returned instead of the article ID (``False`` by default)
:type full_response: bool
:returns: The API response or the ID of the article draft
:raises: :py:exc:`ValueError`,
:py:exc:`TypeError`,
:py:exc:`RuntimeError`
"""
return knowledge_module.create_article(self.sfdc_object, article_data=article_data, sobject=sobject,
full_response=full_response)
[docs]
def update_article(
self,
record_id: str,
article_data: dict,
sobject: Optional[str] = None,
include_status_code: bool = False,
):
"""This method updates an existing knowledge article draft.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_update_fields.htm>`_)
:param record_id: The ID of the article draft record to be updated
:type record_id: str
:param article_data: The article data used to update the article
:type article_data: dict
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:param include_status_code: Determines if the API response status code should be returned (``False`` by default)
:type include_status_code: bool
:returns: A Boolean indicating if the update operation was successful, and optionally the API response status code
:raises: :py:exc:`ValueError`,
:py:exc:`TypeError`,
:py:exc:`RuntimeError`
"""
return knowledge_module.update_article(self.sfdc_object, record_id=record_id, article_data=article_data,
sobject=sobject, include_status_code=include_status_code)
[docs]
def create_draft_from_online_article(self, article_id: str, unpublish: bool = False):
"""This method creates a draft knowledge article from an online article.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/actions_obj_knowledge.htm#createDraftFromOnlineKnowledgeArticle>`_)
:param article_id: The ID of the online article from which to create the draft
:type article_id: str
:param unpublish: Determines if the online article should be unpublished when the draft is created (``False`` by default)
:type unpublish: bool
:returns: The API response from the POST request
:raises: :py:exc:`RuntimeError`
"""
return knowledge_module.create_draft_from_online_article(self.sfdc_object, article_id=article_id,
unpublish=unpublish)
[docs]
def create_draft_from_master_version(
self,
article_id: Optional[str] = None,
knowledge_article_id: Optional[str] = None,
article_data: Optional[dict] = None,
sobject: Optional[str] = None,
full_response: bool = False,
):
"""This method creates an online version of a master article.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.198.0.knowledge_dev.meta/knowledge_dev/knowledge_REST_edit_online_master.htm>`_)
:param article_id: The Article ID from which to create the draft
:type article_id: str, None
:param knowledge_article_id: The Knowledge Article ID (``KnowledgeArticleId``) from which to create the draft
:type knowledge_article_id: str, None
:param article_data: The article data associated with the article from which to create the draft
:type article_data: dict, None
:param sobject: The Salesforce object to query (``Knowledge__kav`` by default)
:type sobject: str, None
:param full_response: Determines if the full API response should be returned instead of the article ID (``False`` by default)
:type full_response: bool
:returns: The API response or the ID of the article draft
:raises: :py:exc:`RuntimeError`
"""
return knowledge_module.create_draft_from_master_version(self.sfdc_object, article_id=article_id,
knowledge_article_id=knowledge_article_id,
article_data=article_data, sobject=sobject,
full_response=full_response)
[docs]
def publish_article(self, article_id, major_version=True, full_response=False):
"""This method publishes a draft knowledge article as a major or minor version.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/knowledge_REST_publish_master_version.htm>`_)
:param article_id: The Article ID to publish
:type article_id: str
:param major_version: Determines if the published article should be a major version (``True`` by default)
:type major_version: bool
:param full_response: Determines if the full API response should be returned (``False`` by default)
:type full_response: bool
:returns: A Boolean value indicating the success of the action or the API response from the PATCH request
:raises: :py:exc:`RuntimeError`
"""
return knowledge_module.publish_article(self.sfdc_object, article_id=article_id,
major_version=major_version, full_response=full_response)
[docs]
def publish_multiple_articles(self, article_id_list, major_version=True):
"""This method publishes multiple knowledge article drafts at one time.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/actions_obj_knowledge.htm#publishKnowledgeArticles>`_)
:param article_id_list: A list of Article IDs to be published
:type article_id_list: list
:param major_version: Determines if the published article should be a major version (``True`` by default)
:type major_version: bool
:returns: The API response from the POST request
:raises: :py:exc:`RuntimeError`,
:py:exc:`TypeError`,
:py:exc:`ValueError`
"""
return knowledge_module.publish_multiple_articles(self.sfdc_object, article_id_list=article_id_list,
major_version=major_version)
[docs]
def assign_data_category(self, article_id, category_group_name, category_name):
"""This method assigns a single data category for a knowledge article.
(`Reference <https://itsmemohit.medium.com/quick-win-15-salesforce-knowledge-rest-apis-bb0725b2040e>`_)
.. versionadded:: 1.2.0
:param article_id: The ID of the article to update
:type article_id: str
:param category_group_name: The unique Data Category Group Name
:type category_group_name: str
:param category_name: The unique Data Category Name
:type category_name: str
:returns: The API response from the POST request
:raises: :py:exc:`RuntimeError`
"""
return knowledge_module.assign_data_category(self.sfdc_object, article_id=article_id,
category_group_name=category_group_name,
category_name=category_name)
[docs]
def archive_article(self, article_id):
"""This function archives a published knowledge article.
(`Reference <https://developer.salesforce.com/docs/atlas.en-us.knowledge_dev.meta/knowledge_dev/knowledge_REST_archive_master_version.htm>`_)
.. versionadded:: 1.3.0
:param article_id: The ID of the article to archive
:type article_id: str
:returns: The API response from the POST request
:raises: :py:exc:`RuntimeError`
"""
return knowledge_module.archive_article(self.sfdc_object, article_id=article_id)
[docs]
def delete_article_draft(self, version_id: str, use_knowledge_management_endpoint: bool = True):
"""This function deletes an unpublished knowledge article draft.
.. versionadded:: 1.4.0
:param version_id: The 15-character or 18-character ``Id`` (Knowledge Article Version ID) value
:type version_id: str
:param use_knowledge_management_endpoint: Leverage the ``/knowledgeManagement/articleVersions/masterVersions/``
endpoint rather than the ``/sobjects/Knowledge__kav/`` endpoint
(``True`` by default)
:type use_knowledge_management_endpoint: bool
:returns: The API response from the DELETE request
:raises: :py:exc:`RuntimeError`
"""
return knowledge_module.delete_article_draft(self.sfdc_object, version_id=version_id,
use_knowledge_management_endpoint=use_knowledge_management_endpoint)
[docs]
def define_connection_info():
"""This function prompts the user for the connection information.
:returns: The connection info in a dictionary
"""
base_url = input('Enter your instance URL: [] ')
org_id = input('Enter the Org ID for your instance: [] ')
username = input('Enter the username of your API user: [] ')
password = input('Enter the password of your API user: [] ')
endpoint_url = input('Enter the endpoint URL: [] ')
client_id = input('Enter the Client ID: [] ')
client_secret = input('Enter the Client Secret: [] ')
security_token = input('Enter the Security Token: [] ')
connection_info = compile_connection_info(base_url, org_id, username, password, endpoint_url,
client_id, client_secret, security_token)
return connection_info
[docs]
def compile_connection_info(base_url, org_id, username, password, endpoint_url,
client_id, client_secret, security_token):
"""This function compiles the connection info into a dictionary that can be consumed by the core object.
:param base_url: The base URL of the Salesforce instance
:type base_url: str
:param org_id: The Org ID of the Salesforce instance
:type org_id: str
:param username: The username of the API user
:type username: str
:param password: The password of the API user
:type password: str
:param endpoint_url: The endpoint URL for the Salesforce instance
:type endpoint_url: str
:param client_id: The Client ID for the Salesforce instance
:type client_id: str
:param client_secret: The Client Secret for the Salesforce instance
:type client_secret: str
:param security_token: The Security Token for the Salesforce instance
:type security_token: str
:returns: The connection info in a dictionary
"""
connection_info = {
'base_url': base_url,
'org_id': org_id,
'username': username,
'password': password,
'endpoint_url': endpoint_url,
'client_id': client_id,
'client_secret': client_secret,
'security_token': security_token,
}
return connection_info
