Experiment: Use typing-extensions.

The main thing gained here is the ability to use Unpack to properly hint
the large set of optional keyword arguments accepted by Akismet, but
it's possible other interesting uses will spring up.

Author: ubernostrum

noxfile.py

@@ -323,7 +323,7 @@ def lint_pylint(session: nox.Session) -> None:
 
     """
     # Pylint requires that all dependencies be importable during the run.
-    session.install("httpx", "pylint")
+    session.install("httpx", "typing-extensions", "pylint")
     session.run(f"python{session.python}", "-Im", "pylint", "--version")
     session.run(f"python{session.python}", "-Im", "pylint", "src/", "tests/")
     clean()

pdm.lock

@@ -29,6 +29,7 @@ name = "akismet"
 description = "A Python interface to the Akismet spam-filtering service."
 dependencies = [
     "httpx",
+    "typing-extensions",
 ]
 keywords = ["akismet", "spam", "spam-filtering"]
 license = { text = "BSD-3-Clause" }
@@ -81,6 +82,7 @@ source-includes = [
     ".readthedocs.yaml",
     "AUTHORS",
     "CONTRIBUTING.rst",
+    "Makefile",
     "docs/",
     "noxfile.py",
     "pdm.lock",

pyproject.toml

@@ -90,7 +90,7 @@
 # SPDX-License-Identifier: BSD-3-Clause
 
 from ._async_client import AsyncClient
-from ._common import USER_AGENT, CheckResponse, Config
+from ._common import USER_AGENT, AkismetArguments, CheckResponse, Config
 from ._exceptions import (
     AkismetError,
     APIKeyError,
@@ -106,6 +106,7 @@
 __all__ = [
     "APIKeyError",
     "Akismet",
+    "AkismetArguments",
     "AkismetError",
     "AsyncClient",
     "CheckResponse",

src/akismet/__init__.py

@@ -9,8 +9,29 @@
 from typing import TYPE_CHECKING, Literal, Optional, Type, Union
 
 import httpx
-
-from . import _common, _exceptions
+from typing_extensions import Self, Unpack
+
+from . import _exceptions
+from ._common import (
+    _API_URL,
+    _API_V11,
+    _API_V12,
+    _COMMENT_CHECK,
+    _KEY_SITES,
+    _OPTIONAL_KEYS,
+    _REQUEST_METHODS,
+    _SUBMISSION_RESPONSE,
+    _SUBMIT_HAM,
+    _SUBMIT_SPAM,
+    _USAGE_LIMIT,
+    _VERIFY_KEY,
+    AkismetArguments,
+    CheckResponse,
+    _configuration_error,
+    _get_async_http_client,
+    _protocol_error,
+    _try_discover_config,
+)
 
 if TYPE_CHECKING:  # pragma: no cover
     import akismet
@@ -139,15 +160,15 @@ def __init__(
         You will almost always want to use :meth:`validated_client` instead.
 
         """
-        self._config = config if config is not None else _common._try_discover_config()
-        self._http_client = http_client or _common._get_async_http_client()
+        self._config = config if config is not None else _try_discover_config()
+        self._http_client = http_client or _get_async_http_client()
 
     @classmethod
     async def validated_client(
         cls,
         config: Optional["akismet.Config"] = None,
         http_client: Optional[httpx.AsyncClient] = None,
-    ) -> "AsyncClient":
+    ) -> Self:
         """
         Constructor of :class:`AsyncClient`.
 
@@ -186,7 +207,7 @@ async def validated_client(
         # alternative constructor in order to achieve API consistency.
         instance = cls(config=config, http_client=http_client)
         if not await instance.verify_key():
-            _common._configuration_error(instance._config)
+            _configuration_error(instance._config)
         return instance
 
     # Async context-manager protocol.
@@ -198,7 +219,7 @@ async def __aenter__(self) -> "AsyncClient":
 
         """
         if not await self.verify_key():
-            _common._configuration_error(self._config)
+            _configuration_error(self._config)
         return self
 
     async def __aexit__(
@@ -215,7 +236,7 @@ async def __aexit__(
 
     async def _request(
         self,
-        method: _common._REQUEST_METHODS,
+        method: _REQUEST_METHODS,
         version: str,
         endpoint: str,
         data: dict,
@@ -244,7 +265,7 @@ async def _request(
         request_kwarg = "data" if method == "POST" else "params"
         try:
             response = await handler(
-                f"{_common._API_URL}/{version}/{endpoint}", **{request_kwarg: data}
+                f"{_API_URL}/{version}/{endpoint}", **{request_kwarg: data}
             )
             response.raise_for_status()
         except httpx.HTTPStatusError as exc:
@@ -260,7 +281,7 @@ async def _request(
         # Since it's possible to construct a client without performing up-front API key
         # validation, we have to watch out here for the possibility that we're making
         # requests with an invalid key, and raise the appropriate exception.
-        if endpoint != _common._VERIFY_KEY and response.text == "invalid":
+        if endpoint != _VERIFY_KEY and response.text == "invalid":
             raise _exceptions.APIKeyError(
                 "Akismet API key and/or site URL are invalid."
             )
@@ -309,7 +330,7 @@ async def _post_request(
            optional argument names.
 
         """
-        unknown_args = [k for k in kwargs if k not in _common._OPTIONAL_KEYS]
+        unknown_args = [k for k in kwargs if k not in _OPTIONAL_KEYS]
         if unknown_args:
             raise _exceptions.UnknownArgumentError(
                 f"Received unknown argument(s) for Akismet operation {endpoint}: "
@@ -337,17 +358,17 @@ async def _submit(self, endpoint: str, user_ip: str, **kwargs: str) -> bool:
 
         """
         response = await self._post_request(
-            _common._API_V11, endpoint, user_ip=user_ip, **kwargs
+            _API_V11, endpoint, user_ip=user_ip, **kwargs
         )
-        if response.text == _common._SUBMISSION_RESPONSE:
+        if response.text == _SUBMISSION_RESPONSE:
             return True
-        _common._protocol_error(endpoint, response)
+        _protocol_error(endpoint, response)
 
     # Public methods corresponding to the methods of the Akismet API.
     # ----------------------------------------------------------------------------
 
     async def comment_check(
-        self, user_ip: str, **kwargs: str
+        self, user_ip: str, **kwargs: Unpack[AkismetArguments]
     ) -> "akismet.CheckResponse":
         """
         Check a piece of user-submitted content to determine whether it is spam.
@@ -392,17 +413,19 @@ async def comment_check(
 
         """
         response = await self._post_request(
-            _common._API_V11, _common._COMMENT_CHECK, user_ip=user_ip, **kwargs
+            _API_V11, _COMMENT_CHECK, user_ip=user_ip, **kwargs
         )
         if response.text == "true":
             if response.headers.get("X-akismet-pro-tip", "") == "discard":
-                return _common.CheckResponse.DISCARD
-            return _common.CheckResponse.SPAM
+                return CheckResponse.DISCARD
+            return CheckResponse.SPAM
         if response.text == "false":
-            return _common.CheckResponse.HAM
-        _common._protocol_error(_common._COMMENT_CHECK, response)
+            return CheckResponse.HAM
+        _protocol_error(_COMMENT_CHECK, response)
 
-    async def submit_ham(self, user_ip: str, **kwargs: str) -> bool:
+    async def submit_ham(
+        self, user_ip: str, **kwargs: Unpack[AkismetArguments]
+    ) -> bool:
         """
         Inform Akismet that a piece of user-submitted comment is not spam.
 
@@ -440,9 +463,11 @@ async def submit_ham(self, user_ip: str, **kwargs: str) -> bool:
            received from the Akismet API.
 
         """
-        return await self._submit(_common._SUBMIT_HAM, user_ip, **kwargs)
+        return await self._submit(_SUBMIT_HAM, user_ip, **kwargs)
 
-    async def submit_spam(self, user_ip: str, **kwargs: str) -> bool:
+    async def submit_spam(
+        self, user_ip: str, **kwargs: Unpack[AkismetArguments]
+    ) -> bool:
         """
         Inform Akismet that a piece of user-submitted comment is spam.
 
@@ -480,9 +505,10 @@ async def submit_spam(self, user_ip: str, **kwargs: str) -> bool:
            received from the Akismet API.
 
         """
-        return await self._submit(_common._SUBMIT_SPAM, user_ip, **kwargs)
+        return await self._submit(_SUBMIT_SPAM, user_ip, **kwargs)
 
-    async def key_sites(  # pylint: disable=too-many-positional-arguments,too-many-arguments
+    async def key_sites(
+        # pylint: disable=too-many-positional-arguments,too-many-arguments
         self,
         month: Optional[str] = None,
         url_filter: Optional[str] = None,
@@ -531,7 +557,7 @@ async def key_sites(  # pylint: disable=too-many-positional-arguments,too-many-a
         ):
             if value is not None:
                 params[argument] = value
-        response = await self._get_request(_common._API_V12, _common._KEY_SITES, params)
+        response = await self._get_request(_API_V12, _KEY_SITES, params)
         if result_format == "csv":
             return response.text
         return response.json()
@@ -546,7 +572,7 @@ async def usage_limit(self) -> dict:
 
         """
         response = await self._get_request(
-            _common._API_V12, _common._USAGE_LIMIT, params={"api_key": self._config.key}
+            _API_V12, _USAGE_LIMIT, params={"api_key": self._config.key}
         )
         return response.json()
 
@@ -575,10 +601,10 @@ async def verify_key(
         if not all([key, url]):
             key, url = self._config
         response = await self._request(
-            "POST", _common._API_V11, _common._VERIFY_KEY, {"key": key, "blog": url}
+            "POST", _API_V11, _VERIFY_KEY, {"key": key, "blog": url}
         )
         if response.text == "valid":
             return True
         if response.text == "invalid":
             return False
-        _common._protocol_error(_common._VERIFY_KEY, response)
+        _protocol_error(_VERIFY_KEY, response)

src/akismet/_async_client.py

@@ -9,8 +9,8 @@
 import os
 import sys
 import textwrap
-import typing
 from importlib.metadata import version
+from typing import Literal, NamedTuple, NoReturn, TypedDict
 
 import httpx
 
@@ -24,7 +24,7 @@
 _API_V12 = "1.2"
 _COMMENT_CHECK = "comment-check"
 _KEY_SITES = "key-sites"
-_REQUEST_METHODS = typing.Literal["GET", "POST"]  # pylint: disable=invalid-name
+_REQUEST_METHODS = Literal["GET", "POST"]  # pylint: disable=invalid-name
 _SUBMISSION_RESPONSE = "Thanks for making the web a better place."
 _SUBMIT_HAM = "submit-ham"
 _SUBMIT_SPAM = "submit-spam"
@@ -81,7 +81,7 @@ class CheckResponse(enum.IntEnum):
     DISCARD = 2
 
 
-class Config(typing.NamedTuple):
+class Config(NamedTuple):
     """
     A :func:`~collections.namedtuple` representing Akismet configuration, consisting
     of a key and a URL.
@@ -96,11 +96,37 @@ class Config(typing.NamedTuple):
     url: str
 
 
+class AkismetArguments(TypedDict, total=False):
+    """
+    A :class:`~typing.TypedDict` representing the optional keyword arguments accepted by
+    most Akismet API operations.
+
+    """
+
+    blog_charset: str
+    blog_lang: str
+    comment_author: str
+    comment_author_email: str
+    comment_author_url: str
+    comment_content: str
+    comment_context: str
+    comment_date_gmt: str
+    comment_post_modified_gmt: str
+    comment_type: str
+    honeypot_field_name: str
+    is_test: bool
+    permalink: str
+    recheck_reason: str
+    referrer: str
+    user_agent: str
+    user_role: str
+
+
 # Private helper functions.
 # -------------------------------------------------------------------------------
 
 
-def _configuration_error(config: Config) -> typing.NoReturn:
+def _configuration_error(config: Config) -> NoReturn:
     """
     Raise an appropriate exception for invalid configuration.
 
@@ -133,7 +159,7 @@ def _get_sync_http_client() -> httpx.Client:
     return httpx.Client(headers={"User-Agent": USER_AGENT}, timeout=_TIMEOUT)
 
 
-def _protocol_error(operation: str, response: httpx.Response) -> typing.NoReturn:
+def _protocol_error(operation: str, response: httpx.Response) -> NoReturn:
     """
     Raise an appropriate exception for unexpected API responses.