Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/azure/core/utils/_connection_string_parser.py: 11%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

18 statements  

1# coding=utf-8 

2# ------------------------------------------------------------------------- 

3# Copyright (c) Microsoft Corporation. All rights reserved. 

4# Licensed under the MIT License. See License.txt in the project root for 

5# license information. 

6# -------------------------------------------------------------------------- 

7from typing import Mapping 

8 

9 

10def parse_connection_string(conn_str: str, case_sensitive_keys: bool = False) -> Mapping[str, str]: 

11 """Parses the connection string into a dict of its component parts, with the option of preserving case 

12 of keys, and validates that each key in the connection string has a provided value. If case of keys 

13 is not preserved (ie. `case_sensitive_keys=False`), then a dict with LOWERCASE KEYS will be returned. 

14 

15 :param str conn_str: String with connection details provided by Azure services. 

16 :param bool case_sensitive_keys: Indicates whether the casing of the keys will be preserved. When `False`(the 

17 default), all keys will be lower-cased. If set to `True`, the original casing of the keys will be preserved. 

18 :rtype: Mapping 

19 :returns: Dict of connection string key/value pairs. 

20 :raises ValueError: if each key in conn_str does not have a corresponding value and 

21 for other bad formatting of connection strings - including duplicate 

22 args, bad syntax, etc. 

23 """ 

24 

25 cs_args = [s.split("=", 1) for s in conn_str.strip().rstrip(";").split(";")] 

26 if any(len(tup) != 2 or not all(tup) for tup in cs_args): 

27 raise ValueError("Connection string is either blank or malformed.") 

28 args_dict = dict(cs_args) 

29 

30 if len(cs_args) != len(args_dict): 

31 raise ValueError("Connection string is either blank or malformed.") 

32 

33 if not case_sensitive_keys: 

34 # if duplicate case insensitive keys are passed in, raise error 

35 new_args_dict = {} 

36 for key in args_dict.keys(): # pylint: disable=consider-using-dict-items 

37 new_key = key.lower() 

38 if new_key in new_args_dict: 

39 raise ValueError("Duplicate key in connection string: {}".format(new_key)) 

40 new_args_dict[new_key] = args_dict[key] 

41 return new_args_dict 

42 

43 return args_dict