Update registry rest client to support bearer authentication

Author: gtato

src/confluent_kafka/schema_registry/schema_registry_client.py

@@ -97,21 +97,28 @@ def __init__(self, conf):
  " configuring ssl.key.location")
 
  userinfo = utils.get_auth_from_url(base_url)
- if 'basic.auth.user.info' in conf_copy:
- if userinfo != ('', ''):
- raise ValueError("basic.auth.user.info configured with"
- " userinfo credentials in the URL."
- " Remove userinfo credentials from the url or"
- " remove basic.auth.user.info from the"
- " configuration")
-
+ url_basic_auth = userinfo != ('', '')
+ config_basic_auth = 'basic.auth.user.info' in conf_copy
+ config_bearer_auth = 'token' in conf_copy
+ if sum([url_basic_auth, config_basic_auth, config_bearer_auth]) > 1:
+ raise ValueError("credentials are specified more than once,"
+ " or multiple authentication mechanisms are configured."
+ " Please specify only one of the following options:"
+ " (1) user credentials on the url"
+ " (2) user credentials using 'basic.auth.user.info' configuration"
+ " (3) token using 'token' configuration.")
+
+ if url_basic_auth:
+ self.session.auth = userinfo
+ elif config_basic_auth:
  userinfo = tuple(conf_copy.pop('basic.auth.user.info', '').split(':'))
-
  if len(userinfo) != 2:
  raise ValueError("basic.auth.user.info must be in the form"
  " of {username}:{password}")
-
- self.session.auth = userinfo if userinfo != ('', '') else None
+ self.session.auth = userinfo
+ elif config_bearer_auth:
+ token = conf_copy.pop('token')
+ self.session.headers['Authorization'] = f'Bearer {token}'
 
  # Any leftover keys are unknown to _RestClient
  if len(conf_copy) > 0: