refactor: improve timestamp pydantic logic

Author: zombyacoff

examples/simple_discord_bot.py

@@ -38,7 +38,7 @@ async def cog_slash_command_error(  # noqa: PLR6301
                 )
             case faceit.exceptions.APIError():
                 await inter.edit_original_response(
-                    f"⚠️ API Error [{error.status_code}]: {error.message}"
+                    f"⚠️ API Error (`{error.status_code}`): {error.message}"
                 )
             case _:
                 await inter.edit_original_response(
@@ -63,8 +63,7 @@ async def stats(
         cs2_game = player.games.get(faceit.GameID.CS2)
         if cs2_game is None:
             return await inter.edit_original_response(
-                f"🔎 Player **{player.nickname}** found, "
-                "but they don't have CS2 linked."
+                f"🔎 Player `{player.nickname}` found, but they don't have CS2 linked."
             )
 
         player_stats = await self.faceit_data.players.stats(

pyproject.toml

@@ -23,28 +23,30 @@ classifiers = [
     "Typing :: Typed",
 ]
 dependencies = [
-    "httpx>=0.27.0,<1.0.0",
-    "pydantic>=2.7.1,<3.0.0",
-    "tenacity>=8.2.3,<9.0.0",
+    "httpx>=0.28.0",
+    "pydantic>=2.13.0",
+    "tenacity>=8.5.0",
 ]
 
 [project.optional-dependencies]
 env = ["python-decouple>=3.8"]
 
 [dependency-groups]
 dev = [
-    "mypy>=1.14",
-    "pre-commit==3.5",
-    "pytest>=7.4",
-    "pytest-asyncio>=0.21.1",
-    "ruff>=0.4.8",
+    "mypy>=2.0.0",
+    "pre-commit>=3.5",
+    "pytest>=9.0.0",
+    "pytest-asyncio>=1.3.0",
+    "ruff>=0.15.0",
 ]
 examples = [
-    "disnake>=2.0.0,<3.0.0",
+    "disnake>=2.0.0",
 ]
 
 [project.urls]
-"Documentation" = "https://docs.faceit.com/docs"
+"Bug Tracker" = "https://github.com/zombyacoff/faceit-python/issues"
+"Releases" = "https://github.com/zombyacoff/faceit-python/releases"
+"Repository" = "https://github.com/zombyacoff/faceit-python"
 
 [build-system]
 requires = ["uv_build>=0.11.14"]

src/faceit/api/base.py

@@ -28,7 +28,7 @@
     RawAPIPageResponse,
     RawAPIResponse,
 )
-from faceit.utils import warn_stacklevel
+from faceit.utils import find_user_stacklevel
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -172,7 +172,7 @@ def _validate_response(
                 "unprocessed data."
             )
             msg = default_warn_msg if warn_msg is None else warn_msg
-            warnings.warn(msg, stacklevel=warn_stacklevel())
+            warnings.warn(msg, stacklevel=find_user_stacklevel())
             return response
         try:
             return validator.model_validate(response)

src/faceit/api/data/matches.py

@@ -18,11 +18,9 @@
 )
 
 _MatchID: TypeAlias = str
-# We use `AfterValidator` with the `_MatchID` type alias instead of `FaceitMatchID` directly
-# to avoid mypy complaints. Mypy cannot fully recognize our custom type as compatible
-# with str, so this approach ensures proper type checking and validation.
 _MatchIDValidated: TypeAlias = Annotated[
-    _MatchID, AfterValidator(FaceitMatchID._validate)
+    _MatchID,
+    AfterValidator(FaceitMatchID),
 ]
 
 

src/faceit/api/pagination.py

@@ -45,20 +45,22 @@
     deduplicate_unhashable,
     deep_get,
     extends,
+    find_user_stacklevel,
     representation,
     validate_positive_int,
-    warn_stacklevel,
 )
 
-_PageType: TypeAlias = ItemPage[Any] | RawAPIPageResponse
-_PageList: TypeAlias = list[_PageType]
-_PageT = TypeVar("_PageT", bound=_PageType)
+ProcessedPages: TypeAlias = list[RawAPIItem] | ItemPage[_T]
+_PageType: TypeAlias = RawAPIPageResponse | ItemPage[_T]
+_PageList: TypeAlias = list[_PageType[_T]]
+_PageT = TypeVar("_PageT", bound=_PageType[Any])
 
 
 if TYPE_CHECKING:
-    _PageClass: TypeAlias = type[ItemPage[Any] | RawAPIPageResponse]
-    _PageFactory: TypeAlias = Callable[[_PageList], _PageClass]
-    _PageFactoryMap: TypeAlias = Mapping["CollectReturnFormat", _PageFactory]
+    _PageFactoryMap: TypeAlias = Mapping[
+        "CollectReturnFormat",
+        Callable[[_PageList[Any]], type[RawAPIPageResponse | ItemPage[Any]]],
+    ]
     _OptionalTimestampPaginationConfig: TypeAlias = (
         "TimestampPaginationConfig | Literal[False]"
     )
@@ -336,7 +338,7 @@ def warn_if_exceeds_safe(max_pages: int, /) -> int:
                     f"The computed number of pages ({max_pages}) exceeds the "
                     f"recommended safe maximum ({self.__class__.SAFE_MAX_PAGES}). "
                     "Proceed at your own risk.",
-                    stacklevel=warn_stacklevel(),
+                    stacklevel=find_user_stacklevel(),
                 )
             return max_pages
 
@@ -396,7 +398,7 @@ def _remove_pagination_args(**kwargs: _T) -> dict[str, _T]:
                 f"Pagination parameters {_PAGINATION_ARGS} should not be "
                 "provided by users. These parameters are managed internally "
                 "by the pagination system.",
-                stacklevel=warn_stacklevel(),
+                stacklevel=find_user_stacklevel(),
             )
         return kwargs
 
@@ -424,7 +426,7 @@ def _validate_unix_config(
 
     @staticmethod
     def _extract_unix_timestamp(
-        cfg: TimestampPaginationConfig, page: _PageType | None, /
+        cfg: TimestampPaginationConfig, page: _PageType[Any] | None, /
     ) -> int | None:
         if not page:
             return None
@@ -460,16 +462,16 @@ def _validate_unix_pagination_parameter(
             warnings.warn(
                 "The parameters 'start' and 'to' will be managed automatically with Unix "
                 "timestamp pagination. Your provided values will be ignored.",
-                stacklevel=warn_stacklevel(),
+                stacklevel=find_user_stacklevel(),
             )
 
     @classmethod
     def _process_collected_pages(
         cls,
-        collection: list[RawAPIPageResponse | ItemPage[_T]],
+        collection: _PageList[_T],
         return_format: CollectReturnFormat,
         deduplicate: bool,  # noqa: FBT001
-    ) -> list[RawAPIItem] | ItemPage[_T]:
+    ) -> ProcessedPages[_T]:
         if cls._COLLECT_RETURN_FORMATS[return_format](collection) is dict:
             raw = chain.from_iterable(
                 p[RAW_RESPONSE_ITEMS_KEY] for p in collection if isinstance(p, dict)
@@ -481,7 +483,7 @@ def _process_collected_pages(
     @classmethod
     def _deduplicate_collection(
         cls, collection: Iterable[RawAPIItem] | ItemPage[_T], /
-    ) -> list[RawAPIItem] | ItemPage[_T]:
+    ) -> ProcessedPages[_T]:
         if not isinstance(collection, ItemPage):
             return deduplicate_unhashable(collection)
         return collection.with_items(deduplicate_unhashable(collection))  # pyright: ignore[reportArgumentType, reportReturnType]
@@ -577,7 +579,7 @@ def collect(
         self: SyncPageIterator[RawAPIPageResponse] | SyncPageIterator[ItemPage[_T]],
         *,
         deduplicate: bool = True,
-    ) -> list[RawAPIItem] | ItemPage[_T]:
+    ) -> ProcessedPages[_T]:
         return self.__class__.gather_from_iterator(self, deduplicate=deduplicate)
 
     @classmethod
@@ -649,7 +651,7 @@ def gather_from_iterator(
         return_format: CollectReturnFormat = CollectReturnFormat.FIRST,
         *,
         deduplicate: bool = True,
-    ) -> list[RawAPIItem] | ItemPage[_T]:
+    ) -> ProcessedPages[_T]:
         return cls._process_collected_pages(list(iterator), return_format, deduplicate)
 
 
@@ -669,7 +671,7 @@ async def collect(
 
     async def collect(
         self: AsyncPageIterator[RawAPIPageResponse] | AsyncPageIterator[ItemPage[_T]],
-    ) -> list[RawAPIItem] | ItemPage[_T]:
+    ) -> ProcessedPages[_T]:
         return await self.__class__.gather_from_iterator(self)
 
     @classmethod
@@ -741,7 +743,7 @@ async def gather_from_iterator(
         return_format: CollectReturnFormat = CollectReturnFormat.FIRST,
         *,
         deduplicate: bool = True,
-    ) -> list[RawAPIItem] | ItemPage[_T]:
+    ) -> ProcessedPages[_T]:
         return cls._process_collected_pages(
             [page async for page in iterator], return_format, deduplicate
         )

src/faceit/http/client.py

@@ -515,7 +515,7 @@ def close(cls) -> Never:
         Async clients should use :meth:`~.aclose` instead.
         """
         msg = (
-            f"Use 'await {cls.__name__}.aclose()' instead of '{cls.__name__}.close().'"
+            f"Use 'await {cls.__name__}.aclose()' instead of '{cls.__name__}.close()'."
         )
         raise RuntimeError(msg)
 

src/faceit/models/custom_types/common.py

@@ -17,7 +17,6 @@
 from pydantic import (
     AfterValidator,
     BeforeValidator,
-    GetCoreSchemaHandler,
     RootModel,
     model_validator,
 )
@@ -29,7 +28,6 @@
 if TYPE_CHECKING:
     from collections.abc import ItemsView, Iterator, KeysView, ValuesView
 
-_INJECTED_KEY: Final = "injected_key"
 _LANG_PLACEHOLDER: Final = "{lang}"
 _LANG_PATTERN: Final = re.compile(rf"/?{re.escape(_LANG_PLACEHOLDER)}/?")
 
@@ -71,17 +69,11 @@ def from_datetime(cls, dt: datetime, /) -> Self:
         return cls(round(dt.timestamp() * cls._UNITS_PER_SEC))
 
     @classmethod
-    def _validate(cls, value: int, /) -> Self:
-        if value >= 0:
-            return cls(value)
-        msg = f"Value {value} is negative. Timestamp cannot be negative"
-        raise ValueError(msg)
-
-    @classmethod
-    def __get_pydantic_core_schema__(
-        cls, _: type[Any], handler: GetCoreSchemaHandler
-    ) -> core_schema.CoreSchema:
-        return core_schema.no_info_after_validator_function(cls._validate, handler(int))
+    def __get_pydantic_core_schema__(cls, *_: Any, **__: Any) -> core_schema.CoreSchema:
+        return core_schema.chain_schema([
+            core_schema.int_schema(ge=0),
+            core_schema.no_info_after_validator_function(cls, core_schema.any_schema()),
+        ])
 
 
 @final
@@ -103,6 +95,9 @@ def as_ms(self) -> TimestampMs:
 NotStrictTimestampSec: TypeAlias = TimestampSec | int
 
 
+_INJECTED_KEY: Final = "injected_key"
+
+
 @final
 class ResponseContainer(RootModel[dict[str, _T]]):
     __slots__ = ()
@@ -128,7 +123,7 @@ def get(self, key: str, /, default: _R | None = None) -> _T | _R | None:
     def __getattr__(self, name: str) -> _T:
         if name in self.root:
             return self.root[name]
-        msg = f"'{self.__class__.__name__}' object has no attribute '{name}'"
+        msg = f"{self.__class__.__name__!r} object has no attribute {name!r}"
         raise AttributeError(msg)
 
     def __iter__(self) -> Iterator[str]:  # type: ignore[override]

src/faceit/types.py

@@ -63,8 +63,8 @@
         "start": int,
         "end": int,
         # Unix timestamps (in milliseconds)
-        "from": NotRequired["TimestampMs"],
-        "to": NotRequired["TimestampMs"],
+        "from": NotRequired[int],
+        "to": NotRequired[int],
     },
 )
 RawAPIResponse: TypeAlias = RawAPIItem | RawAPIPageResponse

src/faceit/utils.py

@@ -198,10 +198,10 @@ def validate_positive_int(value: Any, /, param_name: str = "value") -> int:
     impractical or unavailable.
     """
     if not isinstance(value, int):
-        msg = f"'{param_name}' must be int, got {type(value).__name__}"
+        msg = f"{param_name!r} must be int, got {type(value).__name__}"
         raise TypeError(msg)
     if value <= 0:
-        msg = f"'{param_name}' must be a positive integer, got {value}"
+        msg = f"{param_name!r} must be a positive integer, got {value}"
         raise ValueError(msg)
     return value
 
@@ -233,7 +233,7 @@ def _get_ignored_paths() -> tuple[
     return tuple(prefixes), frozenset(files)
 
 
-def warn_stacklevel() -> int:
+def find_user_stacklevel() -> int:
     """
     Determines the appropriate stack level for warnings emitted by the library,
     so that they point to the user's code instead of internal library frames.