From 366da05cbaeca84229227a64b657d5e41f27bd15 Mon Sep 17 00:00:00 2001
From: A_D <aunderscored@gmail.com>
Date: Wed, 7 Jul 2021 16:27:18 +0200
Subject: [PATCH] added utility methods for using the new actions
---
killswitch.py | 56 ++++++++++++++++++++++++++++++++++++++++++---------
1 file changed, 47 insertions(+), 9 deletions(-)
diff --git a/killswitch.py b/killswitch.py
index d0b6c09e..6a0cc9d5 100644
--- a/killswitch.py
+++ b/killswitch.py
@@ -3,7 +3,8 @@ from __future__ import annotations
from copy import deepcopy
from typing import (
- Any, Dict, List, Mapping, MutableMapping, MutableSequence, NamedTuple, Optional, Sequence, TypedDict, Union, cast
+ Any, Dict, List, Mapping, MutableMapping, MutableSequence, NamedTuple, Optional, Sequence, Tuple, TypedDict,
+ TypeVar, Union, cast
)
import requests
@@ -18,7 +19,7 @@ logger = EDMCLogging.get_main_logger()
OLD_KILLSWITCH_URL = 'https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/killswitches.json'
DEFAULT_KILLSWITCH_URL = 'https://raw.githubusercontent.com/EDCD/EDMarketConnector/releases/killswitches_v2.json'
CURRENT_KILLSWITCH_VERSION = 2
-
+UPDATABLE_DATA = Union[Mapping, Sequence]
_current_version: semantic_version.Version = config.appversion_nobuild()
@@ -36,7 +37,7 @@ class SingleKill(NamedTuple):
"""Return whether or not this SingleKill can apply rules to a dict to make it safe to use."""
return any(x is not None for x in (self.redact_fields, self.delete_fields, self.set_fields))
- def apply_rules(self, target: Dict[str, Any]):
+ def apply_rules(self, target: UPDATABLE_DATA) -> UPDATABLE_DATA:
"""
Apply the rules this SingleKill instance has to make some data okay to send.
@@ -56,8 +57,10 @@ class SingleKill(NamedTuple):
for key, value in (self.set_fields if self .set_fields is not None else {}).items():
_deep_apply(target, key, value)
+ return target
-def _apply(target: Union[MutableMapping, MutableSequence], key: str, to_set: Any = None, delete: bool = False):
+
+def _apply(target: UPDATABLE_DATA, key: str, to_set: Any = None, delete: bool = False):
"""
Set or delete the given target key on the given target.
@@ -74,7 +77,8 @@ def _apply(target: Union[MutableMapping, MutableSequence], key: str, to_set: Any
target[key] = to_set
elif isinstance(target, MutableSequence):
- if (idx := _get_int(key)) is None:
+ idx = _get_int(key)
+ if idx is None:
raise ValueError(f'Cannot use string {key!r} as int for index into Sequence')
if delete:
@@ -91,7 +95,7 @@ def _apply(target: Union[MutableMapping, MutableSequence], key: str, to_set: Any
raise ValueError(f'Dont know how to apply data to {type(target)} {target!r}')
-def _deep_apply(target: dict[str, Any], path: str, to_set=None, delete=False):
+def _deep_apply(target: UPDATABLE_DATA, path: str, to_set=None, delete=False):
"""
Set the given path to the given value, if it exists.
@@ -102,7 +106,7 @@ def _deep_apply(target: dict[str, Any], path: str, to_set=None, delete=False):
:param to_set: the data to set, defaults to None
:param delete: whether or not to delete the key rather than set it
"""
- current: Union[MutableMapping, MutableSequence] = target
+ current = target
key: str = ""
while '.' in path:
if path in current:
@@ -112,10 +116,11 @@ def _deep_apply(target: dict[str, Any], path: str, to_set=None, delete=False):
key, _, path = path.partition('.')
if isinstance(current, Mapping):
- current = current[key]
+ current = current[key] # type: ignore # I really dont know at this point what you want from me mypy.
elif isinstance(current, Sequence):
- if (target_idx := _get_int(key)) is not None:
+ target_idx = _get_int(key) # mypy is broken. doesn't like := here.
+ if target_idx is not None:
current = current[target_idx]
else:
raise ValueError(f'Cannot index sequence with non-int key {key!r}')
@@ -207,6 +212,34 @@ class KillSwitchSet:
"""
return [k for k in self.kill_switches if version in k.version]
+ def check_killswitch(self, name: str, data: UPDATABLE_DATA, log=logger) -> Tuple[bool, UPDATABLE_DATA]:
+ """
+ Check whether or not a killswitch is enabled. If it is, apply rules if any.
+
+ :param name: The killswitch to check
+ :param data: The data to modify if needed
+ :return: A bool indicating if the caller should return, and either the original data or a *COPY* that has
+ been modified by rules
+ """
+ res = self.get_disabled(name)
+ if not res.disabled:
+ return False, data
+
+ log.info(f'Killswitch {name} is enabled. Checking if rules exist to make use safe')
+ if res.kill is None or not res.kill.has_rules:
+ logger.info('No rules exist. Stopping processing')
+ return True, data
+
+ try:
+ new_data = res.kill.apply_rules(deepcopy(data))
+
+ except Exception as e:
+ log.exception(f'Exception occurred while attempting to apply rules! bailing out! {e=}')
+ return True, data
+
+ log.info('Rules applied successfully, allowing execution to continue')
+ return False, new_data
+
def __str__(self) -> str:
"""Return a string representation of KillSwitchSet."""
return f'KillSwitchSet: {str(self.kill_switches)}'
@@ -374,6 +407,11 @@ def get_disabled(id: str, *, version: semantic_version.Version = _current_versio
return active.get_disabled(id, version=version)
+def check_killswitch(name: str, data: UPDATABLE_DATA, log=logger) -> Tuple[bool, UPDATABLE_DATA]:
+ """Query the global KillSwitchSet#check_killswitch method."""
+ return active.check_killswitch(name, data, log)
+
+
def is_disabled(id: str, *, version: semantic_version.Version = _current_version) -> bool:
"""Query the global KillSwitchSet#is_disabled method."""
return active.is_disabled(id, version=version)