kopia lustrzana https://github.com/ihabunek/toot
asyncio
Ivan Habunek
rodzic
1e3cca5204
commit
bebd8e9023
|
|
@ -0,0 +1,16 @@
|
|||
import click
|
||||
|
||||
from pkg_resources import iter_entry_points
|
||||
|
||||
[entry_point] = list(iter_entry_points('console_scripts', name="toot"))
|
||||
|
||||
cli = entry_point.resolve()
|
||||
print(cli)
|
||||
ctx = click.Context(cli)
|
||||
commands = getattr(cli, 'commands', {})
|
||||
|
||||
print(ctx)
|
||||
command: click.Command
|
||||
for name, command in cli.commands.items():
|
||||
print(name, command)
|
||||
print(command.help)
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
[pytest]
|
||||
asyncio_mode = auto
|
||||
addopts = -v
|
||||
|
|
@ -1,5 +1,7 @@
|
|||
faker
|
||||
flake8
|
||||
psycopg2-binary
|
||||
pytest
|
||||
pytest-asyncio
|
||||
pytest-xdist[psutil]
|
||||
vermin
|
||||
|
|
|
|||
2
setup.py
2
setup.py
|
|
@ -38,6 +38,8 @@ setup(
|
|||
"beautifulsoup4>=4.5.0,<5.0",
|
||||
"wcwidth>=0.1.7",
|
||||
"urwid>=2.0.0,<3.0",
|
||||
"aiohttp>=3.5.0,<4.0",
|
||||
"click~=8.1.3"
|
||||
],
|
||||
entry_points={
|
||||
'console_scripts': [
|
||||
|
|
|
|||
|
|
@ -0,0 +1,3 @@
|
|||
from toot.asynch.commands import cli
|
||||
|
||||
cli()
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
Toot Async Refactor
|
||||
===================
|
||||
|
||||
Moving to `aiohttp` for full async support + websockets for streaming.
|
||||
|
||||
Design goals:
|
||||
|
||||
- keep the functional API design
|
||||
- not coupled with underlying library: requests/aiohttp/httpx
|
||||
- can easily read returned data decoded from json
|
||||
- can read returned data without it being decoded (currently not possible)
|
||||
- can read returned headers
|
||||
- raises an exception on error
|
||||
- asynchronous by default, with synchronous variants
|
||||
- rewrite TUI to use async functions instead of the current callback hell
|
||||
|
||||
To think about:
|
||||
|
||||
- ability to reuse a aiohttp ClientSession without _having_ to do so. toot's own
|
||||
Session object? are there performance issues if a new session is creted each
|
||||
time?
|
||||
- couple app and user into one "context" object to avoid having to pass two
|
||||
params for each fn?
|
||||
- further namespace CLI commands? for example "mute" can be applied to both a
|
||||
status and an account, having "toot status mute" and "toot account mute"
|
||||
would solve it at expense of verbosity and breaking BC. alternatively, "toot
|
||||
mute" could act on a status or account.
|
||||
- how to implement updating credentials via commandline?
|
||||
https://mastodon.example/api/v1/accounts/update_credentials
|
||||
|
||||
Unrelated to async refactor:
|
||||
|
||||
- how to configure toot? global settings, per-server settings, ...
|
||||
- is it worth to switch to `click` for CLI, see:
|
||||
https://click.palletsprojects.com/en/8.1.x/why/#why-not-argparse
|
||||
|
||||
Yak shaving:
|
||||
|
||||
- update mastodon template for api methods to include `name="..."` so individual
|
||||
API endpoints can be linked in their API docs.
|
||||
https://github.com/mastodon/documentation/blob/master/layouts/shortcodes/api-method.html
|
||||
|
|
@ -0,0 +1,177 @@
|
|||
import uuid
|
||||
|
||||
from toot import CLIENT_NAME, CLIENT_WEBSITE, config, App, User
|
||||
from typing import Literal, List, Optional
|
||||
from toot.asynch.http import request, Params, Response
|
||||
from toot.utils import str_bool
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# Types
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
Visibility = Literal["public", "unlisted", "private", "direct"]
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# Accounts
|
||||
# https://docs.joinmastodon.org/methods/accounts/
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
async def register_account(
|
||||
app: App,
|
||||
auth_token: str,
|
||||
username: str,
|
||||
email: str,
|
||||
password: str,
|
||||
locale: str = "en",
|
||||
agreement: bool = True
|
||||
) -> Response:
|
||||
url = f"{app.base_url}/api/v1/accounts"
|
||||
headers = {"Authorization": f"Bearer {auth_token}"}
|
||||
|
||||
json = {
|
||||
"username": username,
|
||||
"email": email,
|
||||
"password": password,
|
||||
"agreement": agreement,
|
||||
"locale": locale
|
||||
}
|
||||
|
||||
return await request("POST", url, json=json, headers=headers)
|
||||
|
||||
|
||||
async def verify_credentials(app, user):
|
||||
return await auth_get(app, user, "/api/v1/accounts/verify_credentials")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# Apps
|
||||
# https://docs.joinmastodon.org/methods/apps/
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
async def create_app(domain: str, scheme: str = "https") -> Response:
|
||||
url = f"{scheme}://{domain}/api/v1/apps"
|
||||
|
||||
json = {
|
||||
"client_name": CLIENT_NAME,
|
||||
"redirect_uris": "urn:ietf:wg:oauth:2.0:oob",
|
||||
"scopes": "read write follow",
|
||||
"website": CLIENT_WEBSITE,
|
||||
}
|
||||
|
||||
return await request("POST", url, json=json)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# Instance
|
||||
# https://docs.joinmastodon.org/methods/instance/
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
async def instance_v1(url: str) -> Response:
|
||||
return await request("GET", f"{url}/api/v1/instance")
|
||||
|
||||
|
||||
async def instance_v2(url: str) -> Response:
|
||||
return await request("GET", f"{url}/api/v2/instance")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# Statuses
|
||||
# https://docs.joinmastodon.org/methods/statuses/
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
async def post_status(
|
||||
app: App,
|
||||
user: User,
|
||||
status: str,
|
||||
visibility: Visibility = "public",
|
||||
media_ids: Optional[List[int]] = None,
|
||||
sensitive: bool = False,
|
||||
spoiler_text: Optional[str] = None,
|
||||
in_reply_to_id: Optional[int] = None,
|
||||
language: Optional[str] = None,
|
||||
scheduled_at: Optional[str] = None,
|
||||
content_type: Optional[str] = None,
|
||||
) -> Response:
|
||||
"""
|
||||
Posts a new status.
|
||||
https://github.com/tootsuite/documentation/blob/master/Using-the-API/API.md#posting-a-new-status
|
||||
"""
|
||||
|
||||
# Idempotency key assures the same status is not posted multiple times
|
||||
# if the request is retried.
|
||||
headers = {"Idempotency-Key": uuid.uuid4().hex}
|
||||
|
||||
params = {
|
||||
"status": status,
|
||||
"media_ids[]": media_ids,
|
||||
"visibility": visibility,
|
||||
"sensitive": str_bool(sensitive),
|
||||
"spoiler_text": spoiler_text,
|
||||
"in_reply_to_id": in_reply_to_id,
|
||||
"language": language,
|
||||
"scheduled_at": scheduled_at
|
||||
}
|
||||
|
||||
if content_type:
|
||||
params["content_type"] = content_type
|
||||
|
||||
return await auth_request(app, user, "POST", "/api/v1/statuses", json=params, headers=headers)
|
||||
|
||||
|
||||
async def get_status(app: App, user: User, id: int) -> Response:
|
||||
return await auth_request(app, user, "GET", f"/api/v1/statuses/{id}")
|
||||
|
||||
|
||||
async def delete_status(app: App, user: User, id: int) -> Response:
|
||||
return await auth_request(app, user, "DELETE", f"/api/v1/statuses/{id}")
|
||||
|
||||
|
||||
async def timeline(app: App, user: User) -> Response:
|
||||
return await auth_request(app, user, "GET", "/api/v1/timelines/home")
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# OAuth
|
||||
# https://docs.joinmastodon.org/methods/apps/oauth/
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
async def app_token(app):
|
||||
json = {
|
||||
"client_id": app.client_id,
|
||||
"client_secret": app.client_secret,
|
||||
"grant_type": "client_credentials",
|
||||
"redirect_uri": "urn:ietf:wg:oauth:2.0:oob",
|
||||
"scope": "read write"
|
||||
}
|
||||
|
||||
return await request("POST", f"{app.base_url}/oauth/token", json=json)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# ???
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
async def search_accounts(app, user, query):
|
||||
return await auth_request(app, user, "GET", "/api/v1/accounts/search", params={"q": query})
|
||||
|
||||
|
||||
# ------------------------------------------------------------------------------
|
||||
# Common
|
||||
# ------------------------------------------------------------------------------
|
||||
|
||||
async def anon_get(url: str, params: Optional[Params] = None):
|
||||
return await request("GET", url, params=params)
|
||||
|
||||
|
||||
async def auth_get(app, user, path, params: Optional[Params] = None):
|
||||
url = app.base_url + path
|
||||
headers = {"Authorization": f"Bearer {user.access_token}"}
|
||||
return await request("GET", url, params=params, headers=headers)
|
||||
|
||||
|
||||
async def auth_request(app, user, method, path, /, *, headers={}, **kwargs):
|
||||
url = app.base_url + path
|
||||
headers.update({"Authorization": f"Bearer {user.access_token}"})
|
||||
return await request(method, url, headers=headers, **kwargs)
|
||||
|
|
@ -0,0 +1,227 @@
|
|||
import asyncio
|
||||
import click
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import sys
|
||||
|
||||
|
||||
from functools import wraps
|
||||
from typing import List, NamedTuple, Optional, Tuple
|
||||
|
||||
from toot import App, User, __version__, config
|
||||
from toot.asynch import api
|
||||
from toot.asynch.entities import Account, InstanceV2, Status, from_dict, from_response
|
||||
from toot.output import echo, print_out
|
||||
from toot.utils import EOF_KEY, editor_input, multiline_input
|
||||
|
||||
# Allow overriding options using environment variables
|
||||
# https://click.palletsprojects.com/en/8.1.x/options/?highlight=auto_env#values-from-environment-variables
|
||||
|
||||
# Tweak the Click context
|
||||
# https://click.palletsprojects.com/en/8.1.x/api/#context
|
||||
CONTEXT = dict(
|
||||
# Enable using environment variables to set options
|
||||
auto_envvar_prefix='TOOT',
|
||||
# Add shorthand -h for invoking help
|
||||
help_option_names=['-h', '--help'],
|
||||
# Give help some more room (default is 80)
|
||||
max_content_width=100,
|
||||
# Always show default values for options
|
||||
show_default=True,
|
||||
)
|
||||
|
||||
|
||||
def async_command(f):
|
||||
# Integrating click with asyncio:
|
||||
# https://github.com/pallets/click/issues/85#issuecomment-503464628
|
||||
@wraps(f)
|
||||
def wrapper(*args, **kwargs):
|
||||
return asyncio.run(f(*args, **kwargs))
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def validate_language(ctx, param, value: str) -> str:
|
||||
if value and len(value) != 3:
|
||||
raise click.BadParameter(
|
||||
"Expected a 3 letter abbreviation according to ISO 639-2 standard."
|
||||
)
|
||||
|
||||
return value
|
||||
|
||||
|
||||
# Data object to add to Click context
|
||||
class Obj(NamedTuple):
|
||||
app: Optional[App]
|
||||
user: Optional[User]
|
||||
color: bool
|
||||
debug: bool
|
||||
json: bool
|
||||
quiet: bool
|
||||
|
||||
|
||||
@click.group(context_settings=CONTEXT)
|
||||
@click.option("--debug/--no-debug", default=False, help="Log debug info to stderr")
|
||||
@click.option("--color/--no-color", default=sys.stdout.isatty(), help="Use ANSI color in output")
|
||||
@click.option("--quiet/--no-quiet", default=False, help="Don't print anything to stdout")
|
||||
@click.option("--json/--no-json", default=False, help="Print data as JSON rather than human readable textv")
|
||||
@click.version_option(version=__version__, prog_name="toot")
|
||||
@click.pass_context
|
||||
def cli(ctx, debug: bool, color: bool, quiet: bool, json: bool):
|
||||
user, app = config.get_active_user_app()
|
||||
ctx.color = color
|
||||
ctx.obj = Obj(app, user, color, debug, json, quiet)
|
||||
if debug:
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("url", required=False)
|
||||
@click.pass_context
|
||||
@async_command
|
||||
async def instance(ctx, url: Optional[str]):
|
||||
base_url = url or ctx.obj.app.base_url
|
||||
response = await api.instance_v2(base_url)
|
||||
|
||||
if ctx.obj.json:
|
||||
click.echo(response.body)
|
||||
else:
|
||||
instance = from_response(InstanceV2, response)
|
||||
click.secho(instance.title, fg="green")
|
||||
click.secho(url, fg="blue")
|
||||
click.echo(f"Running Mastodon {instance.version}")
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.pass_context
|
||||
@async_command
|
||||
async def whoami(ctx):
|
||||
response = await api.verify_credentials(ctx.obj.app, ctx.obj.user)
|
||||
|
||||
if ctx.obj.json:
|
||||
click.echo(response.body)
|
||||
else:
|
||||
account = from_response(Account, response)
|
||||
click.echo(click.style(account.acct, fg="green", bold=True))
|
||||
click.echo(click.style(account.display_name, fg="yellow"))
|
||||
click.echo(account.note_plaintext)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.pass_context
|
||||
@async_command
|
||||
async def timeline(ctx):
|
||||
response = await api.timeline(ctx.obj.app, ctx.obj.user)
|
||||
|
||||
if ctx.obj.json:
|
||||
click.echo(response.body)
|
||||
else:
|
||||
timeline = [from_dict(Status, s) for s in response.json()]
|
||||
for status in timeline:
|
||||
click.echo()
|
||||
click.echo(status.original.account.username)
|
||||
click.echo(status.original.content)
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.argument("text", required=False)
|
||||
@click.option(
|
||||
"-e", "--editor", is_flag=True,
|
||||
flag_value=os.environ.get("EDITOR"),
|
||||
show_default=os.environ.get("EDITOR"),
|
||||
help="""Use an editor to compose your toot, defaults to editor defined in
|
||||
the $EDITOR environment variable."""
|
||||
)
|
||||
@click.option(
|
||||
"-m", "--media", multiple=True,
|
||||
help="""Path to a media file to attach (specify multiple times to attach up
|
||||
to 4 files)""",
|
||||
)
|
||||
@click.option(
|
||||
"-d", "--description", multiple=True,
|
||||
help="""Plain-text description of the media for accessibility purposes, one
|
||||
per attached media""",
|
||||
)
|
||||
@click.option(
|
||||
"-l", "--language",
|
||||
help="ISO 639-2 language code of the toot, to skip automatic detection",
|
||||
callback=validate_language
|
||||
)
|
||||
def post(
|
||||
text: str,
|
||||
editor: str,
|
||||
media: Tuple[str, ...],
|
||||
description: Tuple[str, ...],
|
||||
language: Optional[str],
|
||||
):
|
||||
if editor and not sys.stdin.isatty():
|
||||
raise click.UsageError("Cannot run editor if not in tty.")
|
||||
|
||||
if media and len(media) > 4:
|
||||
raise click.UsageError("Cannot attach more than 4 files.")
|
||||
|
||||
echo("unstyled <red>posting</red> <dim>dim</dim> <underline><red>unde</red><blue>rline</blue></underline> <b>bold</b> unstlyed")
|
||||
echo("<bold>Bold<italic> bold and italic </bold>italic</italic>")
|
||||
echo("<bold red underline>foo</>bar")
|
||||
echo("\\<bold red underline>foo</>bar")
|
||||
echo("plain <blue>blue <underline> blue <green>and</green> underline </underline> blue </blue> plain")
|
||||
# echo("Done")
|
||||
# media_ids = _upload_media(app, user, args)
|
||||
# status_text = _get_status_text(text, editor)
|
||||
|
||||
# if not status_text and not media_ids:
|
||||
# raise click.UsageError("You must specify either text or media to post.")
|
||||
|
||||
# response = api.post_status(
|
||||
# app, user, status_text,
|
||||
# visibility=args.visibility,
|
||||
# media_ids=media_ids,
|
||||
# sensitive=args.sensitive,
|
||||
# spoiler_text=args.spoiler_text,
|
||||
# in_reply_to_id=args.reply_to,
|
||||
# language=args.language,
|
||||
# scheduled_at=args.scheduled_at,
|
||||
# content_type=args.content_type
|
||||
# )
|
||||
|
||||
# if "scheduled_at" in response:
|
||||
# print_out("Toot scheduled for: <green>{}</green>".format(response["scheduled_at"]))
|
||||
# else:
|
||||
# print_out("Toot posted: <green>{}</green>".format(response.get('url')))
|
||||
|
||||
|
||||
def _get_status_text(text, editor):
|
||||
isatty = sys.stdin.isatty()
|
||||
|
||||
if not text and not isatty:
|
||||
text = sys.stdin.read().rstrip()
|
||||
|
||||
if isatty:
|
||||
if editor:
|
||||
text = editor_input(editor, text)
|
||||
elif not text:
|
||||
print_out("Write or paste your toot. Press <yellow>{}</yellow> to post it.".format(EOF_KEY))
|
||||
text = multiline_input()
|
||||
|
||||
return text
|
||||
|
||||
|
||||
def _upload_media(app, user, args):
|
||||
# Match media to corresponding description and upload
|
||||
media = args.media or []
|
||||
descriptions = args.description or []
|
||||
uploaded_media = []
|
||||
|
||||
for idx, file in enumerate(media):
|
||||
description = descriptions[idx].strip() if idx < len(descriptions) else None
|
||||
result = _do_upload(app, user, file, description)
|
||||
uploaded_media.append(result)
|
||||
|
||||
return [m["id"] for m in uploaded_media]
|
||||
|
||||
|
||||
def _do_upload(app, user, file: str, description: Optional[str]):
|
||||
print("Faking upload:", file, description)
|
||||
id = random.randint(1, 99999)
|
||||
return {"id": id, "text_url": f"http://example.com/{id}"}
|
||||
|
|
@ -0,0 +1,289 @@
|
|||
import dataclasses
|
||||
|
||||
from dataclasses import dataclass, is_dataclass
|
||||
from datetime import date, datetime
|
||||
from typing import Dict, List, Optional, Type, TypeVar, Union
|
||||
from typing import get_type_hints
|
||||
|
||||
from toot.asynch.http import Response
|
||||
from toot.typing_compat import get_args, get_origin
|
||||
from toot.utils import get_text
|
||||
|
||||
|
||||
@dataclass
|
||||
class AccountField:
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Account/#Field
|
||||
"""
|
||||
name: str
|
||||
value: str
|
||||
verified_at: Optional[datetime]
|
||||
|
||||
|
||||
@dataclass
|
||||
class CustomEmoji:
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/CustomEmoji/
|
||||
"""
|
||||
shortcode: str
|
||||
url: str
|
||||
static_url: str
|
||||
visible_in_picker: bool
|
||||
category: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Account:
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Account/
|
||||
"""
|
||||
id: str
|
||||
username: str
|
||||
acct: str
|
||||
url: str
|
||||
display_name: str
|
||||
note: str
|
||||
avatar: str
|
||||
avatar_static: str
|
||||
header: str
|
||||
header_static: str
|
||||
locked: bool
|
||||
fields: List[AccountField]
|
||||
emojis: List[CustomEmoji]
|
||||
bot: bool
|
||||
group: bool
|
||||
discoverable: Optional[bool]
|
||||
noindex: Optional[bool]
|
||||
moved: Optional["Account"]
|
||||
suspended: Optional[bool]
|
||||
limited: Optional[bool]
|
||||
created_at: datetime
|
||||
last_status_at: Optional[date]
|
||||
statuses_count: int
|
||||
followers_count: int
|
||||
following_count: int
|
||||
|
||||
@property
|
||||
def note_plaintext(self) -> str:
|
||||
return get_text(self.note)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Application:
|
||||
name: str
|
||||
website: Optional[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class MediaAttachment:
|
||||
id: str
|
||||
type: str
|
||||
url: str
|
||||
preview_url: str
|
||||
remote_url: Optional[str]
|
||||
meta: dict
|
||||
description: str
|
||||
blurhash: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class StatusMention:
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Status/#Mention
|
||||
"""
|
||||
id: str
|
||||
username: str
|
||||
url: str
|
||||
acct: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class StatusTag:
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Status/#Tag
|
||||
"""
|
||||
name: str
|
||||
url: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class PollOption:
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Poll/#Option
|
||||
"""
|
||||
title: str
|
||||
votes_count: Optional[int]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Poll:
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Poll/
|
||||
"""
|
||||
id: str
|
||||
expires_at: Optional[datetime]
|
||||
expired: bool
|
||||
multiple: bool
|
||||
votes_count: int
|
||||
voters_count: Optional[int]
|
||||
options: List[PollOption]
|
||||
emojis: List[CustomEmoji]
|
||||
voted: Optional[bool]
|
||||
own_votes: Optional[List[int]]
|
||||
|
||||
|
||||
@dataclass
|
||||
class PreviewCard:
|
||||
url: str
|
||||
title: str
|
||||
description: str
|
||||
type: str
|
||||
author_name: str
|
||||
author_url: str
|
||||
provider_name: str
|
||||
provider_url: str
|
||||
html: str
|
||||
width: int
|
||||
height: int
|
||||
image: Optional[str]
|
||||
embed_url: str
|
||||
blurhash: Optional[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class FilterKeyword:
|
||||
id: str
|
||||
keyword: str
|
||||
whole_word: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class FilterStatus:
|
||||
id: str
|
||||
status_id: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Filter:
|
||||
id: str
|
||||
title: str
|
||||
context: List[str]
|
||||
expires_at: Optional[datetime]
|
||||
filter_action: str
|
||||
keywords: List[FilterKeyword]
|
||||
statuses: List[FilterStatus]
|
||||
|
||||
|
||||
@dataclass
|
||||
class FilterResult:
|
||||
filter: Filter
|
||||
keyword_matches: Optional[List[str]]
|
||||
status_matches: Optional[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Status:
|
||||
id: str
|
||||
uri: str
|
||||
created_at: datetime
|
||||
account: Account
|
||||
content: str
|
||||
visibility: str
|
||||
sensitive: bool
|
||||
spoiler_text: str
|
||||
media_attachments: List[MediaAttachment]
|
||||
application: Optional[Application]
|
||||
mentions: List[StatusMention]
|
||||
tags: List[StatusTag]
|
||||
emojis: List[CustomEmoji]
|
||||
reblogs_count: int
|
||||
favourites_count: int
|
||||
replies_count: int
|
||||
url: Optional[str]
|
||||
in_reply_to_id: Optional[str]
|
||||
in_reply_to_account_id: Optional[str]
|
||||
reblog: Optional["Status"]
|
||||
poll: Optional[Poll]
|
||||
card: Optional[PreviewCard]
|
||||
language: Optional[str]
|
||||
text: Optional[str]
|
||||
edited_at: Optional[datetime]
|
||||
favourited: Optional[bool]
|
||||
reblogged: Optional[bool]
|
||||
muted: Optional[bool]
|
||||
bookmarked: Optional[bool]
|
||||
pinned: Optional[bool]
|
||||
filtered: Optional[List[FilterResult]]
|
||||
|
||||
@property
|
||||
def original(self) -> "Status":
|
||||
return self.reblog or self
|
||||
|
||||
|
||||
@dataclass
|
||||
class InstanceV2:
|
||||
domain: str
|
||||
title: str
|
||||
version: str
|
||||
source_url: str
|
||||
description: str
|
||||
usage: dict # TODO expand
|
||||
thumbnail: dict # TODO expand
|
||||
languages: List[str]
|
||||
configuration: dict # TODO expand
|
||||
registrations: dict # TODO expand
|
||||
contact: dict # TODO expand
|
||||
rules: List[dict] # TODO expand
|
||||
|
||||
|
||||
# Generic data class instance
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
def from_dict(cls: Type[T], data: Dict) -> T:
|
||||
def _fields():
|
||||
hints = get_type_hints(cls)
|
||||
for field in dataclasses.fields(cls):
|
||||
default = field.default if field.default is not dataclasses.MISSING else None
|
||||
field_type = prune_optional(hints[field.name])
|
||||
value = data.get(field.name, default)
|
||||
yield convert(field_type, value)
|
||||
|
||||
return cls(*_fields())
|
||||
|
||||
|
||||
def from_response(cls: Type[T], response: Response) -> T:
|
||||
return from_dict(cls, response.json())
|
||||
|
||||
|
||||
def convert(field_type, value):
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
if field_type in [str, int, bool, dict]:
|
||||
return value
|
||||
|
||||
if field_type == datetime:
|
||||
return datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%f%z")
|
||||
|
||||
if field_type == date:
|
||||
return date.fromisoformat(value)
|
||||
|
||||
if get_origin(field_type) == list:
|
||||
(inner_type,) = get_args(field_type)
|
||||
return [convert(inner_type, x) for x in value]
|
||||
|
||||
if is_dataclass(field_type):
|
||||
return from_dict(field_type, value)
|
||||
|
||||
raise ValueError(f"Not implemented for type '{field_type}'")
|
||||
|
||||
|
||||
def prune_optional(field_type):
|
||||
"""For `Optional[<type>]` returnes the encapsulated `<type>`."""
|
||||
if get_origin(field_type) == Union:
|
||||
args = get_args(field_type)
|
||||
if len(args) == 2 and args[1] == type(None):
|
||||
return args[0]
|
||||
|
||||
return field_type
|
||||
|
|
@ -0,0 +1,80 @@
|
|||
import asyncio
|
||||
import logging
|
||||
import json
|
||||
|
||||
from http import HTTPStatus
|
||||
from dataclasses import dataclass
|
||||
from toot import __version__
|
||||
from typing import Mapping, Dict, Optional, Tuple
|
||||
from aiohttp import ClientSession, ClientResponse, TraceConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
Params = Dict[str, str]
|
||||
Headers = Dict[str, str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Response():
|
||||
body: str
|
||||
headers: Mapping[str, str]
|
||||
|
||||
def json(self):
|
||||
return json.loads(self.body)
|
||||
|
||||
|
||||
class ResponseError(Exception):
|
||||
"""Raised when the API retruns a response with status code >= 400."""
|
||||
def __init__(self, status_code, error, description):
|
||||
self.status_code = status_code
|
||||
self.error = error
|
||||
self.description = description
|
||||
|
||||
status_message = HTTPStatus(status_code).phrase
|
||||
msg = f"HTTP {status_code} {status_message}"
|
||||
msg += f". Error: {error}" if error else ""
|
||||
msg += f". Description: {description}" if description else ""
|
||||
super().__init__(msg)
|
||||
|
||||
|
||||
async def request(method, url, **kwargs) -> Response:
|
||||
common_headers = {"User-Agent": f"toot/{__version__}"}
|
||||
trace_config = logger_trace_config()
|
||||
|
||||
async with ClientSession(headers=common_headers, trace_configs=[trace_config]) as session:
|
||||
async with session.request(method, url, **kwargs) as response:
|
||||
if not response.ok:
|
||||
error, description = await get_error(response)
|
||||
raise ResponseError(response.status, error, description)
|
||||
|
||||
body = await response.text()
|
||||
return Response(body, response.headers)
|
||||
|
||||
|
||||
async def get_error(response: ClientResponse) -> Tuple[Optional[str], Optional[str]]:
|
||||
"""Attempt to extract the error and error description from response body.
|
||||
|
||||
See: https://docs.joinmastodon.org/entities/error/
|
||||
"""
|
||||
try:
|
||||
data = await response.json()
|
||||
return data.get("error"), data.get("error_description")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return None, None
|
||||
|
||||
|
||||
def logger_trace_config() -> TraceConfig:
|
||||
async def on_request_start(session, context, params):
|
||||
context.start = asyncio.get_event_loop().time()
|
||||
logger.debug(f">>> {params.method} {params.url}")
|
||||
|
||||
async def on_request_end(session, context, params):
|
||||
elapsed = round(100 * (asyncio.get_event_loop().time() - context.start))
|
||||
logger.debug(f"<<< {params.method} {params.url} HTTP {params.response.status} {elapsed}ms")
|
||||
|
||||
trace_config = TraceConfig()
|
||||
trace_config.on_request_start.append(on_request_start)
|
||||
trace_config.on_request_end.append(on_request_end)
|
||||
return trace_config
|
||||
|
|
@ -0,0 +1,287 @@
|
|||
import inspect
|
||||
import json
|
||||
import typing
|
||||
|
||||
from datetime import date, datetime
|
||||
from typing import List, Optional
|
||||
from typing import get_origin, get_args
|
||||
from functools import cache
|
||||
|
||||
from toot.utils import get_text
|
||||
|
||||
|
||||
@cache
|
||||
def get_type_hints_cached(cls):
|
||||
return typing.get_type_hints(cls)
|
||||
|
||||
|
||||
def prune_optional(hint):
|
||||
if get_origin(hint) == typing.Union:
|
||||
args = get_args(hint)
|
||||
if len(args) == 2 and args[1] == type(None):
|
||||
return args[0]
|
||||
|
||||
return hint
|
||||
|
||||
|
||||
class Entity:
|
||||
def __init__(self, json_data: str):
|
||||
self._json = json_data
|
||||
self._data = json.loads(json_data)
|
||||
|
||||
def __getattr__(self, name):
|
||||
hints = get_type_hints_cached(self.__class__)
|
||||
if name not in hints:
|
||||
raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
|
||||
|
||||
# TODO: read default value from field definition somehow
|
||||
default = None
|
||||
value = self._data.get(name, default)
|
||||
hint = prune_optional(hints[name])
|
||||
return self.convert(value, hint)
|
||||
|
||||
def __repr__(self):
|
||||
def _fields():
|
||||
hints = get_type_hints_cached(self.__class__)
|
||||
for name, hint in hints.items():
|
||||
hint = prune_optional(hints[name])
|
||||
value = self._data.get(name)
|
||||
if value is None:
|
||||
yield f"{name}=None"
|
||||
elif hint in [str, date, datetime]:
|
||||
yield f"{name}='{value}'"
|
||||
elif hint in [int, bool, dict]:
|
||||
yield f"{name}={value}"
|
||||
else:
|
||||
yield f"{name}=..."
|
||||
|
||||
name = self.__class__.__name__
|
||||
fields = ", ".join(_fields())
|
||||
return f"{name}({fields})"
|
||||
|
||||
@property
|
||||
def __dict__(self):
|
||||
return self._data
|
||||
|
||||
# TODO: override __dict__?
|
||||
# TODO: make readonly
|
||||
|
||||
# def __setattribute__(self, name):
|
||||
# raise Exception("Entities are read-only")
|
||||
|
||||
# def __delattribute__(self, name):
|
||||
# raise Exception("Entities are read-only")
|
||||
|
||||
def convert(self, value, hint):
|
||||
if hint in [str, int, bool, dict]:
|
||||
return value
|
||||
|
||||
if hint == datetime:
|
||||
return datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%f%z")
|
||||
|
||||
if hint == date:
|
||||
return date.fromisoformat(value)
|
||||
|
||||
if get_origin(hint) == list:
|
||||
(inner_hint,) = get_args(hint)
|
||||
return [self.convert(v, inner_hint) for v in value]
|
||||
|
||||
if inspect.isclass(hint) and issubclass(hint, Entity):
|
||||
return hint(value)
|
||||
|
||||
raise ValueError(f"hint??? {hint}")
|
||||
|
||||
|
||||
class AccountField(Entity):
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Account/#Field
|
||||
"""
|
||||
name: str
|
||||
value: str
|
||||
verified_at: Optional[datetime]
|
||||
|
||||
|
||||
class CustomEmoji(Entity):
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/CustomEmoji/
|
||||
"""
|
||||
shortcode: str
|
||||
url: str
|
||||
static_url: str
|
||||
visible_in_picker: bool
|
||||
category: str
|
||||
|
||||
|
||||
class Account(Entity):
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Account/
|
||||
"""
|
||||
id: str
|
||||
username: str
|
||||
acct: str
|
||||
url: str
|
||||
display_name: str
|
||||
note: str
|
||||
avatar: str
|
||||
avatar_static: str
|
||||
header: str
|
||||
header_static: str
|
||||
locked: bool
|
||||
fields: List[AccountField]
|
||||
emojis: List[CustomEmoji]
|
||||
bot: bool
|
||||
group: bool
|
||||
discoverable: Optional[bool]
|
||||
noindex: Optional[bool]
|
||||
moved: Optional["Account"]
|
||||
suspended: bool = False
|
||||
limited: bool = False
|
||||
created_at: datetime
|
||||
last_status_at: Optional[date]
|
||||
statuses_count: int
|
||||
followers_count: int
|
||||
following_count: int
|
||||
|
||||
@property
|
||||
def note_plaintext(self) -> str:
|
||||
return get_text(self.note)
|
||||
|
||||
|
||||
class Application(Entity):
|
||||
name: str
|
||||
website: str
|
||||
|
||||
|
||||
class MediaAttachment(Entity):
|
||||
id: str
|
||||
type: str
|
||||
url: str
|
||||
preview_url: str
|
||||
remote_url: Optional[str]
|
||||
meta: dict
|
||||
description: str
|
||||
blurhash: str
|
||||
|
||||
|
||||
class StatusMention(Entity):
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Status/#Mention
|
||||
"""
|
||||
id: str
|
||||
username: str
|
||||
url: str
|
||||
acct: str
|
||||
|
||||
|
||||
class StatusTag(Entity):
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Status/#Tag
|
||||
"""
|
||||
name: str
|
||||
url: str
|
||||
|
||||
|
||||
class PollOption(Entity):
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Poll/#Option
|
||||
"""
|
||||
title: str
|
||||
votes_count: Optional[int]
|
||||
|
||||
|
||||
class Poll(Entity):
|
||||
"""
|
||||
https://docs.joinmastodon.org/entities/Poll/
|
||||
"""
|
||||
id: str
|
||||
expires_at: Optional[datetime]
|
||||
expired: bool
|
||||
multiple: bool
|
||||
votes_count: int
|
||||
voters_count: Optional[int]
|
||||
options: List[PollOption]
|
||||
emojis: List[CustomEmoji]
|
||||
voted: Optional[bool]
|
||||
own_votes: Optional[List[int]]
|
||||
|
||||
|
||||
class PreviewCard(Entity):
|
||||
url: str
|
||||
title: str
|
||||
description: str
|
||||
type: str
|
||||
author_name: str
|
||||
author_url: str
|
||||
provider_name: str
|
||||
provider_url: str
|
||||
html: str
|
||||
width: int
|
||||
height: int
|
||||
image: Optional[str]
|
||||
embed_url: str
|
||||
blurhash: Optional[str]
|
||||
|
||||
|
||||
class FilterKeyword(Entity):
|
||||
id: str
|
||||
keyword: str
|
||||
whole_word: str
|
||||
|
||||
|
||||
class FilterStatus(Entity):
|
||||
id: str
|
||||
status_id: str
|
||||
|
||||
|
||||
class Filter(Entity):
|
||||
id: str
|
||||
title: str
|
||||
context: List[str]
|
||||
expires_at: Optional[datetime]
|
||||
filter_action: str
|
||||
keywords: List[FilterKeyword]
|
||||
statuses: List[FilterStatus]
|
||||
|
||||
|
||||
class FilterResult(Entity):
|
||||
filter: Filter
|
||||
keyword_matches: Optional[List[str]]
|
||||
status_matches: Optional[str]
|
||||
|
||||
|
||||
class Status(Entity):
|
||||
id: str
|
||||
uri: str
|
||||
created_at: datetime
|
||||
account: Account
|
||||
content: str
|
||||
visibility: str
|
||||
sensitive: bool
|
||||
spoiler_text: str
|
||||
media_attachments: List[MediaAttachment]
|
||||
application: Optional[Application]
|
||||
mentions: List[StatusMention]
|
||||
tags: List[StatusTag]
|
||||
emojis: List[CustomEmoji]
|
||||
reblogs_count: int
|
||||
favourites_count: int
|
||||
replies_count: int
|
||||
url: Optional[str]
|
||||
in_reply_to_id: Optional[str]
|
||||
in_reply_to_account_id: Optional[str]
|
||||
reblog: Optional["Status"]
|
||||
poll: Optional[Poll]
|
||||
card: Optional[PreviewCard]
|
||||
language: Optional[str]
|
||||
text: Optional[str]
|
||||
edited_at: Optional[datetime]
|
||||
favourited: bool = False
|
||||
reblogged: bool = False
|
||||
muted: bool = False
|
||||
bookmarked: bool = False
|
||||
pinned: bool = False
|
||||
filtered: List[FilterResult]
|
||||
|
||||
@property
|
||||
def original(self):
|
||||
return self.reblog or self
|
||||
|
|
@ -0,0 +1,50 @@
|
|||
from datetime import datetime
|
||||
import asyncio
|
||||
import httpx
|
||||
import aiohttp
|
||||
|
||||
|
||||
async def httpx_get(client, url):
|
||||
start = datetime.now()
|
||||
response = await client.get(url)
|
||||
response.raise_for_status()
|
||||
text = response.json()
|
||||
print("httpx ", url, datetime.now() - start)
|
||||
|
||||
|
||||
async def aiohttp_get(session, url):
|
||||
start = datetime.now()
|
||||
async with session.get(url) as response:
|
||||
text = await response.json()
|
||||
print("aiohttp", url, datetime.now() - start)
|
||||
|
||||
|
||||
urls = [
|
||||
"https://chaos.social/api/v1/instance",
|
||||
"https://chaos.social/api/v1/instance/peers",
|
||||
"https://chaos.social/api/v1/timelines/public",
|
||||
]
|
||||
|
||||
|
||||
async def test_httpx():
|
||||
start = datetime.now()
|
||||
async with httpx.AsyncClient() as client:
|
||||
for url in urls:
|
||||
for _ in range(3):
|
||||
await httpx_get(client, url)
|
||||
print("TOTAL", datetime.now() - start)
|
||||
|
||||
|
||||
async def test_aiohttp():
|
||||
start = datetime.now()
|
||||
async with aiohttp.ClientSession() as session:
|
||||
for url in urls:
|
||||
for _ in range(3):
|
||||
await aiohttp_get(session, url)
|
||||
print("TOTAL", datetime.now() - start)
|
||||
|
||||
|
||||
def run():
|
||||
asyncio.run(test_httpx())
|
||||
print("")
|
||||
asyncio.run(test_aiohttp())
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
|
|
@ -924,7 +925,8 @@ def main():
|
|||
user, app = config.get_active_user_app()
|
||||
|
||||
try:
|
||||
run_command(app, user, command_name, args)
|
||||
coro = run_command(app, user, command_name, args)
|
||||
asyncio.run(coro)
|
||||
except (ConsoleError, ApiError) as e:
|
||||
print_err(str(e))
|
||||
sys.exit(1)
|
||||
|
|
|
|||
|
|
@ -124,6 +124,14 @@ USE_ANSI_COLOR = use_ansi_color()
|
|||
QUIET = "--quiet" in sys.argv
|
||||
|
||||
|
||||
def echo(message, nl=True, err=False):
|
||||
import click
|
||||
ctx = click.get_current_context()
|
||||
if not ctx.obj.quiet:
|
||||
message = colorize(message)
|
||||
click.echo(message, nl=nl, err=err)
|
||||
|
||||
|
||||
def print_out(*args, **kwargs):
|
||||
if not QUIET:
|
||||
args = [colorize(a) if USE_ANSI_COLOR else strip_tags(a) for a in args]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,95 @@
|
|||
import asyncio
|
||||
import httpx
|
||||
import logging
|
||||
import urwid
|
||||
|
||||
from toot import App, User, config, __version__
|
||||
from urwid import font
|
||||
|
||||
from toot.tui.constants import PALETTE
|
||||
|
||||
logging.basicConfig(filename="debug.log", level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
urwid.set_encoding('UTF-8')
|
||||
|
||||
|
||||
class Toot:
|
||||
def __init__(self, user: User, app: App):
|
||||
logger.info("init")
|
||||
self.user = user
|
||||
self.app = app
|
||||
self.config = config.load_config()
|
||||
self.layout = loading_screen()
|
||||
|
||||
# Default max status length, updated on startup
|
||||
self.max_toot_chars = 500
|
||||
|
||||
def create_layout(self):
|
||||
self.txt = urwid.Text(u"Hello World")
|
||||
return urwid.Filler(self.txt, "top")
|
||||
|
||||
async def boot(self):
|
||||
logger.info("boot")
|
||||
await asyncio.gather(
|
||||
self.load_max_toot_chars()
|
||||
)
|
||||
logger.info(f"self.max_toot_chars: {self.max_toot_chars}")
|
||||
|
||||
async def load_max_toot_chars(self):
|
||||
"""Some instances may have a non-default limit on toot size."""
|
||||
instance = await get_instance(self.app.instance)
|
||||
if "max_toot_chars" in instance:
|
||||
self.max_toot_chars = instance["max_toot_chars"]
|
||||
|
||||
def run(self):
|
||||
asyncio_loop = asyncio.get_event_loop()
|
||||
main_loop = urwid.MainLoop(
|
||||
self.layout,
|
||||
event_loop=urwid.AsyncioEventLoop(loop=asyncio_loop),
|
||||
palette=PALETTE,
|
||||
unhandled_input=self.handle_keypress
|
||||
)
|
||||
self._boot_task = asyncio_loop.create_task(self.boot())
|
||||
main_loop.run()
|
||||
|
||||
def handle_keypress(self, key):
|
||||
if key.lower() == "q":
|
||||
raise urwid.ExitMainLoop()
|
||||
|
||||
|
||||
async def get_instance(domain):
|
||||
url = "https://{}/api/v1/instance".format(domain)
|
||||
logger.info(f">>> {url}")
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(url)
|
||||
return response.json()
|
||||
|
||||
|
||||
def main():
|
||||
user, app = config.get_active_user_app()
|
||||
if user and app:
|
||||
Toot(user, app).run()
|
||||
|
||||
|
||||
def loading_screen():
|
||||
# NB: Padding with width="clip" will convert the fixed BigText widget
|
||||
# to a flow widget so it can be used in a Pile.
|
||||
big_text = "Toot {}".format(__version__)
|
||||
big_text = urwid.BigText(("intro_bigtext", big_text), font.Thin6x6Font())
|
||||
big_text = urwid.Padding(big_text, align="center", width="clip")
|
||||
|
||||
contents = urwid.Pile([
|
||||
big_text,
|
||||
urwid.Divider(),
|
||||
urwid.Text([
|
||||
"Maintained by ",
|
||||
("intro_smalltext", "@ihabunek"),
|
||||
" and contributors"
|
||||
], align="center"),
|
||||
urwid.Divider(),
|
||||
urwid.Text(("intro_smalltext", "Loading toots..."), align="center"),
|
||||
])
|
||||
|
||||
return urwid.Filler(contents)
|
||||
Ładowanie…
Reference in New Issue