adobe_tools - Add/Remove Federated User (internal)

WP_adobeapi_addfederateduser.py

def _add_federated_user(
  config_data, access_token, email, country, firstname, lastname
):
  """Add user to domain."""
  add_dict = {
    'user': email,
    'do': [
      {
        'createFederatedID': {
          'email': email,
          'country': country,
          'firstname': firstname,
          'lastname': lastname,
        }
      }
    ]
  }
  body = json.dumps([add_dict])
  url = "https://" + config_data['host'] + config_data['endpoint'] + \
        "/action/" + config_data['org_id']
  res = requests.post(
    url,
    headers=headers(config_data, access_token),
    data=body
  )
  if res.status_code != 200:
    print(res.status_code)
    print(res.headers)
    print(res.text)
  else:
    results = json.loads(res.text)
    if results.get('notCompleted') == 1:
      print("Not completed!")
      print(results.get('errors'))
      return False
    if results.get('completed') == 1:
      print("Completed!")
      return True

    
def _remove_user_from_org(config_data, access_token, user):
  """Remove user from organization."""
  add_dict = {
    'user': user,
    'do': [
      {
        'removeFromOrg': {}
      }
    ]
  }
  body = json.dumps([add_dict])
  url = "https://" + config_data['host'] + config_data['endpoint'] + \
        "/action/" + config_data['org_id']
  res = requests.post(
    url,
    headers=headers(config_data, access_token),
    data=body
  )
  if res.status_code != 200:
    print(res.status_code)
    print(res.headers)
    print(res.text)
  else:
    results = json.loads(res.text)
    if results.get('notCompleted') == 1:
      print("Not completed!")
      print(results.get('errors'))
      return False
    if results.get('completed') == 1:
      print("Completed!")
      return True