Use obj.model_dump instead of obj.dict for Pydantic V2 (#14)

* Use `model_dump` instead of `dict` for Pydantic V2

* Improve type hints

* Fully type hints

Author: waketzheng

.github/workflows/build.yml

@@ -18,20 +18,20 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.8, 3.9, "3.10", "3.11"]
+        python-version: [3.8, 3.9, "3.10", 3.11, 3.12]
 
     steps:
-      - uses: actions/checkout@v1
+      - uses: actions/checkout@v4
         with:
           fetch-depth: 9
           submodules: false
 
       - name: Use Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: ${{ matrix.python-version }}
 
-      - uses: actions/cache@v1
+      - uses: actions/cache@v4
         id: depcache
         with:
           path: deps
@@ -58,6 +58,10 @@ jobs:
           isort --check-only . 2>&1
           black --check . 2>&1
 
+      - name: Check type hints
+        run: mypy .
+        if: matrix.python-version == 3.12
+
       - name: Upload pytest test results
         uses: actions/upload-artifact@master
         with:
@@ -90,7 +94,7 @@ jobs:
     if: github.event_name == 'release'
     steps:
       - name: Download a distribution artifact
-        uses: actions/download-artifact@v2
+        uses: actions/download-artifact@v4
         with:
           name: dist-package-3.10
           path: dist

CHANGELOG.md

@@ -5,9 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [1.1.6] - 2022-11-05 :snake:
+## [1.1.6] - 2024-11-07 :snake:
+- Drop python3.6 and python3.7 support
+- Use `obj.model_dump` instead of `obj.dict` for Pydantic V2
 - Workflow maintenance
-- Applies `black`, `flake8`, `isort`
+- Applies `black`, `flake8`, `isort`, `mypy`
 
 ## [1.1.5] - 2022-03-14 :tulip:
 - Adds `py.typed` file

Makefile

@@ -22,4 +22,11 @@ testcov:
 
 
 test:
-	pytest -v -W ignore
+	pytest -v -W ignore
+
+
+check:
+	flake8 .
+	isort --check-only .
+	black --check .
+	mypy .

essentials/caching.py

@@ -1,15 +1,23 @@
 import functools
 import time
 from collections import OrderedDict
-from typing import Any, Callable, Generic, Iterable, Iterator, Tuple, TypeVar
+from typing import TYPE_CHECKING, Any, Generic, Iterable, Iterator, Tuple, TypeVar
+
+if TYPE_CHECKING:
+    from typing import Callable, TypeVarTuple, Unpack
+
+    PosArgsT = TypeVarTuple("PosArgsT")
+    T_Retval = TypeVar("T_Retval")
+    FuncType = Callable[[Unpack[PosArgsT]], T_Retval]
+    FuncDecoType = Callable[[FuncType], FuncType]
 
 T = TypeVar("T")
 
 
 class Cache(Generic[T]):
     """In-memory LRU cache implementation."""
 
-    def __init__(self, max_size: int = 500):
+    def __init__(self, max_size: int = 500) -> None:
         self._bag: OrderedDict[Any, Any] = OrderedDict()
         self._max_size = -1
         self.max_size = max_size
@@ -50,7 +58,7 @@ def get(self, key, default=None) -> T:
     def set(self, key, value) -> None:
         self[key] = value
 
-    def _check_size(self):
+    def _check_size(self) -> None:
         while len(self._bag) > self.max_size:
             self._bag.popitem(last=False)
 
@@ -85,7 +93,7 @@ class CachedItem(Generic[T]):
 
     __slots__ = ("_value", "_time")
 
-    def __init__(self, value: T):
+    def __init__(self, value: T) -> None:
         self._value = value
         self._time = time.time()
 
@@ -94,7 +102,7 @@ def value(self) -> T:
         return self._value
 
     @value.setter
-    def value(self, value: T):
+    def value(self, value: T) -> None:
         self._value = value
         self._time = time.time()
 
@@ -107,8 +115,8 @@ class ExpiringCache(Cache[T]):
     """A cache whose items can expire by a given function."""
 
     def __init__(
-        self, expiration_policy: Callable[[CachedItem[T]], bool], max_size: int = 500
-    ):
+        self, expiration_policy: "Callable[[CachedItem[T]], bool]", max_size: int = 500
+    ) -> None:
         super().__init__(max_size)
         assert expiration_policy is not None
         self.expiration_policy = expiration_policy
@@ -120,12 +128,12 @@ def full(self) -> bool:
     def expired(self, item: CachedItem) -> bool:
         return self.expiration_policy(item)
 
-    def _remove_expired_items(self):
+    def _remove_expired_items(self) -> None:
         for key, item in list(self._bag.items()):
             if self.expired(item):
                 del self[key]
 
-    def _check_size(self):
+    def _check_size(self) -> None:
         if self.full:
             self._remove_expired_items()
         super()._check_size()
@@ -148,7 +156,7 @@ def __setitem__(self, key, value: T) -> None:
             self._check_size()
 
     @classmethod
-    def with_max_age(cls, max_age: float, max_size: int = 500):
+    def with_max_age(cls, max_age: float, max_size: int = 500) -> "ExpiringCache":
         """
         Returns an instance of ExpiringCache whose items are invalidated
         when they were set more than a given number of seconds ago.
@@ -174,7 +182,7 @@ def __iter__(self) -> Iterator[Tuple[Any, T]]:
                 yield (key, item.value)
 
 
-def lazy(max_seconds: int = 1, cache=None):
+def lazy(max_seconds: int = 1, cache=None) -> "FuncDecoType":
     """
     Wraps a function so that it is called up to once
     every max_seconds, by input arguments.

essentials/decorators/logs.py

@@ -2,15 +2,17 @@
 from functools import wraps
 from inspect import iscoroutinefunction
 from logging import Logger
-from typing import Callable, Optional
+from typing import Callable, Optional, TypeVar
 from uuid import uuid4
 
 from essentials.diagnostics import StopWatch
 
+T = TypeVar("T")
 IdFactory = Callable[[], str]
+FuncType = Callable[..., T]
 
 
-def _default_id_factory():
+def _default_id_factory() -> str:
     return str(uuid4())
 
 
@@ -25,7 +27,7 @@ def log(
     completed_msg="%s; completed; call id: %s; elapsed %s ms",
     completed_msg_with_output="%s; completed; call id: %s; elapsed %s ms; output: %s",
     exc_message="%s; unhandled exception; call id: %s; elapsed %s ms",
-):
+) -> Callable[[FuncType], FuncType]:
 
     if not id_factory:
         id_factory = _default_id_factory

essentials/decorators/retry.py

@@ -2,20 +2,22 @@
 import time
 from functools import wraps
 from inspect import iscoroutinefunction
-from typing import Callable, Optional, Tuple, Type, Union
+from typing import Callable, Optional, Tuple, Type, TypeVar, Union
 
+T = TypeVar("T")
+FuncType = Callable[..., T]
 CatchException = Union[Tuple[Type[Exception]], Type[Exception], None]
 OnException = Optional[Callable[[Type[Exception], int], None]]
 
 
 def _get_retry_async_wrapper(
-    fn,
+    fn: FuncType,
     times: int,
     delay: float,
     catch_exceptions_types: CatchException,
     on_exception: OnException,
     loop,
-):
+) -> FuncType:
     @wraps(fn)
     async def async_wrapper(*args, **kwargs):
         attempt = 0
@@ -46,7 +48,7 @@ def retry(
     catch_exceptions_types: CatchException = None,
     on_exception: OnException = None,
     loop=None,
-):
+) -> Callable[[FuncType], FuncType]:
     if catch_exceptions_types is None:
         catch_exceptions_types = Exception
 

essentials/diagnostics.py

@@ -2,47 +2,48 @@
 Timer class, edited from: Python Cookbook,
 3rd Edition by Brian K. Jones, David Beazley
 """
+
 import time
 
 
 class StopWatch:
-    def __init__(self, func=time.perf_counter):
+    def __init__(self, func=time.perf_counter) -> None:
         self._elapsed_s = 0.0
         self._func = func
         self._start = None
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<StopWatch elapsed s.: {self.elapsed_s}>"
 
-    def start(self):
+    def start(self) -> None:
         if self._start is not None:
             raise RuntimeError("StopWatch already running")
 
         self._start = self._func()
 
-    def stop(self):
+    def stop(self) -> None:
         if self._start is None:
             raise RuntimeError("StopWatch not running")
 
         self._elapsed_s += self._func() - self._start
 
-    def reset(self):
+    def reset(self) -> None:
         self._start = None
         self._elapsed_s = 0.0
 
     @property
-    def elapsed_s(self):
+    def elapsed_s(self) -> float:
         return self._elapsed_s
 
     @property
-    def elapsed_ms(self):
+    def elapsed_ms(self) -> float:
         if self.elapsed_s > 0.0:
             return self.elapsed_s * 1000
         return 0.0
 
-    def __enter__(self):
+    def __enter__(self) -> "StopWatch":
         self.start()
         return self
 
-    def __exit__(self, *args):
+    def __exit__(self, *args) -> None:
         self.stop()

essentials/folders.py

@@ -1,23 +1,24 @@
 import ntpath
 import os
+from typing import Tuple
 
 
-def ensure_folder(path):
+def ensure_folder(path) -> None:
     os.makedirs(path, exist_ok=True)
 
 
-def split_path(filepath):
+def split_path(filepath) -> Tuple[str, str, str]:
     """Splits a file path into folder path, file name and extension"""
     head, tail = ntpath.split(filepath)
     filename, extension = os.path.splitext(tail)
     return head, filename, extension
 
 
-def get_file_extension(filepath):
+def get_file_extension(filepath) -> str:
     _, file_extension = os.path.splitext(filepath)
     return file_extension
 
 
-def get_path_leaf(path):
+def get_path_leaf(path) -> str:
     head, tail = ntpath.split(path)
     return tail or ntpath.basename(head)

essentials/json.py

@@ -2,22 +2,26 @@
 This module defines a user-friendly json encoder,
 supporting time objects, UUID and bytes.
 """
+
 import base64
 import dataclasses
 import json
 from datetime import date, datetime, time
 from enum import Enum
+from typing import Any
 from uuid import UUID
 
 __all__ = ["FriendlyEncoder", "dumps"]
 
 
 class FriendlyEncoder(json.JSONEncoder):
-    def default(self, obj):
+    def default(self, obj: Any) -> Any:
         try:
             return json.JSONEncoder.default(self, obj)
         except TypeError:
             if hasattr(obj, "dict"):
+                if hasattr(obj, "model_dump"):
+                    return obj.model_dump()
                 return obj.dict()
             if isinstance(obj, time):
                 return obj.strftime("%H:%M:%S")
@@ -32,7 +36,7 @@ def default(self, obj):
             if isinstance(obj, Enum):
                 return obj.value
             if dataclasses.is_dataclass(obj):
-                return dataclasses.asdict(obj)
+                return dataclasses.asdict(obj)  # type:ignore[arg-type]
             raise
 
 
@@ -48,7 +52,7 @@ def dumps(
     default=None,
     sort_keys=False,
     **kw
-):
+) -> str:
     if cls is None:
         cls = FriendlyEncoder
     return json.dumps(

essentials/meta.py

@@ -1,14 +1,20 @@
 import warnings
 from functools import wraps
 from inspect import iscoroutinefunction
+from typing import Callable, Optional, TypeVar
+
+T = TypeVar("T")
+FuncType = Callable[..., T]
 
 
 class DeprecatedException(Exception):
-    def __init__(self, param_name):
+    def __init__(self, param_name: str) -> None:
         super().__init__("The function `%s` is deprecated" % param_name)
 
 
-def deprecated(message=None, raise_exception=False):
+def deprecated(
+    message: Optional[str] = None, raise_exception=False
+) -> Callable[[FuncType], FuncType]:
     """
     This is a decorator which can be used to mark functions
     as deprecated. It will result in a warning being emitted