Skip to content

Commit

Permalink
refactor: 重构数据类与相关 Mixin
Browse files Browse the repository at this point in the history
  • Loading branch information
FHU-yezi committed Jan 26, 2025
1 parent 99201b7 commit 7e3eabc
Show file tree
Hide file tree
Showing 5 changed files with 177 additions and 156 deletions.
210 changes: 113 additions & 97 deletions jkit/_base.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -10,74 +10,54 @@
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:
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:
Expand All @@ -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)"
)
35 changes: 17 additions & 18 deletions jkit/article.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -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,
Expand Down Expand Up @@ -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:
Expand All @@ -237,6 +234,8 @@ async def check(self) -> None:
) from None

raise
else:
self._checked = True

@property
async def id(self) -> int:
Expand Down
Loading

0 comments on commit 7e3eabc

Please sign in to comment.