range(0x00, 0x09),

range(0x0A, 0x0A),

range(0x0B, 0x0D),

range(0x0E, 0x20),

range(0xD800, 0xE000),

range(0xFFFE, 0x10000),

"""Helper class to convert JSON documents with datetime, decimals, or bytes."""

def default(self, o):

import yaml # leave import here, to avoid breaking our Lambda tests!

if isinstance(o, decimal.Decimal):

if o % 1 > 0:

return float(o)

else:

return int(o)

if isinstance(o, (datetime, date)):

return timestamp_millis(o)

if isinstance(o, yaml.ScalarNode):

if o.tag == "tag:yaml.org,2002:int":

return int(o.value)

if o.tag == "tag:yaml.org,2002:float":

return float(o.value)

if o.tag == "tag:yaml.org,2002:bool":

return bool(o.value)

return str(o.value)

try:

if isinstance(o, bytes):

return to_str(o)

return super(CustomEncoder, self).default(o)

except Exception:

"""Helper class to run a shell command in a background thread."""

def __init__(

self,

cmd: Union[str, List[str]],

params: Any = None,

outfile: Union[str, int] = None,

env_vars: Dict[str, str] = None,

stdin: bool = False,

auto_restart: bool = False,

quiet: bool = True,

inherit_cwd: bool = False,

inherit_env: bool = True,

log_listener: Callable = None,

stop_listener: Callable = None,

strip_color: bool = False,

):

params = not_none_or(params, {})

env_vars = not_none_or(env_vars, {})

self.stopped = False

self.cmd = cmd

self.process = None

self.outfile = outfile

self.stdin = stdin

self.env_vars = env_vars

self.inherit_cwd = inherit_cwd

self.inherit_env = inherit_env

self.auto_restart = auto_restart

self.log_listener = log_listener

self.stop_listener = stop_listener

self.strip_color = strip_color

self.started = threading.Event()

FuncThread.__init__(self, self.run_cmd, params, quiet=quiet)

def run_cmd(self, params):

while True:

self.do_run_cmd()

if (

INFRA_STOPPED

or not self.auto_restart

or not self.process

or self.process.returncode == 0

):

return self.process.returncode if self.process else None

LOG.info(

"Restarting process (received exit code %s): %s", self.process.returncode, self.cmd

)

def do_run_cmd(self):

def convert_line(line):

line = to_str(line or "")

if self.strip_color:

# strip color codes

line = re.sub(r"\x1b(\[.*?[@-~]|\].*?(\x07|\x1b\\))", "", line)

return "%s\r\n" % line.strip()

def filter_line(line):

"""Return True if this line should be filtered, i.e., not printed"""

return "(Press CTRL+C to quit)" in line

outfile = self.outfile or os.devnull

if self.log_listener and outfile == os.devnull:

outfile = subprocess.PIPE

try:

self.process = run(

self.cmd,

asynchronous=True,

stdin=self.stdin,

outfile=outfile,

env_vars=self.env_vars,

inherit_cwd=self.inherit_cwd,

inherit_env=self.inherit_env,

)

self.started.set()

if outfile:

if outfile == subprocess.PIPE:

# get stdout/stderr from child process and write to parent output

streams = (

(self.process.stdout, sys.stdout),

(self.process.stderr, sys.stderr),

)

for instream, outstream in streams:

if not instream:

continue

for line in iter(instream.readline, None):

# `line` should contain a newline at the end as we're iterating,

# hence we can safely break the loop if `line` is None or empty string

if line in [None, "", b""]:

break

if not (line and line.strip()) and self.is_killed():

break

line = convert_line(line)

if filter_line(line):

continue

if self.log_listener:

self.log_listener(line, stream=instream)

if self.outfile not in [None, os.devnull]:

outstream.write(line)

outstream.flush()

if self.process:

self.process.wait()

else:

self.process.communicate()

except Exception as e:

self.result_future.set_exception(e)

if self.process and not self.quiet:

LOG.warning('Shell command error "%s": %s', e, self.cmd)

if self.process and not self.quiet and self.process.returncode != 0:

LOG.warning('Shell command exit code "%s": %s', self.process.returncode, self.cmd)

def is_killed(self):

if not self.process:

return True

if INFRA_STOPPED:

return True

# Note: Do NOT import "psutil" at the root scope, as this leads

# to problems when importing this file from our test Lambdas in Docker

# (Error: libc.musl-x86_64.so.1: cannot open shared object file)

import psutil

return not psutil.pid_exists(self.process.pid)

def stop(self, quiet=False):

if self.stopped:

return

if not self.process:

LOG.warning("No process found for command '%s'", self.cmd)

return

parent_pid = self.process.pid

try:

kill_process_tree(parent_pid)

self.process = None

except Exception as e:

if not quiet:

LOG.warning("Unable to kill process with pid %s: %s", parent_pid, e)

try:

self.stop_listener and self.stop_listener(self)

except Exception as e:

if not quiet:

LOG.warning("Unable to run stop handler for shell command thread %s: %s", self, e)

"""Generic JSON serializable object for simplified subclassing"""

def to_json(self, indent=None):

return json.dumps(

self,

default=lambda o: (

(float(o) if o % 1 > 0 else int(o))

if isinstance(o, decimal.Decimal)

else o.__dict__

),

sort_keys=True,

indent=indent,

)

def apply_json(self, j):

if isinstance(j, str):

j = json.loads(j)

self.__dict__.update(j)

def to_dict(self):

return json.loads(self.to_json())

@classmethod

def from_json(cls, j):

j = JsonObject.as_dict(j)

result = cls()

result.apply_json(j)

return result

@classmethod

def from_json_list(cls, json_list):

return [cls.from_json(j) for j in json_list]

@classmethod

def as_dict(cls, obj):

if isinstance(obj, dict):

return obj

return obj.to_dict()

def __str__(self):

return self.to_json()

def __repr__(self):

"""Useful when applying jsonpatch. Use it as follows:

def __delitem__(self, key, *args, **kwargs):

"""A context manager that captures stdout/stderr of the current thread. Use it as follows:

orig_stdout = sys.stdout

orig_stderr = sys.stderr

orig___stdout = sys.__stdout__

orig___stderr = sys.__stderr__

CONTEXTS_BY_THREAD = {}

class LogStreamIO(io.StringIO):

def write(self, s):

if isinstance(s, str) and hasattr(s, "decode"):

s = s.decode("unicode-escape")

return super(CaptureOutput.LogStreamIO, self).write(s)

def __init__(self):

self._stdout = self.LogStreamIO()

self._stderr = self.LogStreamIO()

def __enter__(self):

# Note: import werkzeug here (not at top of file) to allow dependency pruning

from werkzeug.local import LocalProxy

ident = self._ident()

if ident not in self.CONTEXTS_BY_THREAD:

self.CONTEXTS_BY_THREAD[ident] = self

self._set(

LocalProxy(self._proxy(sys.stdout, "stdout")),

LocalProxy(self._proxy(sys.stderr, "stderr")),

LocalProxy(self._proxy(sys.__stdout__, "stdout")),

LocalProxy(self._proxy(sys.__stderr__, "stderr")),

)

return self

def __exit__(self, type, value, traceback):

ident = self._ident()

removed = self.CONTEXTS_BY_THREAD.pop(ident, None)

if not self.CONTEXTS_BY_THREAD:

# reset pointers

self._set(

self.orig_stdout,

self.orig_stderr,

self.orig___stdout,

self.orig___stderr,

)

# get value from streams

removed._stdout.flush()

removed._stderr.flush()

out = removed._stdout.getvalue()

err = removed._stderr.getvalue()

# close handles

removed._stdout.close()

removed._stderr.close()

removed._stdout = out

removed._stderr = err

def _set(self, out, err, __out, __err):

sys.stdout, sys.stderr, sys.__stdout__, sys.__stderr__ = (

out,

err,

__out,

__err,

)

def _proxy(self, original_stream, type):

def proxy():

ident = self._ident()

ctx = self.CONTEXTS_BY_THREAD.get(ident)

if ctx:

return ctx._stdout if type == "stdout" else ctx._stderr

return original_stream

return proxy

def _ident(self):

# TODO: On some systems we seem to be running into a stack overflow with LAMBDA_EXECUTOR=local here!

return threading.current_thread().ident

def stdout(self):

return self._stream_value(self._stdout)

def stderr(self):

return self._stream_value(self._stderr)

def _stream_value(self, stream):

"""Simple wrapper class that allows us to create a hashset using the object id(..) as the entries' hash value"""

def __init__(self, obj):

self.obj = obj

self._hash = id(obj)

def __hash__(self):

return self._hash

def __eq__(self, other):

# assumption here is that we're comparing only against ObjectIdHash instances!

"""Dummy object that can be arbitrarily accessed - any attributes, as a callable, item assignment, ..."""

def __init__(self, name=None):

self.name = name

def __getattr__(self, name, *args, **kwargs):

return ArbitraryAccessObj(name)

def __call__(self, *args, **kwargs):

if self.name in ["items", "keys", "values"] and not args and not kwargs:

return []

return ArbitraryAccessObj()

def __getitem__(self, *args, **kwargs):

return ArbitraryAccessObj()

def __setitem__(self, *args, **kwargs):

"""Hashable list class that can be used with dicts or hashsets."""

def __hash__(self):

result = 0

for i in self:

result += hash(i)

"""List which can be paginated and filtered. For usage in AWS APIs with paginated responses"""

DEFAULT_PAGE_SIZE = 50

def get_page(

self,

token_generator: Callable,

next_token: str = None,

page_size: int = None,

filter_function: Callable = None,

) -> (list, str):

if filter_function is not None:

result_list = list(filter(filter_function, self))

else:

result_list = self

if page_size is None:

page_size = self.DEFAULT_PAGE_SIZE

if len(result_list) <= page_size:

return result_list, None

start_idx = 0

try:

start_item = next(item for item in result_list if token_generator(item) == next_token)

start_idx = result_list.index(start_item)

except StopIteration:

pass

if start_idx + page_size <= len(result_list):

next_token = token_generator(result_list[start_idx + page_size])

else:

next_token = None

"""A dictionary that is mapped to a json document on disk.

path: Union[str, os.PathLike]

def __init__(self, path: Union[str, os.PathLike], mode=0o664):

super().__init__()

self.path = path

self.mode = mode

self.load()

def load(self):

if not os.path.exists(self.path):

return

if os.path.isdir(self.path):

raise IsADirectoryError

with open(self.path, "r") as fd:

self.update(json.load(fd))

def save(self):

if os.path.isdir(self.path):

raise IsADirectoryError

if not os.path.exists(self.path):

mkdir(os.path.dirname(self.path))

def opener(path, flags):

_fd = os.open(path, flags, self.mode)

os.chmod(path, mode=self.mode, follow_symlinks=True)

return _fd

with open(self.path, "w", opener=opener) as fd:

"""Manages the ports used for starting external services like ElasticSearch, OpenSearch,..."""

def __init__(self):

# cache for locally available ports (ports are reserved for a short period of a few seconds)

self._PORTS_CACHE = cachetools.TTLCache(maxsize=100, ttl=6)

self._PORTS_LOCK = threading.RLock()

def reserve_port(self, port: int = None) -> int:

"""

ports_range = range(config.EXTERNAL_SERVICE_PORTS_START, config.EXTERNAL_SERVICE_PORTS_END)

if port is not None and port not in ports_range:

raise PortNotAvailableException(

f"The requested port ({port}) is not in the configured external "

f"service port range ({ports_range})."

)

with self._PORTS_LOCK:

if port is not None:

return self._check_port(port)

else:

for port_in_range in ports_range:

try:

return self._check_port(port_in_range)

except PortNotAvailableException:

# We ignore the fact that this single port is reserved, we just check the next one

pass

raise PortNotAvailableException(

"No free network ports available to start service instance (currently reserved: %s)",

list(self._PORTS_CACHE.keys()),

)

def _check_port(self, port: int) -> int:

"""Checks if the given port is currently not reserved and can be bound."""

if not self._PORTS_CACHE.get(port) and port_can_be_bound(port):

# reserve the port for a short period of time

self._PORTS_CACHE[port] = "__reserved__"

return port

else:

"""Start the given method in a background thread, and add the thread to the TMP_THREADS shutdown hook"""

_shutdown_hook = kwargs.pop("_shutdown_hook", True)

thread = FuncThread(method, *args, **kwargs)

thread.start()

if _shutdown_hook:

TMP_THREADS.append(thread)

"""

def _decorator(wrapped):

@functools.wraps(wrapped)

def _wrapper(*args, **kwargs):

with lock:

return wrapped(*args, **kwargs)

return _wrapper

"""Function decorator to protect a function from stack overflows -

def _decorator(wrapped):

@functools.wraps(wrapped)

def func(*args, **kwargs):

def _matches(frame):

if frame.function != wrapped.__name__:

return False

frame = frame.frame

if not match_parameters:

return False

# construct dict of arguments this stack frame has been called with

prev_call_args = {

frame.f_code.co_varnames[i]: frame.f_locals[frame.f_code.co_varnames[i]]

for i in range(frame.f_code.co_argcount)

}

# construct dict of arguments the original function has been called with

sig = inspect.signature(wrapped)

this_call_args = dict(zip(sig.parameters.keys(), args))

this_call_args.update(kwargs)

return prev_call_args == this_call_args

matching_frames = [frame[2] for frame in inspect.stack(context=1) if _matches(frame)]

if matching_frames:

raise RecursionError("(Potential) infinite recursion detected")

return wrapped(*args, **kwargs)

return func

if isinstance(s, bytes) and exclude_binary:

return False

if isinstance(s, str):

return True

if include_unicode and isinstance(s, str):

return True

"""Remove a set of attributes from the given dict (in-place)"""

if recursive:

def _remove(o, **kwargs):

if isinstance(o, dict):

remove_attributes(o, attributes)

return o

return recurse_object(obj, _remove)

attributes = attributes if is_list_or_tuple(attributes) else [attributes]

for attr in attributes:

obj.pop(attr, None)

obj: Dict, old_to_new_attributes: Dict[str, str], in_place: bool = False

) -> Dict:

"""Rename a set of attributes in the given dict object. Second parameter is a dict that maps old to

if not in_place:

obj = dict(obj)

for old_name, new_name in old_to_new_attributes.items():

if old_name in obj:

obj[new_name] = obj.pop(old_name)

"""Wrap the given object in a list, or return the object itself if it already is a list."""

if obj is None and not wrap_none:

return obj

if config.EDGE_PORT_HTTP:

result = "ports %s/%s" % (config.EDGE_PORT, config.EDGE_PORT_HTTP)

else:

result = "port %s" % config.EDGE_PORT

result = "%s %s" % (get_service_protocol(), result)

"""Return a list of unique items from the given input iterable.

def contained(item):

for r in result:

if comparator:

cmp_res = comparator(item, r)

if cmp_res is True or str(cmp_res) == "0":

return True

elif item == r:

return True

result = []

for it in inputs:

if not contained(it):

result.append(it)

if not time:

time = datetime.utcnow()

if isinstance(time, (int, float)):

time = datetime.fromtimestamp(time)

for ts_format in [TIMESTAMP_FORMAT, TIMESTAMP_FORMAT_TZ, TIMESTAMP_FORMAT_MICROS]:

try:

return datetime.strptime(ts_str, ts_format)

except ValueError:

pass

if none_values is None:

none_values = [None]

for key, value in source.items():

if isinstance(value, dict):

# get node or create one

node = destination.setdefault(key, {})

merge_recursive(value, node, none_values=none_values, overwrite=overwrite)

else:

if not isinstance(destination, (dict, CaseInsensitiveDict)):

LOG.warning(

"Destination for merging %s=%s is not dict: %s (%s)",

key,

value,

destination,

type(destination),

)

if overwrite or destination.get(key) in none_values:

destination[key] = value

"""Merge all dicts in `*dicts` into a single dict, and return the result. If any of the entries

result = {}

for d in dicts:

if d is None and "default" in kwargs:

return kwargs["default"]

if d:

result.update(d)

"""Recursively apply `func` to `obj` (may be a list, dict, or other object)."""

obj = func(obj, path=path)

if isinstance(obj, list):

for i in range(len(obj)):

tmp_path = "%s[%s]" % (path or ".", i)

obj[i] = recurse_object(obj[i], func, tmp_path)

elif isinstance(obj, dict):

for k, v in obj.items():

tmp_path = "%s%s" % ((path + ".") if path else "", k)

obj[k] = recurse_object(v, func, tmp_path)

"""Recursively changes all dict keys to first character lowercase. Skip children

skip_children_of = ensure_list(skip_children_of or [])

def fix_keys(o, path="", **kwargs):

if any(re.match(r"(^|.*\.)%s($|[.\[].*)" % k, path) for k in skip_children_of):

return o

if isinstance(o, dict):

for k, v in dict(o).items():

o.pop(k)

o[first_char_to_lower(k)] = v

return o

result = recurse_object(obj, fix_keys)

components = string.split("_")

start_idx = 0 if capitalize_first else 1

components = [x.title() for x in components[start_idx:]]

"""Return an XML representation of the given object (dict, list, or primitive).

if isinstance(obj, list):

return "".join([obj_to_xml(o) for o in obj])

if isinstance(obj, dict):

return "".join(["<{k}>{v}</{k}>".format(k=k, v=obj_to_xml(v)) for (k, v) in obj.items()])

"""Strip xmlns attributes from a dict returned by xmltodict.parse."""

if isinstance(obj, list):

return [strip_xmlns(item) for item in obj]

if isinstance(obj, dict):

# Remove xmlns attribute.

obj.pop("@xmlns", None)

if len(obj) == 1 and "#text" in obj:

# If the only remaining key is the #text key, elide the dict

# entirely, to match the structure that xmltodict.parse would have

# returned if the xmlns namespace hadn't been present.

return obj["#text"]

return {k: strip_xmlns(v) for k, v in obj.items()}

"""Return whether the given directory contains any entries (files/folders), including hidden

if not os.path.isdir(directory):

raise Exception(f"Path is not a directory: {directory}")

entries = os.listdir(directory)