refactor: add exceptions.py module with structured exception hierarchy
- Create exceptions.py with 7 exception types: TwitterError, AuthenticationError,
RateLimitError, NotFoundError, NetworkError, QueryIdError, TwitterAPIError
- Remove inline TwitterAPIError from client.py, import from exceptions module
- Replace RuntimeError('Cannot resolve queryId') with QueryIdError
- Replace RuntimeError('User not found') with NotFoundError
- Update test assertion for new TwitterAPIError message format
This commit is contained in:
@@ -463,7 +463,7 @@ class TestTwitterAPIError:
|
|||||||
def test_stores_status_code(self):
|
def test_stores_status_code(self):
|
||||||
err = TwitterAPIError(429, "Rate limited")
|
err = TwitterAPIError(429, "Rate limited")
|
||||||
assert err.status_code == 429
|
assert err.status_code == 429
|
||||||
assert str(err) == "Rate limited"
|
assert "Rate limited" in str(err)
|
||||||
|
|
||||||
def test_is_runtime_error(self):
|
def test_is_runtime_error(self):
|
||||||
err = TwitterAPIError(500, "Server error")
|
err = TwitterAPIError(500, "Server error")
|
||||||
|
|||||||
@@ -33,6 +33,14 @@ from .constants import (
|
|||||||
get_user_agent,
|
get_user_agent,
|
||||||
sync_chrome_version,
|
sync_chrome_version,
|
||||||
)
|
)
|
||||||
|
from .exceptions import (
|
||||||
|
AuthenticationError,
|
||||||
|
NetworkError,
|
||||||
|
NotFoundError,
|
||||||
|
QueryIdError,
|
||||||
|
RateLimitError,
|
||||||
|
TwitterAPIError,
|
||||||
|
)
|
||||||
from .models import Author, Metrics, Tweet, TweetMedia, UserProfile
|
from .models import Author, Metrics, Tweet, TweetMedia, UserProfile
|
||||||
|
|
||||||
TimelineInstructionGetter = Callable[[Any], Any]
|
TimelineInstructionGetter = Callable[[Any], Any]
|
||||||
@@ -103,13 +111,7 @@ _cached_query_ids: Dict[str, str] = {}
|
|||||||
_bundles_scanned = False
|
_bundles_scanned = False
|
||||||
|
|
||||||
|
|
||||||
class TwitterAPIError(RuntimeError):
|
|
||||||
"""Represents HTTP/network errors from Twitter APIs."""
|
|
||||||
|
|
||||||
def __init__(self, status_code, message):
|
|
||||||
# type: (int, str) -> None
|
|
||||||
super().__init__(message)
|
|
||||||
self.status_code = status_code
|
|
||||||
|
|
||||||
def _best_chrome_target():
|
def _best_chrome_target():
|
||||||
# type: () -> str
|
# type: () -> str
|
||||||
@@ -299,7 +301,7 @@ def _resolve_query_id(operation_name, prefer_fallback=True):
|
|||||||
_cached_query_ids[operation_name] = fallback
|
_cached_query_ids[operation_name] = fallback
|
||||||
return fallback
|
return fallback
|
||||||
|
|
||||||
raise RuntimeError('Cannot resolve queryId for "%s"' % operation_name)
|
raise QueryIdError('Cannot resolve queryId for "%s"' % operation_name)
|
||||||
|
|
||||||
|
|
||||||
# Hard ceiling to prevent accidental massive fetches
|
# Hard ceiling to prevent accidental massive fetches
|
||||||
@@ -389,7 +391,7 @@ class TwitterClient:
|
|||||||
data = self._graphql_get("UserByScreenName", variables, features)
|
data = self._graphql_get("UserByScreenName", variables, features)
|
||||||
result = _deep_get(data, "data", "user", "result")
|
result = _deep_get(data, "data", "user", "result")
|
||||||
if not result:
|
if not result:
|
||||||
raise RuntimeError("User @%s not found" % screen_name)
|
raise NotFoundError("User @%s not found" % screen_name)
|
||||||
|
|
||||||
legacy = result.get("legacy", {})
|
legacy = result.get("legacy", {})
|
||||||
return UserProfile(
|
return UserProfile(
|
||||||
|
|||||||
45
twitter_cli/exceptions.py
Normal file
45
twitter_cli/exceptions.py
Normal file
@@ -0,0 +1,45 @@
|
|||||||
|
"""Custom exceptions for twitter-cli.
|
||||||
|
|
||||||
|
Provides a structured exception hierarchy for categorized error handling:
|
||||||
|
- Authentication failures
|
||||||
|
- API errors (rate-limit, not-found, forbidden)
|
||||||
|
- Network errors
|
||||||
|
- Query ID resolution failures
|
||||||
|
|
||||||
|
Modeled after bilibili-cli/xiaohongshu-cli exception patterns.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
|
||||||
|
class TwitterError(RuntimeError):
|
||||||
|
"""Base exception for twitter-cli errors."""
|
||||||
|
|
||||||
|
|
||||||
|
class AuthenticationError(TwitterError):
|
||||||
|
"""Raised when cookies are missing, expired, or invalid."""
|
||||||
|
|
||||||
|
|
||||||
|
class RateLimitError(TwitterError):
|
||||||
|
"""Raised when Twitter rate limits the request (HTTP 429)."""
|
||||||
|
|
||||||
|
|
||||||
|
class NotFoundError(TwitterError):
|
||||||
|
"""Raised when a user or tweet is not found."""
|
||||||
|
|
||||||
|
|
||||||
|
class NetworkError(TwitterError):
|
||||||
|
"""Raised when upstream network requests fail."""
|
||||||
|
|
||||||
|
|
||||||
|
class QueryIdError(TwitterError):
|
||||||
|
"""Raised when a GraphQL queryId cannot be resolved."""
|
||||||
|
|
||||||
|
|
||||||
|
class TwitterAPIError(TwitterError):
|
||||||
|
"""Raised on non-OK Twitter API responses with HTTP status + message."""
|
||||||
|
|
||||||
|
def __init__(self, status_code: int, message: str):
|
||||||
|
self.status_code = status_code
|
||||||
|
self.message = message
|
||||||
|
super().__init__("Twitter API error (HTTP %d): %s" % (status_code, message))
|
||||||
Reference in New Issue
Block a user