Compare commits

...

12 Commits

Author SHA1 Message Date
Daniel Schwarz a89dfe24fe
Merge 31bbb20324 into 4996da61e5 2024-04-17 20:49:58 +02:00
Ivan Habunek 4996da61e5
Add python version for pyright 2024-04-15 08:30:28 +02:00
Ivan Habunek 87acfb8ef4
Fix broken build 2024-04-14 09:05:09 +02:00
Ivan Habunek 927fdc3026
Improve types 2024-04-13 15:30:52 +02:00
Daniel Schwarz 31bbb20324 Make this fix compatible with latest master 2024-03-05 20:08:54 -05:00
Daniel Schwarz 9d59df6c7e Merge branch 'master' into asyncfix 2024-03-05 20:03:05 -05:00
Daniel Schwarz d21b2920cb Fix for compatibility with more recent versions of toot 2024-03-05 19:58:54 -05:00
Daniel Schwarz a5cd9d343c Merge branch 'asyncfix' of https://github.com/danschwarz/toot into asyncfix 2024-03-05 19:58:18 -05:00
Daniel Schwarz c30657dc24
Merge branch 'ihabunek:master' into asyncfix 2023-01-30 13:42:43 -05:00
Daniel Schwarz cb7cbd872a
Merge branch 'ihabunek:master' into asyncfix 2023-01-30 09:16:19 -05:00
Daniel Schwarz ecb9c75f2e React properly to 422: Validation Failed. Status has already been taken errors 2022-12-31 18:16:51 -05:00
Daniel Schwarz fe5b9d1a46 React properly to 422: Validation Failed. Status has already been taken errors 2022-12-13 12:45:07 -05:00
4 changed files with 68 additions and 46 deletions

View File

@ -83,3 +83,4 @@ packages=[
[tool.pyright]
include = ["toot"]
typeCheckingMode = "strict"
pythonVersion = "3.8"

View File

@ -8,7 +8,7 @@ from typing import BinaryIO, List, Optional
from urllib.parse import urlparse, urlencode, quote
from toot import App, User, http, CLIENT_NAME, CLIENT_WEBSITE
from toot.exceptions import ConsoleError
from toot.exceptions import ApiError, ConsoleError
from toot.utils import drop_empty_values, str_bool, str_bool_nullable
@ -53,8 +53,28 @@ def _tag_action(app, user, tag_name, action) -> Response:
return http.post(app, user, url)
def create_app(base_url):
url = f"{base_url}/api/v1/apps"
def _status_toggle_action(app, user, status_id, action, data=None):
url = '/api/v1/statuses/{}/{}'.format(status_id, action)
try:
response = http.post(app, user, url).json()
except ApiError as e:
# For "toggle" operations, Mastodon returns unhelpful
# 422: "Validation failed: Status has already been taken"
# responses when you try to bookmark a status already
# bookmarked, or favourite a status already favourited
# so we just swallow those errors here
if str(e) == "Validation failed: Status has already been taken":
response = None
else:
# not the error we expected; re-raise the exception
raise e
finally:
return response
def create_app(domain, scheme='https'):
url = f"{scheme}://{domain}/api/v1/apps"
json = {
'client_name': CLIENT_NAME,
@ -310,38 +330,40 @@ def delete_status(app, user, status_id):
def favourite(app, user, status_id):
return _status_action(app, user, status_id, 'favourite')
return _status_toggle_action(app, user, status_id, 'favourite')
def unfavourite(app, user, status_id):
return _status_action(app, user, status_id, 'unfavourite')
return _status_toggle_action(app, user, status_id, 'unfavourite')
def reblog(app, user, status_id, visibility="public"):
return _status_action(app, user, status_id, 'reblog', data={"visibility": visibility})
return _status_toggle_action(app, user, status_id, 'reblog', data={"visibility": visibility})
def unreblog(app, user, status_id):
return _status_action(app, user, status_id, 'unreblog')
return _status_toggle_action(app, user, status_id, 'unreblog')
def pin(app, user, status_id):
return _status_action(app, user, status_id, 'pin')
return _status_toggle_action(app, user, status_id, 'pin')
def unpin(app, user, status_id):
return _status_action(app, user, status_id, 'unpin')
return _status_toggle_action(app, user, status_id, 'unpin')
def bookmark(app, user, status_id):
return _status_action(app, user, status_id, 'bookmark')
return _status_toggle_action(app, user, status_id, 'bookmark')
def unbookmark(app, user, status_id):
return _status_action(app, user, status_id, 'unbookmark')
return _status_toggle_action(app, user, status_id, 'unbookmark')
def translate(app, user, status_id):
# don't use status_toggle_action for translate as this is
# not toggling anything server-side; it's a read only operation.
return _status_action(app, user, status_id, 'translate')

View File

@ -14,12 +14,18 @@ import typing as t
from dataclasses import dataclass, is_dataclass
from datetime import date, datetime
from functools import lru_cache
from typing import Any, Dict, Optional, Tuple, Type, TypeVar, Union
from typing import Any, Dict, NamedTuple, Optional, Type, TypeVar, Union
from typing import get_args, get_origin, get_type_hints
from toot.utils import get_text
from toot.utils.datetime import parse_datetime
# Generic data class instance
T = TypeVar("T")
# A dict decoded from JSON
Data = Dict[str, Any]
@dataclass
class AccountField:
@ -76,7 +82,7 @@ class Account:
source: Optional[dict]
@staticmethod
def __toot_prepare__(obj: Dict) -> Dict:
def __toot_prepare__(obj: Data) -> Data:
# Pleroma has not yet converted last_status_at from datetime to date
# so trim it here so it doesn't break when converting to date.
# See: https://git.pleroma.social/pleroma/pleroma/-/issues/1470
@ -266,7 +272,7 @@ class Status:
return self.reblog or self
@staticmethod
def __toot_prepare__(obj: Dict) -> Dict:
def __toot_prepare__(obj: Data) -> Data:
# Pleroma has a bug where created_at is set to an empty string.
# To avoid marking created_at as optional, which would require work
# because we count on it always existing, set it to current datetime.
@ -457,27 +463,25 @@ class List:
# see: https://git.pleroma.social/pleroma/pleroma/-/issues/2918
replies_policy: Optional[str]
# ------------------------------------------------------------------------------
# Generic data class instance
T = TypeVar("T")
class Field(NamedTuple):
name: str
type: Any
default: Any
class ConversionError(Exception):
"""Raised when conversion fails from JSON value to data class field."""
def __init__(
self,
data_class: Type,
field_name: str,
field_type: Type,
field_value: Optional[str]
):
def __init__(self, data_class: type, field: Field, field_value: Optional[str]):
super().__init__(
f"Failed converting field `{data_class.__name__}.{field_name}` "
+ f"of type `{field_type.__name__}` from value {field_value!r}"
f"Failed converting field `{data_class.__name__}.{field.name}` "
+ f"of type `{field.type.__name__}` from value {field_value!r}"
)
def from_dict(cls: Type[T], data: Dict) -> T:
def from_dict(cls: Type[T], data: Data) -> T:
"""Convert a nested dict into an instance of `cls`."""
# Apply __toot_prepare__ if it exists
prepare = getattr(cls, '__toot_prepare__', None)
@ -485,19 +489,19 @@ def from_dict(cls: Type[T], data: Dict) -> T:
data = prepare(data)
def _fields():
for name, type, default in get_fields(cls):
value = data.get(name, default)
converted = _convert_with_error_handling(cls, name, type, value)
yield name, converted
for field in _get_fields(cls):
value = data.get(field.name, field.default)
converted = _convert_with_error_handling(cls, field, value)
yield field.name, converted
return cls(**dict(_fields()))
@lru_cache(maxsize=100)
def get_fields(cls: Type) -> t.List[Tuple[str, Type, Any]]:
@lru_cache
def _get_fields(cls: type) -> t.List[Field]:
hints = get_type_hints(cls)
return [
(
Field(
field.name,
_prune_optional(hints[field.name]),
_get_default_value(field)
@ -506,11 +510,11 @@ def get_fields(cls: Type) -> t.List[Tuple[str, Type, Any]]:
]
def from_dict_list(cls: Type[T], data: t.List[Dict]) -> t.List[T]:
def from_dict_list(cls: Type[T], data: t.List[Data]) -> t.List[T]:
return [from_dict(cls, x) for x in data]
def _get_default_value(field):
def _get_default_value(field: dataclasses.Field):
if field.default is not dataclasses.MISSING:
return field.default
@ -520,21 +524,16 @@ def _get_default_value(field):
return None
def _convert_with_error_handling(
data_class: Type,
field_name: str,
field_type: Type,
field_value: Optional[str]
):
def _convert_with_error_handling(data_class: type, field: Field, field_value: Any) -> Any:
try:
return _convert(field_type, field_value)
return _convert(field.type, field_value)
except ConversionError:
raise
except Exception:
raise ConversionError(data_class, field_name, field_type, field_value)
raise ConversionError(data_class, field, field_value)
def _convert(field_type, value):
def _convert(field_type: Any, value: Any) -> Any:
if value is None:
return None
@ -557,7 +556,7 @@ def _convert(field_type, value):
raise ValueError(f"Not implemented for type '{field_type}'")
def _prune_optional(field_type: Type) -> Type:
def _prune_optional(field_type: type) -> type:
"""For `Optional[<type>]` returns the encapsulated `<type>`."""
if get_origin(field_type) == Union:
args = get_args(field_type)

View File

@ -173,5 +173,5 @@ LANGUAGES = {
}
def language_name(code):
def language_name(code: str) -> str:
return LANGUAGES.get(code, code)