def from_shared_access_key()

in uamqp/authentication/cbs_auth.py [0:0]

• 37 lines of code
• 2 McCabe index (conditional complexity)

    def from_shared_access_key(
            cls,
            uri,
            key_name,
            shared_access_key,
            expiry=None,
            port=None,
            timeout=10,
            retry_policy=TokenRetryPolicy(),
            verify=None,
            http_proxy=None,
            transport_type=TransportType.Amqp,
            encoding='UTF-8',
            **kwargs):
        """Attempt to create a CBS token session using a Shared Access Key such
        as is used to connect to Azure services.

        :param uri: The AMQP endpoint URI. This must be provided as
         a decoded string.
        :type uri: str
        :param key_name: The SAS token username, also referred to as the key
         name or policy name.
        :type key_name: str
        :param shared_access_key: The SAS token password, also referred to as the key.
        :type shared_access_key: str
        :param expiry: The lifetime in seconds for the generated token. Default is 1 hour.
        :type expiry: int
        :param port: The TLS port - default for AMQP is 5671.
        :type port: int
        :param timeout: The timeout in seconds in which to negotiate the token.
         The default value is 10 seconds.
        :type timeout: float
        :param retry_policy: The retry policy for the PUT token request. The default
         retry policy has 3 retries.
        :type retry_policy: ~uamqp.authentication.cbs_auth.TokenRetryPolicy
        :param verify: The path to a user-defined certificate.
        :type verify: str
        :param http_proxy: HTTP proxy configuration. This should be a dictionary with
         the following keys present: 'proxy_hostname' and 'proxy_port'. Additional optional
         keys are 'username' and 'password'.
        :type http_proxy: dict
        :param transport_type: The transport protocol type - default is ~uamqp.TransportType.Amqp.
         ~uamqp.TransportType.AmqpOverWebsocket is applied when http_proxy is set or the
         transport type is explicitly requested.
        :type transport_type: ~uamqp.TransportType
        :param encoding: The encoding to use if hostname is provided as a str.
         Default is 'UTF-8'.
        :type encoding: str
        :keyword int refresh_window: The time in seconds before the token expiration
        time to start the process of token refresh.
        Default value is 10% of the remaining seconds until the token expires.
        """
        expires_in = datetime.timedelta(seconds=expiry or constants.AUTH_EXPIRATION_SECS)
        encoded_uri = compat.quote_plus(uri).encode(encoding)  # pylint: disable=no-member
        encoded_key = compat.quote_plus(key_name).encode(encoding)  # pylint: disable=no-member
        expires_at = time.time() + expires_in.seconds
        token = utils.create_sas_token(
            encoded_key,
            shared_access_key.encode(encoding),
            encoded_uri,
            expires_in)
        return cls(
            uri, uri, token,
            expires_in=expires_in,
            expires_at=expires_at,
            username=key_name,
            password=shared_access_key,
            port=port,
            timeout=timeout,
            retry_policy=retry_policy,
            verify=verify,
            http_proxy=http_proxy,
            transport_type=transport_type,
            encoding=encoding,
            custom_endpoint_hostname=kwargs.pop("custom_endpoint_hostname", None))