Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Introduce a switch for sandbox and dev enviroments #7

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Introduce a switch for sandbox and dev enviroments #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ Do this once when you do not have consumer key, secret, refresh token:
$ python3

>>> from pypardot.client import PardotAPI
>>> p = PardotAPI(version=3) # verion=4 available
>>> p = PardotAPI(version=3) # verion=4 available, set production_instance=False for sandboxes or developer environments
>>> p.setup_salesforce_auth_keys()
```

Expand Down
39 changes: 26 additions & 13 deletions pypardot/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,14 @@
except ImportError:
import simplejson as json

BASE_URI = 'https://pi.pardot.com'

logger = logging.getLogger(__name__)


class PardotAPI(object):
def __init__(self, email=None, password=None, user_key=None,
sf_consumer_key=None, sf_consumer_secret=None,
sf_refresh_token=None, business_unit_id=None,
sf_refresh_token=None, business_unit_id=None,
production_instance=True,
version=4):
self.email = email
self.password = password
Expand All @@ -32,6 +31,15 @@ def __init__(self, email=None, password=None, user_key=None,
self.sf_consumer_key = sf_consumer_key
self.sf_consumer_secret = sf_consumer_secret
self.business_unit_id = business_unit_id

if production_instance == True:
self.domain = 'https://pi.pardot.com'
self.oauth_domain = 'https://login.salesforce.com/'
else:
"""Sandbox or developer environment"""
self.domain = 'https://pi.demo.pardot.com'
self.oauth_domain = 'https://test.salesforce.com/'

self._load_objects()

def _load_objects(self):
Expand All @@ -53,7 +61,7 @@ def setup_salesforce_auth_keys(self, instance_id=None, business_unit_id=None, co
- API Name
- Contact Email.
7. Go to API (Enable OAuth Settings), and select Enable OAuth Settings.
- In the Callback URL field, enter https://login.salesforce.com/
- In the Callback URL field, enter """ + self.oauth_domain + """
- In the Selected OAuth Scopes field, select:
- Access and manage your data (api)
- Perform requests on your behalf at any time (refresh_token, offline_access)
Expand All @@ -73,6 +81,8 @@ def setup_salesforce_auth_keys(self, instance_id=None, business_unit_id=None, co
(From https://developer.pardot.com/kb/authentication/ )""")
sys.stdout.write("What is your business unit ID?: ")
self.business_unit_id = input()
else:
self.business_unit_id = business_unit_id
if not consumer_key or not consumer_secret:
print("""Get your consumer key:
1. Login to Salesforce. Switch to Lightening Experience if on classic (right-top link).
Expand All @@ -83,31 +93,34 @@ def setup_salesforce_auth_keys(self, instance_id=None, business_unit_id=None, co
self.sf_consumer_key = input()
sys.stdout.write("What is your consumer secret?: ")
self.sf_consumer_secret = input()
url = f"https://{instance_id}.salesforce.com/services/oauth2/authorize?response_type=code&client_id={self.sf_consumer_key}&redirect_uri=https://login.salesforce.com/"
else:
self.sf_consumer_key = consumer_key
self.sf_consumer_secret = consumer_secret
url = f"https://{instance_id}.salesforce.com/services/oauth2/authorize?response_type=code&client_id={self.sf_consumer_key}&redirect_uri={self.oauth_domain}"
print(f"""\nOpen the following page in a browser {url}.
Allow access if any alert popup. You will be redirected to a login page, but do not login.""")
sys.stdout.write("Copy and page the entire URL of the login page that contains code: ")
new_url = input()
parsed = urlparse(new_url)
code = parse_qs(parsed.query)["code"][0]

post_url = f"https://login.salesforce.com/services/oauth2/token?code={code}&grant_type=authorization_code&client_id={self.sf_consumer_key}&client_secret={self.sf_consumer_secret}&redirect_uri=https://login.salesforce.com/"
post_url = f"{self.oauth_domain}services/oauth2/token?code={code}&grant_type=authorization_code&client_id={self.sf_consumer_key}&client_secret={self.sf_consumer_secret}&redirect_uri={self.oauth_domain}"
response = requests.post(post_url).json()
self.sftoken = response.get("access_token")
self.sftoken_refresh = response.get("refresh_token")
if not self.sftoken:
raise Exception("Failed to obtain token %s" % response)

def revoke_sf_token(self):
url = "https://login.salesforce.com/services/oauth2/revoke"
url = f"{self.oauth_domain}services/oauth2/revoke"
# header = {"Content-Type": "application/x-www-form-urlencoded"}
response = requests.post(url,
data={"token": self.sftoken})
if response.status_code >= 400:
raise Exception(response)

def refresh_sf_token(self):
url = "https://login.salesforce.com/services/oauth2/token"
url = f"{self.oauth_domain}services/oauth2/token"
data = {"grant_type": "refresh_token",
"client_id": self.sf_consumer_key,
"client_secret": self.sf_consumer_secret,
Expand Down Expand Up @@ -147,7 +160,7 @@ def post(self, object_name, path=None,

try:
self._check_auth(object_name=object_name)
request = requests.post(self._full_path(object_name, self.version, path),
request = requests.post(self._full_path(self.domain, object_name, self.version, path),
headers=headers,
params=params,
data=data,
Expand Down Expand Up @@ -198,7 +211,7 @@ def patch(self, object_name, path=None,

try:
self._check_auth(object_name=object_name)
request = requests.patch(self._full_path(object_name, self.version, path),
request = requests.patch(self._full_path(self.domain, object_name, self.version, path),
headers=headers,
params=params,
data=data,
Expand Down Expand Up @@ -234,7 +247,7 @@ def get(self, object_name, path=None, params=None, retries=0, **kwargs):
headers = self._build_auth_header()
try:
self._check_auth(object_name=object_name)
request = requests.get(self._full_path(object_name, self.version, path), params=params, headers=headers)
request = requests.get(self._full_path(self.domain, object_name, self.version, path), params=params, headers=headers)
response = self._check_response(request)
return response
except PardotAPIError as err:
Expand Down Expand Up @@ -283,9 +296,9 @@ def _handle_expired_token(self, err, retries, method, headers, object_name, path
raise err

@staticmethod
def _full_path(object_name, version, path=None):
def _full_path(domain, object_name, version, path=None):
"""Builds the full path for the API request"""
full = '{0}/api/{1}/version/{2}'.format(BASE_URI, object_name, version)
full = '{0}/api/{1}/version/{2}'.format(domain, object_name, version)
if path:
return full + '{0}'.format(path)
return full
Expand Down