pyupgrade: Replace Text with str.

We uses `pyupgrade --py3-plus` to automatically replace all occurence
of `Text`. But manual fix is required to remove the unused imports. Note
that with this configuration pyupgrade also convert string literals to
.format(...) style, which is manually not included in the commit as well.
This commit is contained in:
PIG208 2021-05-28 19:13:13 +08:00 committed by Tim Abbott
parent a54cccc012
commit e27ac0ddbe
11 changed files with 42 additions and 46 deletions

View file

@ -3,12 +3,11 @@ for extract.py
"""
from json.decoder import JSONDecodeError
from typing import Text
from zulip_bots.bots.monkeytestit.lib import extract, report
def execute(message: Text, apikey: Text) -> Text:
def execute(message: str, apikey: str) -> str:
"""Parses message and returns a dictionary
:param message: The message
@ -69,7 +68,7 @@ def execute(message: Text, apikey: Text) -> Text:
return "Unknown command. Available commands: `check <website> " "[params]`"
def failed(message: Text) -> Text:
def failed(message: str) -> str:
"""Simply attaches a failed marker to a message
:param message: The message

View file

@ -1,10 +1,10 @@
"""Used to mainly compose a decorated report for the user
"""
from typing import Dict, List, Text
from typing import Dict, List
def compose(results: Dict) -> Text:
def compose(results: Dict) -> str:
"""Composes a report based on test results
An example would be:
@ -38,7 +38,7 @@ def compose(results: Dict) -> Text:
return response
def print_more_info_url(results: Dict) -> Text:
def print_more_info_url(results: Dict) -> str:
"""Creates info for the test URL from monkeytest.it
Example:
@ -51,7 +51,7 @@ def print_more_info_url(results: Dict) -> Text:
return "More info: {}".format(results["results_url"])
def print_test_id(results: Dict) -> Text:
def print_test_id(results: Dict) -> str:
"""Prints the test-id with attached to the url
:param results: A dictionary containing the results of a check
@ -60,7 +60,7 @@ def print_test_id(results: Dict) -> Text:
return "Test: https://monkeytest.it/test/{}".format(results["test_id"])
def print_failures_checkers(results: Dict) -> Text:
def print_failures_checkers(results: Dict) -> str:
"""Creates info for failures in enabled checkers
Example:
@ -105,7 +105,7 @@ def get_enabled_checkers(results: Dict) -> List:
return enabled_checkers
def print_enabled_checkers(results: Dict) -> Text:
def print_enabled_checkers(results: Dict) -> str:
"""Creates info for enabled checkers. This joins the list of enabled
checkers and format it with the current string response
@ -118,7 +118,7 @@ def print_enabled_checkers(results: Dict) -> Text:
return "Enabled checkers: {}".format(", ".join(get_enabled_checkers(results)))
def print_status(results: Dict) -> Text:
def print_status(results: Dict) -> str:
"""Creates info for the check status.
Example: Status: tests_failed

View file

@ -3,12 +3,12 @@ import importlib.abc
import importlib.util
import os
from pathlib import Path
from typing import Any, Optional, Text, Tuple
from typing import Any, Optional, Tuple
current_dir = os.path.dirname(os.path.abspath(__file__))
def import_module_from_source(path: Text, name: Text) -> Any:
def import_module_from_source(path: str, name: str) -> Any:
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
loader = spec.loader
@ -18,14 +18,14 @@ def import_module_from_source(path: Text, name: Text) -> Any:
return module
def import_module_by_name(name: Text) -> Any:
def import_module_by_name(name: str) -> Any:
try:
return importlib.import_module(name)
except ImportError:
return None
def resolve_bot_path(name: Text) -> Optional[Tuple[Path, Text]]:
def resolve_bot_path(name: str) -> Optional[Tuple[Path, str]]:
if os.path.isfile(name):
bot_path = Path(name)
bot_name = Path(bot_path).stem

View file

@ -7,7 +7,7 @@ import signal
import sys
import time
from contextlib import contextmanager
from typing import IO, Any, Dict, Iterator, List, Optional, Set, Text
from typing import IO, Any, Dict, Iterator, List, Optional, Set
from typing_extensions import Protocol
@ -79,13 +79,13 @@ class BotIdentity:
class BotStorage(Protocol):
def put(self, key: Text, value: Any) -> None:
def put(self, key: str, value: Any) -> None:
...
def get(self, key: Text) -> Any:
def get(self, key: str) -> Any:
...
def contains(self, key: Text) -> bool:
def contains(self, key: str) -> bool:
...
@ -100,13 +100,13 @@ class CachedStorage:
self._cache = init_data
self._dirty_keys: Set[str] = set()
def put(self, key: Text, value: Any) -> None:
def put(self, key: str, value: Any) -> None:
# In the cached storage, values being put to the storage is not flushed to the parent storage.
# It will be marked dirty until it get flushed.
self._cache[key] = value
self._dirty_keys.add(key)
def get(self, key: Text) -> Any:
def get(self, key: str) -> Any:
# Unless the key is not found in the cache, the cached storage will not lookup the parent storage.
if key in self._cache:
return self._cache[key]
@ -123,11 +123,11 @@ class CachedStorage:
key = self._dirty_keys.pop()
self._parent_storage.put(key, self._cache[key])
def flush_one(self, key: Text) -> None:
def flush_one(self, key: str) -> None:
self._dirty_keys.remove(key)
self._parent_storage.put(key, self._cache[key])
def contains(self, key: Text) -> bool:
def contains(self, key: str) -> bool:
if key in self._cache:
return True
else:
@ -139,15 +139,15 @@ class StateHandler:
self._client = client
self.marshal = lambda obj: json.dumps(obj)
self.demarshal = lambda obj: json.loads(obj)
self.state_ = dict() # type: Dict[Text, Any]
self.state_ = dict()
def put(self, key: Text, value: Any) -> None:
def put(self, key: str, value: Any) -> None:
self.state_[key] = self.marshal(value)
response = self._client.update_storage({"storage": {key: self.state_[key]}})
if response["result"] != "success":
raise StateHandlerError("Error updating state: {}".format(str(response)))
def get(self, key: Text) -> Any:
def get(self, key: str) -> Any:
if key in self.state_:
return self.demarshal(self.state_[key])
@ -159,12 +159,12 @@ class StateHandler:
self.state_[key] = marshalled_value
return self.demarshal(marshalled_value)
def contains(self, key: Text) -> bool:
def contains(self, key: str) -> bool:
return key in self.state_
@contextmanager
def use_storage(storage: BotStorage, keys: List[Text]) -> Iterator[BotStorage]:
def use_storage(storage: BotStorage, keys: List[str]) -> Iterator[BotStorage]:
# The context manager for StateHandler that minimizes the number of round-trips to the server.
# It will fetch all the data using the specified keys and store them to
# a CachedStorage that will not communicate with the server until manually