[Python] REPLIT - API CONNECTOR to CONNECTED APP IN SALESFORCE (sandbox)

Hello Wyour textord!

I created in Replit an API connector to connect my App to Salesforce (sandbox) Connected App.

I'm using OAuth JWT flow with:

• Consumer Key (SF_CONSUMER_KEY)
• Consumer Secret (SF_CONSUMER_SECRET)
• Instance URL (SF_INSTANCE_URL)

The errors.

test_salesforce.py::test_connection FAILED [ 20%] test_salesforce.py::test_fetch_data FAILED [ 40%] test_salesforce.py::test_create_record FAILED [ 60%] test_salesforce.py::test_batch_create FAILED [ 80%] test_salesforce.py::test_rate_limiting FAILED [100%]

========================================= FAILURES ========================================= _____________________________________ test_connection ______________________________________

self = <api_connectors.SalesforceConnector object at 0x7f9b5b8b4e50>

async def connect(self) -> bool:
for attempt in range(self.MAX_RETRIES):
try:
from simple_salesforce import Salesforce
import logging

logging.info(f"Attempting Salesforce connection - attempt {attempt + 1}")
username = self.credentials.get('username') or os.getenv('SF_USERNAME')
password = self.credentials.get('password') or os.getenv('SF_PASSWORD')
security_token = self.credentials.get('security_token') or os.getenv('SF_SECURITY_TOKEN')
consumer_key = self.credentials.get('consumer_key') or os.getenv('SF_CONSUMER_KEY')
consumer_secret = self.credentials.get('consumer_secret') or os.getenv('SF_CONSUMER_SECRET')

if not all([username, password, security_token]):
raise SalesforceError("Missing required credentials: username, password, or security token")

> self.client = Salesforce(
username=username,
password=password,
security_token=security_token,
consumer_key=consumer_key,
consumer_secret=consumer_secret,
domain='test' # Specifically for developer/sandbox environments
)

api_connectors.py:54:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.pythonlibs/lib/python3.11/site-packages/simple_salesforce/api.py:153: in __init__
self._refresh_session()
.pythonlibs/lib/python3.11/site-packages/simple_salesforce/api.py:307: in _refresh_session
self.session_id, self.sf_instance = self._salesforce_login_partial()
.pythonlibs/lib/python3.11/site-packages/simple_salesforce/login.py:227: in SalesforceLogin
return soap_login(soap_url, login_soap_request_body,


soap_url = 'https://test.salesforce.com/services/Soap/u/59.0'
request_body = '<?xml version="1.0" encoding="utf-8" ?>\n<env:Envelope\n xmlns:xsd="http://www.w3.org/2001/XMLSchema"\n ...assword>Man-Top-Ata.123!uC70ArGo6ATwhysrY4ETlukrI</n1assword>\n </n1:login>\n </env:Body>\n</env:Envelope>'
headers = {'SOAPAction': 'login', 'charset': 'UTF-8', 'content-type': 'text/xml'}
proxies = {}, session = <requests.sessions.Session object at 0x7f9b5cb07710>

def soap_login(
soap_url: str,
request_body: str,
headers: Optional[Headers],
proxies: Optional[Proxies],
session: Optional[requests.Session] = None) -> Tuple[str, str]:
"""Process SOAP specific login workflow."""
response = (session or requests).post(
soap_url, request_body, headers=headers, proxies=proxies)

if response.status_code != 200:
except_code: Union[str, int, None]
except_msg: str
try:
except_code = getUniqueElementValueFromXmlString(
response.content, 'sf:exceptionCode')
except_msg = (getUniqueElementValueFromXmlString(
response.content, 'sf:exceptionMessage')
or response.content.decode())
except ExpatError:
except_code = response.status_code
except_msg = response.content.decode()
> raise SalesforceAuthenticationFailed(except_code, except_msg)
E simple_salesforce.exceptions.SalesforceAuthenticationFailed: INVALID_LOGIN: Invalid username, password, security token; or user locked out.

.pythonlibs/lib/python3.11/site-packages/simple_salesforce/login.py:253: SalesforceAuthenticationFailed

The above exception was the direct cause of the following exception:

connector = <api_connectors.SalesforceConnector object at 0x7f9b5b8b4e50>

@pytest.mark.asyncio
async def test_connection(connector):
> success = await connector.connect()

test_salesforce.py:41:


self = <api_connectors.SalesforceConnector object at 0x7f9b5b8b4e50>

async def connect(self) -> bool:
for attempt in range(self.MAX_RETRIES):
try:
from simple_salesforce import Salesforce
import logging

logging.info(f"Attempting Salesforce connection - attempt {attempt + 1}")
username = self.credentials.get('username') or os.getenv('SF_USERNAME')
password = self.credentials.get('password') or os.getenv('SF_PASSWORD')
security_token = self.credentials.get('security_token') or os.getenv('SF_SECURITY_TOKEN')
consumer_key = self.credentials.get('consumer_key') or os.getenv('SF_CONSUMER_KEY')
consumer_secret = self.credentials.get('consumer_secret') or os.getenv('SF_CONSUMER_SECRET')

if not all([username, password, security_token]):
raise SalesforceError("Missing required credentials: username, password, or security token")

self.client = Salesforce(
username=username,
password=password,
security_token=security_token,
consumer_key=consumer_key,
consumer_secret=consumer_secret,
domain='test' # Specifically for developer/sandbox environments
)
logging.info("Successfully connected to Salesforce")
return True
except Exception as e:
logging.error(f"Salesforce connection error: {str(e)}")
if attempt == self.MAX_RETRIES - 1:
> raise SalesforceError(f"Failed to connect after {self.MAX_RETRIES} attempts") from e
E api_connectors.SalesforceError: Failed to connect after 3 attempts

api_connectors.py:67: SalesforceError


------------------------------------ Captured log call ------------------------------------- ERROR root:api_connectors.py:65 Salesforce connection error: INVALID_LOGIN: Invalid username, password, security token; or user locked out. ERROR root:api_connectors.py:65 Salesforce connection error: INVALID_LOGIN: Invalid username, password, security token; or user locked out. ERROR root:api_connectors.py:65 Salesforce connection error: INVALID_LOGIN: Invalid username, password, security token; or user locked out.

=========================== short test summary info ==================================

FAILED test_salesforce.py::test_connection - api_connectors.SalesforceError: Failed to connect after 3 attempts FAILED test_salesforce.py::test_fetch_data - api_connectors.SalesforceError: Failed to connect after 3 attempts FAILED test_salesforce.py::test_create_record - api_connectors.SalesforceError: Failed to connect after 3 attempts FAILED test_salesforce.py::test_batch_create - api_connectors.SalesforceError: Failed to connect after 3 attempts FAILED test_salesforce.py::test_rate_limiting - api_connectors.SalesforceError: Failed to connect after 3 attempts

This is my api_connector.py:

from abc import ABC, abstractmethod
from typing import Dict, Any, List
from models import Department
import asyncio
import time
import logging
import os
from datetime import datetime

class BaseAPIConnector(ABC):
def __init__(self, credentials: Dict[str, str], department: Department):
self.credentials = credentials
self.department = department

@abstractmethod
async def connect(self) -> bool:
pass

@abstractmethod
async def fetch_data(self, query: str) -> Dict[str, Any]:
pass

class SalesforceError(Exception):
"""Base exception for Salesforce operations"""
pass

class SalesforceConnector(BaseAPIConnector):
MAX_RETRIES = 3
BATCH_SIZE = 200
API_LIMIT_PER_SECOND = 20

def __init__(self, credentials: Dict[str, str], department: Department):
super().__init__(credentials, department)
self._last_request_time = 0
self._request_count = 0

async def connect(self) -> bool:
for attempt in range(self.MAX_RETRIES):
try:
from simple_salesforce import Salesforce
import logging

logging.info(f"Attempting Salesforce connection - attempt {attempt + 1}")
username = self.credentials.get('username') or os.getenv('SF_USERNAME')
password = self.credentials.get('password') or os.getenv('SF_PASSWORD')
security_token = self.credentials.get('security_token') or os.getenv('SF_SECURITY_TOKEN')
consumer_key = self.credentials.get('consumer_key') or os.getenv('SF_CONSUMER_KEY')
consumer_secret = self.credentials.get('consumer_secret') or os.getenv('SF_CONSUMER_SECRET')

if not all([username, password, security_token]):
raise SalesforceError("Missing required credentials: username, password, or security token")

self.client = Salesforce(
username=username,
password=password,
security_token=security_token,
consumer_key=consumer_key,
consumer_secret=consumer_secret,
domain='test' # Specifically for developer/sandbox environments
)
logging.info("Successfully connected to Salesforce")
return True
except Exception as e:
logging.error(f"Salesforce connection error: {str(e)}")
if attempt == self.MAX_RETRIES - 1:
raise SalesforceError(f"Failed to connect after {self.MAX_RETRIES} attempts") from e
await asyncio.sleep(2 ** attempt) # Exponential backoff
return False

async def fetch_data(self, query: str) -> Dict[str, Any]:
"""Execute SOQL query"""
try:
return self.client.query(query)
except Exception as e:
print(f"Salesforce query error: {str(e)}")
raise

async def _rate_limit(self):
"""Implement rate limiting for API calls"""
current_time = time.time()
if current_time - self._last_request_time < 1: # Within 1 second window
if self._request_count >= self.API_LIMIT_PER_SECOND:
await asyncio.sleep(1 - (current_time - self._last_request_time))
self._request_count = 0
self._last_request_time = time.time()
else:
self._request_count = 0
self._last_request_time = current_time
self._request_count += 1

async def create_record(self, sobject: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new Salesforce record"""
await self._rate_limit()
try:
logging.info(f"Creating new {sobject} record")
sf_object = getattr(self.client, sobject)
result = sf_object.create(data)
logging.info(f"Successfully created {sobject} record")
return result
except Exception as e:
logging.error(f"Salesforce create error: {str(e)}")
raise SalesforceError(f"Failed to create {sobject} record") from e

async def create_records_batch(self, sobject: str, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Create multiple records in batches"""
results = []
for i in range(0, len(records), self.BATCH_SIZE):
batch = records[i:i + self.BATCH_SIZE]
await self._rate_limit()
try:
sf_object = getattr(self.client, sobject)
batch_results = sf_object.bulk.create(batch)
results.extend(batch_results)
logging.info(f"Successfully created batch of {len(batch)} {sobject} records")
except Exception as e:
logging.error(f"Batch creation error: {str(e)}")
raise SalesforceError(f"Failed to create batch of {sobject} records") from e
return results

async def update_record(self, sobject: str, record_id: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""Update an existing Salesforce record"""
try:
sf_object = getattr(self.client, sobject)
return sf_object.update(record_id, data)
except Exception as e:
print(f"Salesforce update error: {str(e)}")
raise

async def delete_record(self, sobject: str, record_id: str) -> bool:
"""Delete a Salesforce record"""
try:
sf_object = getattr(self.client, sobject)
return sf_object.delete(record_id)
except Exception as e:
print(f"Salesforce delete error: {str(e)}")
raise

async def describe_object(self, sobject: str) -> Dict[str, Any]:
"""Get metadata about a Salesforce object"""
try:
sf_object = getattr(self.client, sobject)
return sf_object.describe()
except Exception as e:
print(f"Salesforce describe error: {str(e)}")
raise

async def get_deleted_records(self, sobject: str, start: datetime, end: datetime) -> Dict[str, Any]:
"""Get records deleted between start and end dates"""
try:
sf_object = getattr(self.client, sobject)
return sf_object.deleted(start, end)
except Exception as e:
print(f"Salesforce get deleted records error: {str(e)}")
raise

async def get_updated_records(self, sobject: str, start: datetime, end: datetime) -> Dict[str, Any]:
"""Get records updated between start and end dates"""
try:
sf_object = getattr(self.client, sobject)
return sf_object.updated(start, end)
except Exception as e:
print(f"Salesforce get updated records error: {str(e)}")
raise

class DynamicsConnector(BaseAPIConnector):
async def connect(self) -> bool:
try:
# Implement Dynamics connection logic
return True
except Exception as e:
print(f"Dynamics connection error: {str(e)}")
return False

async def fetch_data(self, query: str) -> Dict[str, Any]:
# Implement Dynamics query execution
pass

class QlikConnector(BaseAPIConnector):
async def connect(self) -> bool:
try:
# Implement Qlik connection logic
return True
except Exception as e:
print(f"Qlik connection error: {str(e)}")
return False

async def fetch_data(self, query: str) -> Dict[str, Any]:
# Implement Qlik query execution
pass


This is my test_salesforce.py

import pytest
import asyncio
import time
import os
from unittest.mock import Mock, patch, MagicMock
from api_connectors import SalesforceConnector, SalesforceError
from models import Department

@pytest.fixture
def test_credentials():
return {
"username": os.getenv('SF_USERNAME'),
"password": os.getenv('SF_PASSWORD'),
"security_token": os.getenv('SF_SECURITY_TOKEN'),
"consumer_key": os.getenv('SF_CONSUMER_KEY'),
"consumer_secret": os.getenv('SF_CONSUMER_SECRET')
}

@pytest.fixture
def mock_sf():
mock = MagicMock()
mock.query.return_value = {'records': []}
mock.Account = MagicMock()
mock.Account.create.return_value = {'success': True, 'id': '001xxx'}
mock.Account.bulk = MagicMock()
mock.Account.bulk.create.return_value = [{'success': True} for _ in range(5)]
return mock

@pytest.fixture
def connector(test_credentials):
with patch('simple_salesforce.Salesforce') as mock_sf_class:
mock_sf_instance = MagicMock()
mock_sf_class.return_value = mock_sf_instance
mock_sf_instance.query.return_value = {'records': []}
connector = SalesforceConnector(test_credentials, Department.MARKETING)
return connector

@pytest.mark.asyncio
async def test_connection(connector):
success = await connector.connect()
assert success == True

@pytest.mark.asyncio
async def test_fetch_data(connector):
await connector.connect()
result = await connector.fetch_data("SELECT Id, Name FROM Account LIMIT 5")
assert 'records' in result

@pytest.mark.asyncio
async def test_create_record(connector):
await connector.connect()
data = {"Name": "Test Account", "Description": "Test Description"}
result = await connector.create_record("Account", data)
assert result['success'] == True

@pytest.mark.asyncio
async def test_batch_create(connector):
await connector.connect()
records = [
{"Name": f"Test Account {i}", "Description": "Batch Test"}
for i in range(5)
]
results = await connector.create_records_batch("Account", records)
assert len(results) == 5

@pytest.mark.asyncio
async def test_rate_limiting(connector):
await connector.connect()
start_time = time.time()
requests = [connector._rate_limit() for _ in range(25)]
await asyncio.gather(*requests)
duration = time.time() - start_time
assert duration >= 1 # Should take at least 1 second due to rate limiting

Continue reading...