diff --git a/noxfile.py b/noxfile.py index ec37516..73711fd 100644 --- a/noxfile.py +++ b/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() diff --git a/pdm.lock b/pdm.lock index d060777..ec6f5ca 100644 --- a/pdm.lock +++ b/pdm.lock @@ -5,7 +5,7 @@ groups = ["default", "tests"] strategy = ["inherit_metadata"] lock_version = "4.5.0" -content_hash = "sha256:5c91e0f67b6dc04dcd7901438e71cc33674d2782fce312dec1dcaf472162a54e" +content_hash = "sha256:36a345b300f04cac3153028b89ff6a517eeb26f628c9099aa666aa29619cc0db" [[metadata.targets]] requires_python = ">=3.9,<3.10" @@ -238,7 +238,6 @@ version = "4.12.2" requires_python = ">=3.8" summary = "Backported and Experimental Type Hints for Python 3.8+" groups = ["default"] -marker = "python_version < \"3.11\" and python_version >= \"3.9\"" files = [ {file = "typing_extensions-4.12.2-py3-none-any.whl", hash = "sha256:04e5ca0351e0f3f85c6853954072df659d0d13fac324d0072316b67d7794700d"}, {file = "typing_extensions-4.12.2.tar.gz", hash = "sha256:1a7ead55c7e559dd4dee8856e3a88b41225abfe1ce8df57b7c13915fe121ffb8"}, diff --git a/pyproject.toml b/pyproject.toml index 1462507..65fea5c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/src/akismet/__init__.py b/src/akismet/__init__.py index 129d1ad..7604cd7 100644 --- a/src/akismet/__init__.py +++ b/src/akismet/__init__.py @@ -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", diff --git a/src/akismet/_async_client.py b/src/akismet/_async_client.py index 6a9da55..00bdb4f 100644 --- a/src/akismet/_async_client.py +++ b/src/akismet/_async_client.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) diff --git a/src/akismet/_common.py b/src/akismet/_common.py index dba5a0d..e4eaae4 100644 --- a/src/akismet/_common.py +++ b/src/akismet/_common.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. diff --git a/src/akismet/_sync_client.py b/src/akismet/_sync_client.py index 2afe8ed..78cdc6c 100644 --- a/src/akismet/_sync_client.py +++ b/src/akismet/_sync_client.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_sync_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_sync_http_client() + self._config = config if config is not None else _try_discover_config() + self._http_client = http_client or _get_sync_http_client() @classmethod def validated_client( cls, config: Optional["akismet.Config"] = None, http_client: Optional[httpx.Client] = None, - ) -> "SyncClient": + ) -> Self: """ Constructor of :class:`SyncClient`. @@ -187,7 +208,7 @@ def validated_client( # constructor in order to achieve API consistency. instance = cls(config=config, http_client=http_client) if not instance.verify_key(): - _common._configuration_error(instance._config) + _configuration_error(instance._config) return instance # Context-manager protocol. @@ -199,7 +220,7 @@ def __enter__(self) -> "SyncClient": """ if not self.verify_key(): - _common._configuration_error(self._config) + _configuration_error(self._config) return self def __exit__( @@ -216,7 +237,7 @@ def __exit__( def _request( self, - method: _common._REQUEST_METHODS, + method: _REQUEST_METHODS, version: str, endpoint: str, data: dict, @@ -245,7 +266,7 @@ def _request( request_kwarg = "data" if method == "POST" else "params" try: response = 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: @@ -261,7 +282,7 @@ 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." ) @@ -308,7 +329,7 @@ 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}: " @@ -335,17 +356,17 @@ def _submit(self, endpoint: str, user_ip: str, **kwargs: str) -> bool: received from the Akismet API. """ - response = self._post_request( - _common._API_V11, endpoint, user_ip=user_ip, **kwargs - ) - if response.text == _common._SUBMISSION_RESPONSE: + response = self._post_request(_API_V11, endpoint, user_ip=user_ip, **kwargs) + 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. # ---------------------------------------------------------------------------- - def comment_check(self, user_ip: str, **kwargs: str) -> "akismet.CheckResponse": + def comment_check( + self, user_ip: str, **kwargs: Unpack[AkismetArguments] + ) -> "akismet.CheckResponse": """ Check a piece of user-submitted content to determine whether it is spam. @@ -389,17 +410,17 @@ def comment_check(self, user_ip: str, **kwargs: str) -> "akismet.CheckResponse": """ response = 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) - def submit_ham(self, user_ip: str, **kwargs: str) -> bool: + def submit_ham(self, user_ip: str, **kwargs: Unpack[AkismetArguments]) -> bool: """ Inform Akismet that a piece of user-submitted comment is not spam. @@ -437,9 +458,9 @@ def submit_ham(self, user_ip: str, **kwargs: str) -> bool: received from the Akismet API. """ - return self._submit(_common._SUBMIT_HAM, user_ip, **kwargs) + return self._submit(_SUBMIT_HAM, user_ip, **kwargs) - def submit_spam(self, user_ip: str, **kwargs: str) -> bool: + def submit_spam(self, user_ip: str, **kwargs: Unpack[AkismetArguments]) -> bool: """ Inform Akismet that a piece of user-submitted comment is spam. @@ -477,7 +498,7 @@ def submit_spam(self, user_ip: str, **kwargs: str) -> bool: received from the Akismet API. """ - return self._submit(_common._SUBMIT_SPAM, user_ip, **kwargs) + return self._submit(_SUBMIT_SPAM, user_ip, **kwargs) def key_sites( # pylint: disable=too-many-positional-arguments,too-many-arguments self, @@ -528,7 +549,7 @@ def key_sites( # pylint: disable=too-many-positional-arguments,too-many-argumen ): if value is not None: params[argument] = value - response = self._get_request(_common._API_V12, _common._KEY_SITES, params) + response = self._get_request(_API_V12, _KEY_SITES, params) if result_format == "csv": return response.text return response.json() @@ -543,7 +564,7 @@ def usage_limit(self) -> dict: """ response = 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() @@ -570,10 +591,10 @@ def verify_key(self, key: Optional[str] = None, url: Optional[str] = None) -> bo if not all([key, url]): key, url = self._config response = 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)