vftdan
/
mastoposter
forked from hkc/mastoposter
1
0
Fork 0
Code Pull Requests Activity

Fix codestyle & typehint

This commit is contained in:
Vftdan 2024-07-01 15:52:42 +02:00
parent d999ca14c4
commit 834a5a7fe2
Signed by: vftdan
GPG Key ID: 5D49CE378C2B20C7
1 changed files with 7 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
mastoposter/sources.py
View File

@ -69,14 +69,16 @@ async def websocket_source(
async def single_status_source( async def single_status_source(
status_url: str, url: str = None, access_token: str = None, retries: int = 5, **kwargs status_url: str, url: str = None, access_token: str = None,
retries: int = 5, **kwargs
) -> AsyncGenerator[Status, None]: ) -> AsyncGenerator[Status, None]:
# TODO: catch exceptions # TODO: catch exceptions
from httpx import Client, HTTPTransport from httpx import Client, HTTPTransport
user_authority = urlparse(url).netloc if url is not None else None user_authority = urlparse(url).netloc if url is not None else None
try: try:
status_url = f"https://{user_authority}/api/v1/statuses/{int(status_url)}" status_url = \
f"https://{user_authority}/api/v1/statuses/{int(status_url)}"
except ValueError: except ValueError:
pass pass
parsed_status_url = urlparse(status_url) parsed_status_url = urlparse(status_url)
@ -101,7 +103,8 @@ async def single_status_source(
) )
status = Status.from_dict(rq.json()) status = Status.from_dict(rq.json())
else: else:
search_instance = user_authority if user_authority is not None else parsed_status_url.netloc search_instance = user_authority if user_authority is not None \
else parsed_status_url.netloc
if search_instance != user_authority: if search_instance != user_authority:
access_token = None access_token = None
@ -115,7 +118,7 @@ async def single_status_source(
f"https://{search_instance}/api/v2/search", f"https://{search_instance}/api/v2/search",
params=params, params=params,
) )
statuses: List[Status] = rq.json().get("statuses", []) statuses: List[dict] = rq.json().get("statuses", [])
if len(statuses) < 1: if len(statuses) < 1:
logger.error("Instance %s hasn't found status %r", logger.error("Instance %s hasn't found status %r",
search_instance, status_url) search_instance, status_url)