salesforcefoundation/salesforce-streaming-client

A Salesforce streaming API client for python, built on salesforce-requests-oauthlib and python-bayeux.

salesforce-streaming-client

A Salesforce streaming API client for python, built on salesforce-requests-oauthlib and python-bayeux.

Please note: as of the 1.0.0 release of python-bayeux (a dependency of salesforce-streaming-client), code using this library will need to call gevent's monkey patching mechanisms. It may have previously been possible to rely on python-bayeux's internal call, but that is no longer supported.

Also, as of the 1.0.0 release of python-bayeux, it no longer officially supports python 2. It may be possible to continue using this library with python 2 but it is not supported.

Tests

To run tests, install py.test and pytest-cov in your virtualenv and

$ py.test --cov=src/salesforce_streaming_client/ --cov-report html:coverage

View test coverage results at ./coverage.

Credits

• Distribute
• modern-package-template
• py.test
• pytest-cov

salesforce-streaming-client's Issues

Update to underlying library causes session creation to fail.

pip list output of packages where instantiating SalesforceOAuth2Session was failing:

certifi 2018.11.29
chardet 3.0.4
Click 7.0
idna 2.8
oauthlib 3.0.1
pip 19.0.2
psycopg2-binary 2.7.7
requests 2.21.0
requests-oauthlib 1.2.0
salesforce-requests-oauthlib 0.1.11
setuptools 40.6.3
six 1.12.0
urllib3 1.24.1

Change Data Capture

Looking through the code it looks like that Change Data Capture events (/data/ are not supported?

def subscribe(self, channel, callback, replay=True, autocreate=True):
    type = None
    if channel.startswith('/u/'):
        type = 'generic'
    elif channel.startswith('/topic/'):
        type = 'push_topic'
    elif channel.startswith('/event/'):
        type = 'event'
    else:
        raise BadSubscriptionException(
            '{0} is not a valid subscription channel.'.format(
                channel
            )
        )

Error with 0.1.2

Hello,

I would like to do a PR for Python3 support (mainly a xrange to range and some iteritemsstuff) but I just found that we use the 0.1.1 version. I tried the 0.1.2 but it does not work for us.

Our error is on this line : 50614ed#diff-15c82b19be1100d35139cde27c69f4ccR233

Objects that we received from SF does not have a payload key, only event, sobject and channel.

I'm not sure why you made this change, can we found a way to have the lib work for us, too ? Let me know if I can do something.

Thanks for all.

403::Organization concurrent user limit exceeded

I wanted to stream the client continuously. Maximum 20 client run cocncurrently in salesforcestreaming after that we got an error like this 403::Organization concurrent user limit exceeded . I am running One client concurrently but after sometime, I got an error like this ..
[aiocometd.exceptions.ServerError: ('Connection closed by the server', {'ext': {'sfdc': {'failureReason': '403::Organization concurrent user limit exceeded'}, 'replay': True, 'payload.format': True}, 'advice': {'reconnect': 'none'}, 'channel': '/meta/handshake', 'id': '1', 'error': '403::Handshake denied', 'successful': False}) ] .
I don't know why this error coming Can you give some solution about this issue?

After Subscribe topic nothing append. Why?

Hi All,
I don't know if I'm using in the right way this lib, but seems that something is missed.
Here the code of my simple Python client:

`import salesforce_streaming_client as sfclient
import time
import logging

logging.basicConfig(filename='out.log',level=logging.DEBUG)

def onMessage():
print "An event arrived"

new_callback_function = lambda new_name: onMessage()

client= sfclient.SalesforceStreamingClient(
oauth_client_id="3MVG9I5UQ_0k_hTm1tiC8kORoabEs.2KFt0oBgXo5gjOpeIeS7lgpgaoIMNKvUhuv4E5tXTSqcrW7ynO6nzI3",
client_secret="1738847514339357499",
username="XXXXXXXXXXXXXXX",
settings_path=None,
sandbox=False,
local_server_settings=('localhost', 60443),
password="XXXXXXXXXXXXXXXX",
ignore_cached_refresh_tokens=False,
version="41.0",
replay_client_id="0001",
cleanup_datadir=True)

some=client.handshake()
print some

client.subscribe(
channel="/event/Device_Action_Event__e",
callback=new_callback_function,
replay=True,
autocreate=True)

while 1:
time.sleep(1)

`

And this is the log content during and after the client execution

##########################################
DEBUG:requests_oauthlib.oauth2_session:Encoding client_id "3MVG9I5UQ_0k_hTm1tiC8kORoabEs.2KFt0oBgXo5gjOpeIeS7lgpgaoIMNKvUhuv4E5tXTSqcrW7ynO6nzI3" with client_secret as Basic auth credentials.
DEBUG:requests_oauthlib.oauth2_session:Requesting url https://login.salesforce.com/services/oauth2/token using method POST.
DEBUG:requests_oauthlib.oauth2_session:Supplying headers {u'Content-Type': u'application/x-www-form-urlencoded;charset=UTF-8', u'Accept': u'application/json'} and data {u'username': u'XXXXXXXXX', u'grant_type': u'password', u'redirect_uri': u'https://localhost:60443', u'client_id': u'3MVG9I5UQ_0k_hTm1tiC8kORoabEs.2KFt0oBgXo5gjOpeIeS7lgpgaoIMNKvUhuv4E5tXTSqcrW7ynO6nzI3', u'client_secret': u'1738847514339357499', u'password': u'XXXXXXXXXXXXX'}
DEBUG:requests_oauthlib.oauth2_session:Passing through key word arguments {'verify': True, 'json': None, 'proxies': None, 'timeout': None, 'auth': <requests.auth.HTTPBasicAuth object at 0x95a1f2c>}.
DEBUG:urllib3.connectionpool:Starting new HTTPS connection (1): login.salesforce.com
DEBUG:urllib3.connectionpool:https://login.salesforce.com:443 "POST /services/oauth2/token HTTP/1.1" 200 None
DEBUG:requests_oauthlib.oauth2_session:Prepared fetch token request body grant_type=password&username=nguarnacci%40ppm.enel&redirect_uri=https%3A%2F%2Flocalhost%3A60443&client_id=3MVG9I5UQ_0k_hTm1tiC8kORoabEs.2KFt0oBgXo5gjOpeIeS7lgpgaoIMNKvUhuv4E5tXTSqcrW7ynO6nzI3&client_secret=1738847514339357499&password=salesforce1kqSZnsRVkXseP3TqOl0RU5pX
DEBUG:requests_oauthlib.oauth2_session:Request to fetch token completed with status 200.
DEBUG:requests_oauthlib.oauth2_session:Request headers were {'Content-Length': '270', 'Accept-Encoding': 'gzip, deflate', 'Accept': u'application/json', 'User-Agent': 'python-requests/2.18.4', 'Connection': 'keep-alive', 'Content-Type': u'application/x-www-form-urlencoded;charset=UTF-8', 'Authorization': 'Basic M01WRzlJNVVRXzBrX2hUbTF0aUM4a09Sb2FiRXMuMktGdDBvQmdYbzVnak9wZUllUzdsZ3BnYW9JTU5LdlVodXY0RTV0WFRTcWNyVzd5bk82bnpJMzoxNzM4ODQ3NTE0MzM5MzU3NDk5'}
DEBUG:requests_oauthlib.oauth2_session:Request body was username=nguarnacci%40ppm.enel&grant_type=password&redirect_uri=https%3A%2F%2Flocalhost%3A60443&client_id=3MVG9I5UQ_0k_hTm1tiC8kORoabEs.2KFt0oBgXo5gjOpeIeS7lgpgaoIMNKvUhuv4E5tXTSqcrW7ynO6nzI3&client_secret=1738847514339357499&password=salesforce1kqSZnsRVkXseP3TqOl0RU5pX
DEBUG:requests_oauthlib.oauth2_session:Response headers were {'X-XSS-Protection': '1; mode=block', 'X-Content-Type-Options': 'nosniff', 'Content-Security-Policy': 'upgrade-insecure-requests', 'Content-Encoding': 'gzip', 'Transfer-Encoding': 'chunked', 'Set-Cookie': 'BrowserId=b5cJYZ3yRlyqw30QZL0TOQ;Path=/;Domain=.salesforce.com;Expires=Sun, 11-Mar-2018 10:42:17 GMT;Max-Age=5184000', 'Strict-Transport-Security': 'max-age=31536000; includeSubDomains', 'Vary': 'Accept-Encoding', 'X-ReadOnlyMode': 'false', 'Cache-Control': 'no-cache,must-revalidate,max-age=0,no-store,private', 'Date': 'Wed, 10 Jan 2018 10:42:17 GMT', 'Content-Type': 'application/json;charset=UTF-8', 'Expires': 'Thu, 01 Jan 1970 00:00:00 GMT'} and content {"access_token":"00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ","instance_url":"https://enelppm.my.salesforce.com","id":"https://login.salesforce.com/id/00D0O000000ZkSUUA0/0050O000007Sm31QAC","token_type":"Bearer","issued_at":"1515580938021","signature":"rFu5Qr5sgYnyPmwfhfqL/zEb0D6WlWz+1elXu+iZRvQ="}.
DEBUG:requests_oauthlib.oauth2_session:Invoking 0 token response hooks.
DEBUG:requests_oauthlib.oauth2_session:Obtained token {u'token_type': u'Bearer', u'issued_at': u'1515580938021', u'signature': u'rFu5Qr5sgYnyPmwfhfqL/zEb0D6WlWz+1elXu+iZRvQ=', u'access_token': u'00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ', u'instance_url': u'https://enelppm.my.salesforce.com', u'id': u'https://login.salesforce.com/id/00D0O000000ZkSUUA0/0050O000007Sm31QAC'}.
INFO:python_bayeux:_send_message(): payload: {'ext': {'replay': True}, 'version': '1.0', 'channel': '/meta/handshake', 'supportedConnectionTypes': ['long-polling'], 'id': '1', 'minimumVersion': '1.0'} kwargs: {}
DEBUG:requests_oauthlib.oauth2_session:Invoking 0 protected resource request hooks.
DEBUG:requests_oauthlib.oauth2_session:Adding token {u'token_type': u'Bearer', u'issued_at': u'1515580938021', u'signature': u'rFu5Qr5sgYnyPmwfhfqL/zEb0D6WlWz+1elXu+iZRvQ=', u'access_token': u'00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ', u'instance_url': u'https://enelppm.my.salesforce.com', u'id': u'https://login.salesforce.com/id/00D0O000000ZkSUUA0/0050O000007Sm31QAC'} to request.
DEBUG:requests_oauthlib.oauth2_session:Requesting url https://enelppm.my.salesforce.com/cometd/41.0/ using method POST.
DEBUG:requests_oauthlib.oauth2_session:Supplying headers {u'Authorization': u'Bearer 00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ'} and data {"ext": {"replay": true}, "version": "1.0", "channel": "/meta/handshake", "supportedConnectionTypes": ["long-polling"], "id": "1", "minimumVersion": "1.0"}
DEBUG:requests_oauthlib.oauth2_session:Passing through key word arguments {'json': None}.
DEBUG:urllib3.connectionpool:Starting new HTTPS connection (1): enelppm.my.salesforce.com
DEBUG:urllib3.connectionpool:https://enelppm.my.salesforce.com:443 "POST /cometd/41.0/ HTTP/1.1" 200 228
INFO:python_bayeux:_send_message(): response status code: 200 response.text: [{"ext":{"replay":true,"payload.format":true},"minimumVersion":"1.0","clientId":"13l1uaz5a53s9isw15x44ywf5zm6h","supportedConnectionTypes":["long-polling"],"channel":"/meta/handshake","id":"1","version":"1.0","successful":true}]
INFO:python_bayeux:_send_message(): payload: {'clientId': u'13l1uaz5a53s9isw15x44ywf5zm6h', 'connectionType': 'long-polling', 'channel': '/meta/connect', 'id': '2'} kwargs: {'timeout': None}
DEBUG:requests_oauthlib.oauth2_session:Invoking 0 protected resource request hooks.
DEBUG:requests_oauthlib.oauth2_session:Adding token {u'token_type': u'Bearer', u'issued_at': u'1515580938021', u'signature': u'rFu5Qr5sgYnyPmwfhfqL/zEb0D6WlWz+1elXu+iZRvQ=', u'access_token': u'00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ', u'instance_url': u'https://enelppm.my.salesforce.com', u'id': u'https://login.salesforce.com/id/00D0O000000ZkSUUA0/0050O000007Sm31QAC'} to request.
DEBUG:requests_oauthlib.oauth2_session:Requesting url https://enelppm.my.salesforce.com/cometd/41.0/ using method POST.
DEBUG:requests_oauthlib.oauth2_session:Supplying headers {u'Authorization': u'Bearer 00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ'} and data {"clientId": "13l1uaz5a53s9isw15x44ywf5zm6h", "connectionType": "long-polling", "channel": "/meta/connect", "id": "2"}
DEBUG:requests_oauthlib.oauth2_session:Passing through key word arguments {'json': None, 'timeout': None}.
DEBUG:urllib3.connectionpool:https://enelppm.my.salesforce.com:443 "POST /cometd/41.0/ HTTP/1.1" 200 160
INFO:python_bayeux:_send_message(): response status code: 200 response.text: [{"clientId":"13l1uaz5a53s9isw15x44ywf5zm6h","advice":{"interval":0,"timeout":110000,"reconnect":"retry"},"channel":"/meta/connect","id":"2","successful":true}]
INFO:python_bayeux:_send_message(): payload: {'ext': {'replay': True}, 'version': '1.0', 'channel': '/meta/handshake', 'supportedConnectionTypes': ['long-polling'], 'id': '1', 'minimumVersion': '1.0'} kwargs: {}
DEBUG:requests_oauthlib.oauth2_session:Invoking 0 protected resource request hooks.
DEBUG:requests_oauthlib.oauth2_session:Adding token {u'token_type': u'Bearer', u'issued_at': u'1515580938021', u'signature': u'rFu5Qr5sgYnyPmwfhfqL/zEb0D6WlWz+1elXu+iZRvQ=', u'access_token': u'00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ', u'instance_url': u'https://enelppm.my.salesforce.com', u'id': u'https://login.salesforce.com/id/00D0O000000ZkSUUA0/0050O000007Sm31QAC'} to request.
DEBUG:requests_oauthlib.oauth2_session:Requesting url https://enelppm.my.salesforce.com/cometd/41.0/ using method POST.
DEBUG:requests_oauthlib.oauth2_session:Supplying headers {u'Authorization': u'Bearer 00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ'} and data {"ext": {"replay": true}, "version": "1.0", "channel": "/meta/handshake", "supportedConnectionTypes": ["long-polling"], "id": "1", "minimumVersion": "1.0"}
DEBUG:requests_oauthlib.oauth2_session:Passing through key word arguments {'json': None}.
DEBUG:urllib3.connectionpool:https://enelppm.my.salesforce.com:443 "POST /cometd/41.0/ HTTP/1.1" 200 227
INFO:python_bayeux:_send_message(): response status code: 200 response.text: [{"ext":{"replay":true,"payload.format":true},"minimumVersion":"1.0","clientId":"13mnj1dh0rqgrzd1ixo0sru5zhe1","supportedConnectionTypes":["long-polling"],"channel":"/meta/handshake","id":"1","version":"1.0","successful":true}]
INFO:python_bayeux:_send_message(): payload: {'clientId': u'13mnj1dh0rqgrzd1ixo0sru5zhe1', 'connectionType': 'long-polling', 'channel': '/meta/connect', 'id': '2'} kwargs: {'timeout': None}
DEBUG:requests_oauthlib.oauth2_session:Invoking 0 protected resource request hooks.
DEBUG:requests_oauthlib.oauth2_session:Adding token {u'token_type': u'Bearer', u'issued_at': u'1515580938021', u'signature': u'rFu5Qr5sgYnyPmwfhfqL/zEb0D6WlWz+1elXu+iZRvQ=', u'access_token': u'00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ', u'instance_url': u'https://enelppm.my.salesforce.com', u'id': u'https://login.salesforce.com/id/00D0O000000ZkSUUA0/0050O000007Sm31QAC'} to request.
DEBUG:requests_oauthlib.oauth2_session:Requesting url https://enelppm.my.salesforce.com/cometd/41.0/ using method POST.
DEBUG:requests_oauthlib.oauth2_session:Supplying headers {u'Authorization': u'Bearer 00D0O000000ZkSU!ARcAQFI3kRYeblhaa1veEx2Jw70NGpky.I9n8GUI2cXSF.ltF.x84NYE04z2hC3YnAp7CHDKmLqmuc4dUoeruwsucFiNYaoZ'} and data {"clientId": "13mnj1dh0rqgrzd1ixo0sru5zhe1", "connectionType": "long-polling", "channel": "/meta/connect", "id": "2"}
DEBUG:requests_oauthlib.oauth2_session:Passing through key word arguments {'json': None, 'timeout': None}.
DEBUG:urllib3.connectionpool:https://enelppm.my.salesforce.com:443 "POST /cometd/41.0/ HTTP/1.1" 200 159
INFO:python_bayeux:_send_message(): response status code: 200 response.text: [{"clientId":"13mnj1dh0rqgrzd1ixo0sru5zhe1","advice":{"interval":0,"timeout":110000,"reconnect":"retry"},"channel":"/meta/connect","id":"2","successful":true}]
INFO:python_bayeux:enqueueing subscription for channel /event/Device_Action_Event__e

####################################################################

Seems all works fine but nothing append when I throw an event from Salesforce side on the that topic channel (considering that another client wrote in Java subscribed at the same time to the same topic works fine).

Could you help me to understand?

xrange and python3

Hi there,

replacing "xrange" with "range" is the only fix to make this python3 compatible (line 113).

cheers and thanks for the module, works great.

-ivan

Callback url limitation

self.callback_url = 'https://{0}:{1}'.format(
            self.local_server_settings[0],
            str(self.local_server_settings[1])
        )

in salesforce_requ:ests_oauthlib has stricted the callback url to have the https://domain:port format. so urls like https://localhost:4444/oauth/_callback are not possible?

Timestamp parsing breaks on different format

https://github.com/SalesforceFoundation/salesforce-streaming-client/blob/master/src/salesforce_streaming_client/__init__.py#L97-L101

Hi, I'm seeing a ValueError at this line, very rarely, when I receive a timestamp without the fractional seconds included. Just wondered if this was a known issue or something that I can work around.

This output is with the following modification:

def string_to_datetime(date_string):
    print('Parsing string:', date_string)
    try:
        return datetime.strptime(
            date_string,
            '%Y-%m-%dT%H:%M:%S.%fZ'
        ).replace(tzinfo=pytz.utc)
    except ValueError:
        return datetime.strptime(
            date_string,
            '%Y-%m-%dT%H:%M:%SZ'
        ).replace(tzinfo=pytz.utc)

Parsing string: 2019-09-11T11:31:03.855000Z
Parsing string: 2019-09-11T11:31:03.877000Z
Parsing string: 2019-09-11T11:31:04.020000Z
Parsing string: 2019-09-11T11:31:29.272000Z
Parsing string: 2019-09-11T11:31:29.396000Z
Parsing string: 2019-09-11T11:31:29.539000Z
Parsing string: 2019-09-11T09:11:26Z

Traceback (most recent call last):
  File "salesforce_to_kafka.py", line 224, in <module>
    run(None, None)
  File "salesforce_to_kafka.py", line 211, in run
    gevent.with_timeout(60*7, client.block)
  File "/Users/matt/.pyenv/versions/salesforce_to_kafka/lib/python3.6/site-packages/gevent/timeout.py", line 367, in with_timeout
    return function(*args, **kwds)
  File "/Users/matt/.pyenv/versions/salesforce_to_kafka/lib/python3.6/site-packages/python_bayeux/__init__.py", line 443, in block
    self._execute_greenlet()
  File "/Users/matt/.pyenv/versions/salesforce_to_kafka/lib/python3.6/site-packages/python_bayeux/__init__.py", line 268, in _execute_greenlet
    getattr(self, callback)(message_queue_message)
  File "/Users/matt/.pyenv/versions/salesforce_to_kafka/lib/python3.6/site-packages/salesforce_streaming_client/__init__.py", line 511, in generic_callback
    'created_date': string_to_datetime(created_date),
  File "/Users/matt/.pyenv/versions/salesforce_to_kafka/lib/python3.6/site-packages/salesforce_streaming_client/__init__.py", line 101, in string_to_datetime
    '%Y-%m-%dT%H:%M:%S.%fZ'
  File "/Users/matt/.pyenv/versions/3.6.4/lib/python3.6/_strptime.py", line 565, in _strptime_datetime
    tt, fraction = _strptime(data_string, format)
  File "/Users/matt/.pyenv/versions/3.6.4/lib/python3.6/_strptime.py", line 362, in _strptime
    (data_string, format))
ValueError: time data '2019-09-11T09:11:26Z' does not match format '%Y-%m-%dT%H:%M:%S.%fZ'

Thanks,

Matt

Error While running py.test

ImportError while importing test module '/root/Streaming/src/salesforce-streaming-client/tests/test_it.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
Streaming/src/salesforce-streaming-client/tests/test_it.py:33: in
from salesforce_streaming_client import SalesforceStreamingClient
E ImportError: No module named salesforce_streaming_client

can you please tell me, where is problem ?