Salesforce Connector

import http.server
import urllib.parse
import webbrowser
import requests

CLIENT_ID = "YOUR_CONSUMER_KEY"
CLIENT_SECRET = "YOUR_CONSUMER_SECRET"

REDIRECT_URI = "http://localhost:8080/callback"
LOGIN_URL = "https://login.salesforce.com"


def _get_oauth_base(login_url):
    if not login_url:
        return "https://login.salesforce.com"
    if "test.salesforce.com" in login_url.lower() or ".sandbox." in login_url.lower():
        return "https://test.salesforce.com"
    return "https://login.salesforce.com"


OAUTH_BASE = _get_oauth_base(LOGIN_URL)


class CallbackHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path.startswith("/callback"):
            query = urllib.parse.urlparse(self.path).query
            params = urllib.parse.parse_qs(query)
            code = params.get("code", [None])[0]

            if code:
                token_url = f"{OAUTH_BASE}/services/oauth2/token"
                data = {
                    "grant_type": "authorization_code",
                    "code": code,
                    "client_id": CLIENT_ID,
                    "client_secret": CLIENT_SECRET,
                    "redirect_uri": REDIRECT_URI,
                }
                resp = requests.post(token_url, data=data)
                tokens = resp.json()

                self.send_response(200)
                self.send_header("Content-type", "text/html")
                self.end_headers()

                if "refresh_token" in tokens:
                    print("\n" + "=" * 50)
                    print("SUCCESS! Copy this refresh token:\n")
                    print(tokens["refresh_token"])
                    print("\n" + "=" * 50)
                    print(f"\nInstance URL: {tokens.get('instance_url')}")
                    self.wfile.write(
                        b"<h1>Success! Check your terminal for the refresh token.</h1>"
                    )
                else:
                    print(f"\nError: {tokens}")
                    self.wfile.write(f"<h1>Error</h1><pre>{tokens}</pre>".encode())
            else:
                self.send_response(400)
                self.end_headers()
                self.wfile.write(b"No code received")

    def log_message(self, format, *args):
        pass


auth_url = (
    f"{OAUTH_BASE}/services/oauth2/authorize"
    f"?response_type=code"
    f"&client_id={CLIENT_ID}"
    f"&redirect_uri={REDIRECT_URI}"
    f"&scope=api%20refresh_token"
)

print("Opening browser for Salesforce login...")
print(f"\nIf browser doesn't open, go to:\n{auth_url}\n")
webbrowser.open(auth_url)

server = http.server.HTTPServer(("localhost", 8080), CallbackHandler)
print("Waiting for callback on http://localhost:8080...")
server.handle_request()

You have access to Salesforce APIs. Use these credentials:

CLIENT_ID = "YOUR_CONSUMER_KEY"
CLIENT_SECRET = "YOUR_CONSUMER_SECRET"
REFRESH_TOKEN = "{{salesforce_refresh_token}}"
INSTANCE_URL = "YOUR_INSTANCE_URL"
LOGIN_URL = "https://login.salesforce.com"

When writing Salesforce scripts, always use this pattern:

import requests

# Determine OAuth base URL
if "test.salesforce.com" in LOGIN_URL.lower() or ".sandbox." in LOGIN_URL.lower():
    oauth_base = "https://test.salesforce.com"
else:
    oauth_base = "https://login.salesforce.com"

# Get access token
resp = requests.post(
    f"{oauth_base}/services/oauth2/token",
    data={
        "grant_type": "refresh_token",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "refresh_token": REFRESH_TOKEN,
    }
)
access_token = resp.json()["access_token"]

# Query Salesforce
resp = requests.get(
    f"{INSTANCE_URL}/services/data/v60.0/query",
    headers={"Authorization": f"Bearer {access_token}"},
    params={"q": "YOUR_SOQL_QUERY"}
)
data = resp.json()

import requests

CLIENT_ID = ""
CLIENT_SECRET = ""
REFRESH_TOKEN = ""
INSTANCE_URL = ""
LOGIN_URL = "https://login.salesforce.com"

try:
    oauth_base = "https://test.salesforce.com" if "sandbox" in LOGIN_URL.lower() else "https://login.salesforce.com"

    response = requests.post(
        f"{oauth_base}/services/oauth2/token",
        data={
            "grant_type": "refresh_token",
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "refresh_token": REFRESH_TOKEN,
        }
    )
    access_token = response.json()["access_token"]
    print(f"Connected. Token: {access_token[:20]}...")

    response = requests.get(
        f"{INSTANCE_URL}/services/data/v60.0/query",
        headers={"Authorization": f"Bearer {access_token}"},
        params={"q": "SELECT Id, Name FROM Account LIMIT 3"}
    )
    records = response.json()["records"]
    print(f"Found {len(records)} accounts:")
    for r in records:
        print(f"  {r['Name']}")

    print("\nSuccess: Connection verified")
except Exception as e:
    print(f"\nFailed: {e}")