Migrate cryptocurrency.py to use pydantic models

Author: linuxdaemon

.gitignore

@@ -65,3 +65,4 @@ config.json
 .vagrant/
 .mypy_cache/
 .DS_Store
+*.bak

.pre-commit-config.yaml

@@ -55,6 +55,12 @@ repos:
     args:
     - "--py38-plus"
 
+- repo: https://github.com/PyCQA/autoflake
+  rev: 0544741e2b4a22b472d9d93e37d4ea9153820bb1     # frozen: v2.3.1
+  hooks:
+  - id: autoflake
+
+
 - repo: local
   hooks:
   - id: mypy

.vscode/settings.json

@@ -14,5 +14,6 @@
     "evenBetterToml.formatter.allowedBlankLines": 1,
     "evenBetterToml.formatter.arrayAutoCollapse": true,
     "evenBetterToml.formatter.arrayAutoExpand": false,
-    "evenBetterToml.formatter.arrayTrailingComma": true
+    "evenBetterToml.formatter.arrayTrailingComma": true,
+    "python.analysis.diagnosticMode": "workspace"
 }

cloudbot/config.py

@@ -4,6 +4,7 @@
 import time
 from collections import OrderedDict
 from pathlib import Path
+from typing import Dict, Optional, cast
 
 logger = logging.getLogger("cloudbot")
 
@@ -15,19 +16,21 @@ def __init__(self, bot, *, filename="config.json"):
         self.path = Path(self.filename).resolve()
         self.bot = bot
 
-        self._api_keys = {}
+        self._api_keys: Dict[str, Optional[str]] = {}
 
         # populate self with config data
         self.load_config()
 
-    def get_api_key(self, name, default=None):
+    def get_api_key(
+        self, name: str, default: Optional[str] = None
+    ) -> Optional[str]:
         try:
             return self._api_keys[name]
         except LookupError:
             self._api_keys[name] = value = self.get("api_keys", {}).get(
                 name, default
             )
-            return value
+            return cast(Optional[str], value)
 
     def load_config(self):
         """(re)loads the bot config from the config file"""

cloudbot/event.py

@@ -2,7 +2,7 @@
 import enum
 import logging
 from functools import partial
-from typing import Any, Iterator, Mapping
+from typing import Any, Iterator, Mapping, Optional
 
 from irclib.parser import Message
 
@@ -245,7 +245,7 @@ def admin_log(self, message, broadcast=False):
             if conn and conn.connected:
                 conn.admin_log(message, console=not broadcast)
 
-    def reply(self, *messages, target=None):
+    def reply(self, *messages: str, target: Optional[str] = None) -> None:
         """sends a message to the current channel/user with a prefix"""
         reply_ping = self.conn.config.get("reply_ping", True)
         if target is None:

cloudbot/hook.py

@@ -4,6 +4,18 @@
 import re
 import warnings
 from enum import Enum, IntEnum, unique
+from typing import (
+    Any,
+    Callable,
+    List,
+    Optional,
+    Sequence,
+    TypeVar,
+    Union,
+    overload,
+)
+
+from typing_extensions import ParamSpec
 
 from cloudbot.event import EventType
 from cloudbot.util import HOOK_ATTR
@@ -186,10 +198,36 @@ def _hook_warn():
     )
 
 
-def command(*args, **kwargs):
+_T = TypeVar("_T")
+_P = ParamSpec("_P")
+_Func = Callable[_P, _T]
+
+
+@overload
+def command(arg: Callable[_P, _T], /) -> Callable[_P, _T]: ...
+
+
+@overload
+def command(
+    arg: Optional[Union[str, Sequence[str]]] = None,
+    /,
+    *args: Union[str, Sequence[str]],
+    **kwargs: Any,
+) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]: ...
+
+
+def command(
+    arg: Optional[Union[Callable[_P, _T], str, Sequence[str]]] = None,
+    /,
+    *args: Union[str, Sequence[str]],
+    **kwargs: Any,
+) -> Union[Callable[_P, _T], Callable[[Callable[_P, _T]], Callable[_P, _T]]]:
     """External command decorator. Can be used directly as a decorator, or with args to return a decorator."""
 
-    def _command_hook(func, alias_param=None):
+    def _command_hook(
+        func: Callable[_P, _T],
+        alias_param: Optional[Sequence[Union[Sequence[str], str]]] = None,
+    ) -> Callable[_P, _T]:
         hook = _get_hook(func, "command")
         if hook is None:
             hook = _CommandHook(func)
@@ -198,13 +236,17 @@ def _command_hook(func, alias_param=None):
         hook.add_hook(alias_param, kwargs)
         return func
 
-    if len(args) == 1 and callable(args[0]):
+    if arg is not None and not isinstance(arg, (str, collections.abc.Sequence)):
         # this decorator is being used directly
         _hook_warn()
-        return _command_hook(args[0])
+        return _command_hook(arg)
+
+    arg_list: List[Union[str, Sequence[str]]] = list(args)
+    if arg:
+        arg_list.insert(0, arg)
 
     # this decorator is being used indirectly, so return a decorator function
-    return lambda func: _command_hook(func, alias_param=args)
+    return lambda func: _command_hook(func, alias_param=arg_list)
 
 
 def irc_raw(triggers_param, **kwargs):
@@ -332,10 +374,22 @@ def _config_hook(func):
     return _config_hook
 
 
-def on_start(param=None, **kwargs):
+@overload
+def on_start(
+    **kwargs: Any,
+) -> Callable[[Callable[_P, _T]], Callable[_P, _T]]: ...
+
+
+@overload
+def on_start(param: Callable[_P, _T], /) -> Callable[_P, _T]: ...
+
+
+def on_start(
+    param: Optional[Callable[_P, _T]] = None, /, **kwargs: Any
+) -> Union[Callable[_P, _T], Callable[[Callable[_P, _T]], Callable[_P, _T]]]:
     """External on_start decorator. Can be used directly as a decorator, or with args to return a decorator"""
 
-    def _on_start_hook(func):
+    def _on_start_hook(func: Callable[_P, _T]) -> Callable[_P, _T]:
         hook = _get_hook(func, "on_start")
         if hook is None:
             hook = _Hook(func, "on_start")

cloudbot/util/colors.py

@@ -154,7 +154,7 @@ def get_available_colours():
     return ret[:-2]
 
 
-def parse(string):
+def parse(string: str) -> str:
     """
     parse: Formats a string, replacing words wrapped in $( ) with actual colours or formatting.
     example:

cloudbot/util/func_utils.py

@@ -1,4 +1,5 @@
 import inspect
+from typing import Any, Callable, Mapping, TypeVar
 
 
 class ParameterError(Exception):
@@ -12,7 +13,14 @@ def __init__(self, name, valid_args):
         self.valid_args = list(valid_args)
 
 
-def call_with_args(func, arg_data):
+_T = TypeVar("_T")
+
+
+def call_with_args(func: Callable[..., _T], arg_data: Mapping[str, Any]) -> _T:
+    """
+    >>> call_with_args(lambda a: a, {'a':1, 'b':2})
+    1
+    """
     sig = inspect.signature(func, follow_wrapped=False)
     try:
         args = [

cloudbot/util/web.py

@@ -17,7 +17,16 @@
 import logging
 import time
 from operator import attrgetter
-from typing import Dict, Optional, Union
+from typing import (
+    Dict,
+    Generic,
+    Iterable,
+    Iterator,
+    Optional,
+    Tuple,
+    TypeVar,
+    Union,
+)
 
 import requests
 from requests import (
@@ -41,76 +50,79 @@
 
 # Public API
 
+_T = TypeVar("_T")
 
-class Registry:
-    class Item:
-        def __init__(self, item):
-            self.item = item
-            self.working = True
-            self.last_check = 0.0
-            self.uses = 0
 
-        def failed(self):
-            self.working = False
-            self.last_check = time.time()
+class RegistryItem(Generic[_T]):
+    def __init__(self, item: _T) -> None:
+        self.item = item
+        self.working = True
+        self.last_check = 0.0
+        self.uses = 0
+
+    def failed(self) -> None:
+        self.working = False
+        self.last_check = time.time()
 
-        @property
-        def should_use(self):
-            if self.working:
-                return True
+    @property
+    def should_use(self) -> bool:
+        if self.working:
+            return True
+
+        if (time.time() - self.last_check) > (5 * 60):
+            # It's been 5 minutes, try again
+            self.working = True
+            return True
 
-            if (time.time() - self.last_check) > (5 * 60):
-                # It's been 5 minutes, try again
-                self.working = True
-                return True
+        return False
 
-            return False
 
+class Registry(Generic[_T]):
     def __init__(self):
-        self._items: Dict[str, "Registry.Item"] = {}
+        self._items: Dict[str, RegistryItem[_T]] = {}
 
-    def register(self, name, item):
+    def register(self, name: str, item: _T) -> None:
         if name in self._items:
             raise ValueError("Attempt to register duplicate item")
 
-        self._items[name] = self.Item(item)
+        self._items[name] = RegistryItem(item)
 
-    def get(self, name):
+    def get(self, name: str) -> Optional[_T]:
         val = self._items.get(name)
         if val:
             return val.item
 
-        return val
+        return None
 
-    def get_item(self, name):
+    def get_item(self, name: str) -> Optional[RegistryItem[_T]]:
         return self._items.get(name)
 
-    def get_working(self) -> Optional["Item"]:
+    def get_working(self) -> Optional[RegistryItem[_T]]:
         working = [item for item in self._items.values() if item.should_use]
 
         if not working:
             return None
 
         return min(working, key=attrgetter("uses"))
 
-    def remove(self, name):
+    def remove(self, name: str) -> None:
         del self._items[name]
 
-    def items(self):
+    def items(self) -> Iterable[Tuple[str, RegistryItem[_T]]]:
         return self._items.items()
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[str]:
         return iter(self._items)
 
-    def __getitem__(self, item):
+    def __getitem__(self, item: str) -> _T:
         return self._items[item].item
 
-    def set_working(self):
+    def set_working(self) -> None:
         for item in self._items.values():
             item.working = True
 
 
-def shorten(url, custom=None, key=None, service=DEFAULT_SHORTENER):
+def shorten(url: str, custom=None, key=None, service=DEFAULT_SHORTENER):
     impl = shorteners[service]
     return impl.shorten(url, custom, key)
 
@@ -140,7 +152,12 @@ class NoPasteException(Exception):
     """No pastebins succeeded"""
 
 
-def paste(data, ext="txt", service=DEFAULT_PASTEBIN, raise_on_no_paste=False):
+def paste(
+    data: Union[str, bytes],
+    ext="txt",
+    service=DEFAULT_PASTEBIN,
+    raise_on_no_paste=False,
+) -> str:
     if service:
         impl = pastebins.get_item(service)
     else:
@@ -218,12 +235,12 @@ class Pastebin:
     def __init__(self):
         pass
 
-    def paste(self, data, ext):
+    def paste(self, data, ext) -> str:
         raise NotImplementedError
 
 
-shorteners = Registry()
-pastebins = Registry()
+shorteners = Registry[Shortener]()
+pastebins = Registry[Pastebin]()
 
 # Internal Implementations
 
@@ -346,7 +363,7 @@ def __init__(self, base_url):
         super().__init__()
         self.url = base_url
 
-    def paste(self, data, ext):
+    def paste(self, data, ext) -> str:
         if isinstance(data, str):
             encoded = data.encode()
         else: