"""Common utils"""
import collections
import json
import os
import re
import sys
import zipfile
from datetime import datetime
from typing import Any, Dict, Tuple, Union
from bson import ObjectId
[docs]SearchParameter = collections.namedtuple('SearchParameter', ['key', 'value'])
[docs]SEARCH_RGX = re.compile(r'[^\s]+:[^\s]+')
[docs]def to_object_id(_id: Union[ObjectId, str, None]) -> ObjectId:
"""Create ObjectId based on str, or return original value."""
if not isinstance(_id, ObjectId):
assert isinstance(_id, str)
_id = ObjectId(_id)
return _id
[docs]class JSONEncoder(json.JSONEncoder):
"""Handles some of the built in types"""
[docs] def default(self, o):
if isinstance(o, ObjectId):
return str(o)
if isinstance(o, datetime):
return str(o)
return json.JSONEncoder.default(self, o)
[docs]def zipdir(path: str, zf: zipfile.ZipFile):
"""Walk in zip file"""
for root, _, files in os.walk(path):
for file in files:
arcname = os.path.relpath(os.path.join(root, file), os.path.join(path))
zf.write(os.path.join(root, file), arcname)
[docs]def parse_search_string(search_string: str) -> Tuple[Dict, str]:
"""Separate keywords fro mserach string"""
found_matches = re.findall(SEARCH_RGX, search_string)
search_parameters = dict([match.split(':') for match in found_matches])
return search_parameters, ' '.join(re.sub(SEARCH_RGX, '', search_string).split())
[docs]def query_yes_no(question: str, default: str = 'yes') -> bool:
"""Ask a yes/no question via input() and return their answer.
Args:
question (str): String that is presented to the user.
default (str): 'yes' or 'no' default value
The 'answer' return value is True for 'yes' or False for 'no'.
"""
valid = {'yes': True, 'y': True, 'ye': True,
'no': False, 'n': False}
if default is None:
prompt = ' [y/n] '
elif default == 'yes':
prompt = ' [Y/n] '
elif default == 'no':
prompt = ' [y/N] '
else:
raise ValueError(f"Invalid default answer: `{default}`")
while True:
sys.stdout.write(question + prompt)
choice = input().lower()
if default is not None and choice == '':
return valid[default]
elif choice in valid:
return valid[choice]
else:
sys.stdout.write("Please respond with 'yes' or 'no' "
"(or 'y' or 'n').\n")
[docs]def str_to_bool(val: Union[str, bool]) -> bool:
"""Convert string value to boolean"""
if isinstance(val, bool):
return val
if val.lower() in TRUES:
return True
if val.lower() in FALSES:
return False
raise ValueError(f"Invalid value: `{val}`. Expected boolean or string-boolean i.e. `true`")
[docs]def update_dict_recursively(dest: Dict[Any, Any], donor: Dict[Any, Any]) -> Dict[Any, Any]:
"""Update dictionary in place"""
for k, v in donor.items():
if isinstance(v, collections.Mapping):
dest[k] = update_dict_recursively(dest.get(k, {}), v) # type: ignore
else:
dest[k] = v
return dest