diff --git a/jkit/_base.py b/jkit/_base.py index 32457eb..ea9a74b 100644 --- a/jkit/_base.py +++ b/jkit/_base.py @@ -1,7 +1,7 @@ from __future__ import annotations from abc import ABCMeta, abstractmethod -from typing import Callable, ClassVar, TypeVar +from typing import Any, Callable, ClassVar, TypeVar from msgspec import Struct, convert, to_builtins from msgspec import ValidationError as MsgspecValidationError @@ -10,36 +10,17 @@ from jkit.exceptions import ValidationError T = TypeVar("T", bound="DataObject") -P1 = TypeVar("P1", bound="CheckableMixin") -P2 = TypeVar("P2", bound="SlugAndUrlMixin") -P3 = TypeVar("P3", bound="IdAndUrlMixin") +P1 = TypeVar("P1", bound="CheckableResourceMixin") +P2 = TypeVar("P2", bound="SlugAndUrlResourceMixin") +P3 = TypeVar("P3", bound="IdAndUrlResourceMixin") -class DataObject(Struct, frozen=True, eq=True, kw_only=True): - def _validate(self: T) -> T: - if not CONFIG.data_validation.enabled: - return self +def _check_func_placeholder(x: Any) -> bool: # noqa: ANN401 + raise NotImplementedError - try: - return convert(to_builtins(self), type=self.__class__) - except MsgspecValidationError as e: - raise ValidationError(e.args[0]) from None - def __repr__(self) -> str: - field_strings: list[str] = [] - for field in self.__struct_fields__: - value = self.__getattribute__(field) - - if isinstance(value, str) and len(value) >= 100: # noqa: PLR2004 - formatted_value = value[:100] + "[truncated...]" - else: - formatted_value = value.__repr__() - - field_strings.append(f"{field}={formatted_value}") - - return ( - self.__class__.__name__ + "(\n " + ",\n ".join(field_strings) + "\n)" - ) +def _convert_func_placeholder(x: Any) -> Any: # noqa: ANN401 + raise NotImplementedError class ResourceObject: @@ -47,37 +28,36 @@ def __repr__(self) -> str: return f"{self.__class__.__name__}()" -class CheckableMixin(metaclass=ABCMeta): - def __init__(self) -> None: - self._checked = False +class SlugAndUrlResourceMixin: + _resource_readable_name: ClassVar[str] = "" - @abstractmethod - async def check(self) -> None: - raise NotImplementedError + _slug_check_func: Callable[[str], bool] = _check_func_placeholder + _url_check_func: Callable[[str], bool] = _check_func_placeholder - async def _require_check(self) -> None: - if self._checked or not CONFIG.resource_check.auto_check: - return + _url_to_slug_func: Callable[[str], str] = _convert_func_placeholder + _slug_to_url_func: Callable[[str], str] = _convert_func_placeholder - await self.check() - self._checked = True + def __init__(self, *, slug: str | None = None, url: str | None = None) -> None: + class_ = self.__class__ + name = class_._resource_readable_name - def _as_checked(self: P1) -> P1: - if not CONFIG.resource_check.force_check_safe_data: - self._checked = True + if slug is None and url is None: + raise ValueError(f"必须提供 {name} Slug 或 {name} Url") - return self + if slug is not None and url is not None: + raise ValueError(f"{name} Slug 与 {name} Url 不可同时提供") + if slug is not None: + if not class_._slug_check_func(slug): + raise ValueError(f"{slug} 不是有效的 {name} Slug") -class SlugAndUrlMixin: - _slug_check_func: ClassVar[Callable[[str], bool] | None] = None - _slug_to_url_func: ClassVar[Callable[[str], str] | None] = None - _url_to_slug_func: ClassVar[Callable[[str], str] | None] = None + self._slug = slug - def __init__(self, *, slug: str | None = None, url: str | None = None) -> None: - del slug, url + if url is not None: + if not class_._url_check_func(url): + raise ValueError(f"{url} 不是有效的 {name} Url") - self._slug = "" + self._slug = class_._url_to_slug_func(url) @classmethod def from_slug(cls: type[P2], slug: str, /) -> P2: @@ -93,77 +73,113 @@ def slug(self) -> str: @property def url(self) -> str: - if not self.__class__._slug_to_url_func: - raise AssertionError - return self.__class__._slug_to_url_func(self._slug) - @classmethod - def _check_params( - cls, - *, - object_readable_name: str, - slug: str | None, - url: str | None, - ) -> str: - # 如果同时提供了 Slug 和 Url - if slug is not None and url is not None: - raise ValueError( - f"{object_readable_name} Slug 与{object_readable_name}链接不可同时提供" - ) + def __eq__(self, other: object) -> bool: + return isinstance(other, self.__class__) and self.slug == other.slug - # 如果提供了 Slug - if slug is not None: - if not cls._slug_check_func: - raise AssertionError + def __repr__(self) -> str: + return f'{self.__class__.__name__}(slug="{self.slug}")' - if not cls._slug_check_func(slug): - raise ValueError(f"{slug} 不是有效的{object_readable_name} Slug") - return slug - # 如果提供了 Url - elif url is not None: # noqa: RET505 - if not cls._url_to_slug_func: - raise AssertionError +class IdAndUrlResourceMixin: + _resource_readable_name: ClassVar[str] = "" - # 转换函数中会对 Url 进行检查,并在 Url 无效时抛出异常 - return cls._url_to_slug_func(url) + _id_check_func: Callable[[int], bool] = _check_func_placeholder + _url_check_func: Callable[[str], bool] = _check_func_placeholder - # 如果 Slug 与 Url 均未提供 - raise ValueError( - f"必须提供{object_readable_name} Slug 或{object_readable_name}链接" - ) + _url_to_id_func: Callable[[str], int] = _convert_func_placeholder + _id_to_url_func: Callable[[int], str] = _convert_func_placeholder - def __eq__(self, other: object) -> bool: - return isinstance(other, self.__class__) and self.slug == other.slug + def __init__(self, *, id: int | None = None, url: str | None = None) -> None: + class_ = self.__class__ + name = class_._resource_readable_name - def __repr__(self) -> str: - return f'{self.__class__.__name__}(slug="{self.slug}")' + if id is None and url is None: + raise ValueError(f"必须提供 {name} Id 或 {name} Url") + + if id is not None and url is not None: + raise ValueError(f"{name} Id 与 {name} Url 不可同时提供") + + if id is not None: + if not class_._id_check_func(id): + raise ValueError(f"{id} 不是有效的 {name} Id") + + self._id = id + + if url is not None: + if not class_._url_check_func(url): + raise ValueError(f"{url} 不是有效的 {name} Url") + self._id = class_._url_to_id_func(url) -class IdAndUrlMixin(metaclass=ABCMeta): @classmethod - @abstractmethod def from_id(cls: type[P3], id: int, /) -> P3: - raise NotImplementedError + return cls(id=id) @classmethod - @abstractmethod def from_url(cls: type[P3], url: str, /) -> P3: - raise NotImplementedError + return cls(url=url) @property - @abstractmethod def id(self) -> int: - raise NotImplementedError + return self._id @property - @abstractmethod def url(self) -> str: - raise NotImplementedError + return self.__class__._id_to_url_func(self._id) def __eq__(self, other: object) -> bool: - return isinstance(other, self.__class__) and self.id == other.id + return isinstance(other, self.__class__) and self._id == other._id def __repr__(self) -> str: - return f'{self.__class__.__name__}(id="{self.id}")' + return f"{self.__class__.__name__}(id={self.id})" + + +class CheckableResourceMixin(metaclass=ABCMeta): + def __init__(self) -> None: + self._checked = False + + @abstractmethod + async def check(self) -> None: + raise NotImplementedError + + async def _require_check(self) -> None: + if self._checked or not CONFIG.resource_check.auto_check: + return + + await self.check() + self._checked = True + + def _as_checked(self: P1) -> P1: + if not CONFIG.resource_check.force_check_safe_data: + self._checked = True + + return self + + +class DataObject(Struct, frozen=True, eq=True, kw_only=True): + def _validate(self: T) -> T: + if not CONFIG.data_validation.enabled: + return self + + try: + return convert(to_builtins(self), type=self.__class__) + except MsgspecValidationError as e: + raise ValidationError(e.args[0]) from None + + def __repr__(self) -> str: + field_strings: list[str] = [] + for field in self.__struct_fields__: + value = self.__getattribute__(field) + + if isinstance(value, str) and len(value) >= 100: # noqa: PLR2004 + formatted_value = value[:100] + "[truncated...]" + else: + formatted_value = value.__repr__() + + field_strings.append(f"{field}={formatted_value}") + + return ( + self.__class__.__name__ + "(\n " + ",\n ".join(field_strings) + "\n)" + ) diff --git a/jkit/article.py b/jkit/article.py index d9fdfb2..ee34ef1 100644 --- a/jkit/article.py +++ b/jkit/article.py @@ -11,10 +11,10 @@ from httpx import HTTPStatusError from jkit._base import ( - CheckableMixin, + CheckableResourceMixin, DataObject, ResourceObject, - SlugAndUrlMixin, + SlugAndUrlResourceMixin, ) from jkit._network import send_request from jkit._normalization import ( @@ -28,7 +28,7 @@ _RESOURCE_UNAVAILABLE_STATUS_CODE, ) from jkit.exceptions import ResourceUnavailableError -from jkit.identifier_check import is_article_slug +from jkit.identifier_check import is_article_slug, is_article_url from jkit.identifier_convert import article_slug_to_url, article_url_to_slug from jkit.msgspec_constraints import ( CollectionSlug, @@ -203,24 +203,21 @@ class ArticleFeaturedCommentInfo( score: PositiveInt -class Article(ResourceObject, CheckableMixin, SlugAndUrlMixin): +class Article(ResourceObject, SlugAndUrlResourceMixin, CheckableResourceMixin): + _resource_readable_name = "文章" + _slug_check_func = is_article_slug - _slug_to_url_func = article_slug_to_url + _url_check_func = is_article_url + _url_to_slug_func = article_url_to_slug + _slug_to_url_func = article_slug_to_url - def __init__( - self, - *, - slug: str | None = None, - url: str | None = None, - ) -> None: - super().__init__() - - self._slug = self._check_params( - object_readable_name="文章", - slug=slug, - url=url, - ) + def __init__(self, *, slug: str | None = None, url: str | None = None) -> None: + SlugAndUrlResourceMixin.__init__(self, slug=slug, url=url) + CheckableResourceMixin.__init__(self) + + def __repr__(self) -> str: + return SlugAndUrlResourceMixin.__repr__(self) async def check(self) -> None: try: @@ -237,6 +234,8 @@ async def check(self) -> None: ) from None raise + else: + self._checked = True @property async def id(self) -> int: diff --git a/jkit/collection.py b/jkit/collection.py index 3fb51c0..1a34cc1 100644 --- a/jkit/collection.py +++ b/jkit/collection.py @@ -9,16 +9,16 @@ from httpx import HTTPStatusError from jkit._base import ( - CheckableMixin, + CheckableResourceMixin, DataObject, ResourceObject, - SlugAndUrlMixin, + SlugAndUrlResourceMixin, ) from jkit._network import send_request from jkit._normalization import normalize_assets_amount, normalize_datetime from jkit.constants import _RESOURCE_UNAVAILABLE_STATUS_CODE from jkit.exceptions import ResourceUnavailableError -from jkit.identifier_check import is_collection_slug +from jkit.identifier_check import is_collection_slug, is_collection_url from jkit.identifier_convert import collection_slug_to_url, collection_url_to_slug from jkit.msgspec_constraints import ( ArticleSlug, @@ -98,19 +98,21 @@ def to_article_obj(self) -> Article: return Article.from_slug(self.slug)._as_checked() -class Collection(ResourceObject, CheckableMixin, SlugAndUrlMixin): +class Collection(ResourceObject, SlugAndUrlResourceMixin, CheckableResourceMixin): + _resource_readable_name = "专题" + _slug_check_func = is_collection_slug - _slug_to_url_func = collection_slug_to_url + _url_check_func = is_collection_url + _url_to_slug_func = collection_url_to_slug + _slug_to_url_func = collection_slug_to_url def __init__(self, *, slug: str | None = None, url: str | None = None) -> None: - super().__init__() + SlugAndUrlResourceMixin.__init__(self, slug=slug, url=url) + CheckableResourceMixin.__init__(self) - self._slug = self._check_params( - object_readable_name="专题", - slug=slug, - url=url, - ) + def __repr__(self) -> str: + return SlugAndUrlResourceMixin.__repr__(self) async def check(self) -> None: try: @@ -127,6 +129,8 @@ async def check(self) -> None: ) from None raise + else: + self._checked = True @property async def info(self) -> CollectionInfo: diff --git a/jkit/notebook.py b/jkit/notebook.py index fe5b9f4..b73228e 100644 --- a/jkit/notebook.py +++ b/jkit/notebook.py @@ -6,17 +6,17 @@ from httpx import HTTPStatusError from jkit._base import ( - CheckableMixin, + CheckableResourceMixin, DataObject, - IdAndUrlMixin, + IdAndUrlResourceMixin, ResourceObject, ) from jkit._network import send_request from jkit._normalization import normalize_assets_amount, normalize_datetime from jkit.constants import _RESOURCE_UNAVAILABLE_STATUS_CODE from jkit.exceptions import ResourceUnavailableError -from jkit.identifier_check import is_notebook_id -from jkit.identifier_convert import notebook_id_to_url +from jkit.identifier_check import is_notebook_id, is_notebook_url +from jkit.identifier_convert import notebook_id_to_url, notebook_url_to_id from jkit.msgspec_constraints import ( ArticleSlug, NonEmptyStr, @@ -94,25 +94,21 @@ def to_article_obj(self) -> Article: return Article.from_slug(self.slug)._as_checked() -class Notebook(ResourceObject, CheckableMixin, IdAndUrlMixin): - def __init__(self, *, id: int) -> None: - super().__init__() +class Notebook(ResourceObject, IdAndUrlResourceMixin, CheckableResourceMixin): + _resource_readable_name = "文集" - if not is_notebook_id(id): - raise ValueError(f"文集 ID 无效:{id}") - self._id = id + _id_check_func = is_notebook_id + _url_check_func = is_notebook_url - @classmethod - def from_id(cls: type[T], id: int, /) -> T: - return cls(id=id) + _url_to_id_func = notebook_url_to_id + _id_to_url_func = notebook_id_to_url - @property - def id(self) -> int: - return self._id + def __init__(self, *, id: int | None = None, url: str | None = None) -> None: + IdAndUrlResourceMixin.__init__(self, id=id, url=url) + CheckableResourceMixin.__init__(self) - @property - def url(self) -> str: - return notebook_id_to_url(self._id) + def __repr__(self) -> str: + return IdAndUrlResourceMixin.__repr__(self) async def check(self) -> None: try: @@ -129,6 +125,8 @@ async def check(self) -> None: ) from None raise + else: + self._checked = True @property async def info(self) -> NotebookInfo: diff --git a/jkit/user.py b/jkit/user.py index 955fb4d..252935e 100644 --- a/jkit/user.py +++ b/jkit/user.py @@ -11,16 +11,16 @@ from httpx import HTTPStatusError from jkit._base import ( - CheckableMixin, + CheckableResourceMixin, DataObject, ResourceObject, - SlugAndUrlMixin, + SlugAndUrlResourceMixin, ) from jkit._network import send_request from jkit._normalization import normalize_assets_amount, normalize_datetime from jkit.constants import _RESOURCE_UNAVAILABLE_STATUS_CODE from jkit.exceptions import ResourceUnavailableError -from jkit.identifier_check import is_user_slug +from jkit.identifier_check import is_user_slug, is_user_url from jkit.identifier_convert import user_slug_to_url, user_url_to_slug from jkit.msgspec_constraints import ( ArticleSlug, @@ -149,19 +149,21 @@ def to_article_obj(self) -> Article: return Article.from_slug(self.slug)._as_checked() -class User(ResourceObject, CheckableMixin, SlugAndUrlMixin): +class User(ResourceObject, SlugAndUrlResourceMixin, CheckableResourceMixin): + _resource_readable_name = "用户" + _slug_check_func = is_user_slug - _slug_to_url_func = user_slug_to_url + _url_check_func = is_user_url + _url_to_slug_func = user_url_to_slug + _slug_to_url_func = user_slug_to_url def __init__(self, *, slug: str | None = None, url: str | None = None) -> None: - super().__init__() + SlugAndUrlResourceMixin.__init__(self, slug=slug, url=url) + CheckableResourceMixin.__init__(self) - self._slug = self._check_params( - object_readable_name="用户", - slug=slug, - url=url, - ) + def __repr__(self) -> str: + return SlugAndUrlResourceMixin.__repr__(self) async def check(self) -> None: try: @@ -178,6 +180,8 @@ async def check(self) -> None: ) from None raise + else: + self._checked = True @property async def id(self) -> int: