Step 2: Build the Python Client

Create a Python file in any editor. For example, create a file named PubSubAPIClient.py.

To import these modules, add these import statements at the top of the file.

1import grpc
2import requests
3import threading
4import io
5import pubsub_api_pb2 as pb2
6import pubsub_api_pb2_grpc as pb2_grpc
7import avro.schema
8import avro.io
9import time
10import certifi
11import json

Set a semaphore at the beginning of the program.
1semaphore = threading.Semaphore(1)
Because of Python gRPC’s design, the program shuts down immediately if no response comes back in the milliseconds between calling an RPC and the end of the program. When you set a semaphore, the client keeps running indefinitely. In this example, you define the semaphore globally. Because this sample starts one subscription via one Subscribe RPC call, the global semaphore works. However, the global semaphore isn’t appropriate for multiple subscriptions because the semaphore is shared. If you change this sample code to make multiple Subscribe RPC calls, make sure that each Subscribe call uses a separate semaphore and the semaphore isn’t shared among subscriptions.
Create a global variable to store the replay ID.
1latest_replay_id = None

Set up the gRPC channel, and generate the stub. The gRPC channel points to the Pub/Sub API endpoint api.pubsub.salesforce.com:7443. In this quick start, we use port 7443, but you can alternatively use port 443. If your company uses a firewall for network security, make sure that the Pub/Sub API endpoint is allowed in the firewall configuration.

1with open(certifi.where(), 'rb') as f:
2    creds = grpc.ssl_channel_credentials(f.read())
3with grpc.secure_channel('api.pubsub.salesforce.com:7443', creds) as channel:
4    # All of the code in the rest of the tutorial will go inside this block.
5    # Make sure that the indentation of the new code you add starts from this comment’s block.

Retrieve your session token. Replace the username, password, and API login URL with values for your Salesforce org. Replace MyDomainName with your org’s My Domain name. For more information, see My Domain URL Formats in Salesforce Help. If you’re using a production instance, the API login URL is https://MyDomainName.my.salesforce.com/services/Soap/u/59.0/. If it’s a sandbox, the URL is https://MyDomainName--SandboxName.sandbox.my.salesforce.com/services/Soap/u/59.0/. Send a POST request formatted like this to the login URL.
1username = '{your username}' 2password = '{your password}' 3url = '{the appropriate login URL}' 4headers = {'content-type': 'text/xml', 'SOAPAction': 'login'} 5xml = f"""<soapenv:Envelope xmlns:soapenv='http://schemas.xmlsoap.org/soap/envelope/' 6xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' 7xmlns:urn='urn:partner.soap.sforce.com'><soapenv:Body> 8<urn:login> 9<urn:username><![CDATA[{username}]]></urn:username> 10<urn:password><![CDATA[{password}]]></urn:password> 11</urn:login></soapenv:Body></soapenv:Envelope>""" 12res = requests.post(url, data=xml, headers=headers, verify=False) 13#Optionally, print the content field returned 14print(res.content)
- When you copy the code snippet, ensure that it aligns with the indented block of code above it. In Python, code blocks are delimited by their indentation.
- If you haven’t set up a range of trusted IP addresses for your org, append a security token to your password. For more information, see Reset Your Security Token and Set Trusted IP Ranges for Your Org.
- The code example uses username and password authentication for simplicity, but we recommend that you use OAuth in production apps. For more information, see Supported Authentication.
Note
Run this client by entering this command on the command line.
1python3 PubSubAPIClient.py
When the request returns, you have XML-formatted data in the response content field res.content. It contains a session ID wrapped within the <sessionId> tags. It also contains the server URL wrapped within the <serverUrl> tags and the org ID within the <organizationId> tags under <userInfo>. Note those values because you use them in the next step.

After getting the session information, you can delete or comment out the login code you added in the previous step. To comment out a line in Python, prepend it with #.
Note
Store the authorization information, including session ID, instance URL, and org ID, in a tuple called authmetadata. Each element in this tuple is also a tuple. You use this information when subscribing to a channel or publishing events.
1. Replace the sessionid placeholder value with the session ID that you got from the previous step.
2. Replace the instanceurl placeholder value with the first part of the server URL that you got from the previous step, without the path portion. For example, https://MyDomainName.my.salesforce.com.
3. Replace the tenantid placeholder value with the org ID that you got from the previous step. For more information about the headers, see Include Authorization Headers in RPC Method Calls.
  1sessionid = '{the session ID you just got from the XML}' 2instanceurl = '{the first part of the server URL ending in .com}' 3tenantid = '{organization ID}' 4authmetadata = (('accesstoken', sessionid), 5('instanceurl', instanceurl), 6('tenantid', tenantid))
Generate your stub object like this. PubSubStub comes from the pubsub_api_pb2_grpc.py file, which you generated in the previous step.
1stub = pb2_grpc.PubSubStub(channel)