Browse Source

poetry run poe format

tooling/add_formater
ScreamoBMO 2 years ago
parent
commit
a594c32ac6
  1. 2
      hexbear/__init__.py
  2. 18
      hexbear/v1/__init__.py
  3. 11
      hexbear/v1/admin.py
  4. 12
      hexbear/v1/categories.py
  5. 86
      hexbear/v1/comment.py
  6. 139
      hexbear/v1/community.py
  7. 116
      hexbear/v1/lib.py
  8. 25
      hexbear/v1/modlog.py
  9. 136
      hexbear/v1/post.py
  10. 56
      hexbear/v1/private_message.py
  11. 105
      hexbear/v1/reports.py
  12. 11
      hexbear/v1/search.py
  13. 77
      hexbear/v1/site.py
  14. 11
      hexbear/v1/sitemod.py
  15. 221
      hexbear/v1/user.py
  16. 28
      hexbear/v1/views.py
  17. 7
      hexbear/v1/ws.py
  18. 42
      poetry.lock
  19. 1
      pyproject.toml

2
hexbear/__init__.py

@ -1 +1 @@
from . import v1
from . import v1

18
hexbear/v1/__init__.py

@ -1,4 +1,16 @@
from . import (
admin, categories, comment, community, modlog, post, private_message,
reports, search, site, sitemod, user, views, lib
)
admin,
categories,
comment,
community,
modlog,
post,
private_message,
reports,
search,
site,
sitemod,
user,
views,
lib,
)

11
hexbear/v1/admin.py

@ -2,19 +2,22 @@ import typing as T
from .views import UserView
from . import lib
class AddAdminRequest(T.NamedTuple):
user_id: int
added: bool
auth: str
class AddAdminResponse(T.NamedTuple):
admins: list[UserView]
async def add(payload: AddAdminRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/admin/add',
"post",
endpoint="/api/v1/admin/add",
payload=payload,
response_type=AddAdminResponse,
base_url=base_url
)
base_url=base_url,
)

12
hexbear/v1/categories.py

@ -1,21 +1,25 @@
import typing as T
from . import lib
class Category(T.NamedTuple):
id: int
name: str
class ListCategoriesRequest(T.NamedTuple):
pass
class ListCategoriesResponse(T.NamedTuple):
categories: list[Category]
async def get_list(payload: ListCategoriesRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/categories',
"get",
endpoint="/api/v1/categories",
payload=payload,
response_type=ListCategoriesResponse,
base_url=base_url
)
base_url=base_url,
)

86
hexbear/v1/comment.py

@ -3,11 +3,13 @@ from uuid import UUID
from . import lib
from .views import CommentView
class CommentResponse(T.NamedTuple):
comment: CommentView
recipient_ids: list[int]
form_id: T.Optional[str]
class CreateCommentRequest(T.NamedTuple):
content: str
parent_id: T.Optional[int]
@ -15,14 +17,16 @@ class CreateCommentRequest(T.NamedTuple):
form_id: T.Optional[str]
auth: str
async def create(payload: CreateCommentRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/comment',
"post",
endpoint="/api/v1/comment",
payload=payload,
response_type=CommentResponse,
base_url=base_url
)
base_url=base_url,
)
class EditCommentRequest(T.NamedTuple):
content: str
@ -30,28 +34,32 @@ class EditCommentRequest(T.NamedTuple):
form_id: T.Optional[str]
auth: str
async def edit(payload: EditCommentRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/comment',
"put",
endpoint="/api/v1/comment",
payload=payload,
response_type=CommentResponse,
base_url=base_url
)
base_url=base_url,
)
class DeleteCommentRequest(T.NamedTuple):
edit_id: int
deleted: bool
auth: str
async def delete(payload: DeleteCommentRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/comment/delete',
"post",
endpoint="/api/v1/comment/delete",
payload=payload,
response_type=CommentResponse,
base_url=base_url
)
base_url=base_url,
)
class RemoveCommentRequest(T.NamedTuple):
edit_id: int
@ -59,73 +67,83 @@ class RemoveCommentRequest(T.NamedTuple):
reason: T.Optional[str]
auth: str
async def remove(payload: RemoveCommentRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/comment/remove',
"post",
endpoint="/api/v1/comment/remove",
payload=payload,
response_type=CommentResponse,
base_url=base_url
)
base_url=base_url,
)
class MarkCommentAsReadRequest(T.NamedTuple):
edit_id: int
read: bool
auth: str
async def mark_as_read(payload: MarkCommentAsReadRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/comment/mark_as_read',
"post",
endpoint="/api/v1/comment/mark_as_read",
payload=payload,
response_type=CommentResponse,
base_url=base_url
)
base_url=base_url,
)
class CreateCommentLikeRequest(T.NamedTuple):
comment_id: int
score: int
auth: str
async def create_like(payload: CreateCommentLikeRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/comment/like',
"post",
endpoint="/api/v1/comment/like",
payload=payload,
response_type=CommentResponse,
base_url=base_url
)
base_url=base_url,
)
class SaveCommentRequest(T.NamedTuple):
comment_id: int
save: bool
auth: str
async def save(payload: SaveCommentRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/comment/save',
"put",
endpoint="/api/v1/comment/save",
payload=payload,
response_type=CommentResponse,
base_url=base_url
)
base_url=base_url,
)
class GetCommentsRequest(T.NamedTuple):
type_: lib.ListingType = 'All'
sort: lib.SortType = 'New'
type_: lib.ListingType = "All"
sort: lib.SortType = "New"
page: T.Optional[int] = None
limit: T.Optional[int] = None
community_id: T.Optional[int] = None
auth: T.Optional[str] = None
class GetCommentsResponse(T.NamedTuple):
comments: list[CommentView]
async def get_comments(payload: GetCommentsRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/comment/list',
"get",
endpoint="/api/v1/comment/list",
payload=payload,
response_type=GetCommentsResponse,
base_url=base_url
)
base_url=base_url,
)

139
hexbear/v1/community.py

@ -3,6 +3,7 @@ from datetime import datetime
from . import lib
from .views import UserView, CommunityView, CommunityModeratorView
class CreateCommunityRequest(T.NamedTuple):
name: str
title: str
@ -13,38 +14,44 @@ class CreateCommunityRequest(T.NamedTuple):
nsfw: bool
auth: str
class CommunityResponse(T.NamedTuple):
community: CommunityView
async def create(payload: CreateCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/community',
"post",
endpoint="/api/v1/community",
payload=payload,
response_type=CommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class GetCommunityRequest(T.NamedTuple):
id: T.Optional[int] = None
name: T.Optional[str] = None
auth: T.Optional[str] = None
class GetCommunityResponse(T.NamedTuple):
community: CommunityView
moderators: list[CommunityModeratorView]
online: int
admins: list[UserView] # hexbear
sitemods: list[UserView] # hexbear
admins: list[UserView] # hexbear
sitemods: list[UserView] # hexbear
async def get(payload: GetCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/community',
"get",
endpoint="/api/v1/community",
payload=payload,
response_type=GetCommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class EditCommunityRequest(T.NamedTuple):
edit_id: int
@ -56,14 +63,16 @@ class EditCommunityRequest(T.NamedTuple):
nsfw: bool
auth: str
async def edit(payload: EditCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/community',
"put",
endpoint="/api/v1/community",
payload=payload,
response_type=CommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class ListCommunitiesRequest(T.NamedTuple):
sort: lib.SortType
@ -71,45 +80,52 @@ class ListCommunitiesRequest(T.NamedTuple):
limit: T.Optional[int] = None
auth: T.Optional[str] = None
class ListCommunitiesResponse(T.NamedTuple):
communities: list[CommunityView]
async def get_list(payload: ListCommunitiesRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/community/list',
"get",
endpoint="/api/v1/community/list",
payload=payload,
response_type=ListCommunitiesResponse,
base_url=base_url
)
base_url=base_url,
)
class FollowCommunityRequest(T.NamedTuple):
community_id: int
follow: bool
auth: str
async def follow(payload: FollowCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/community/follow',
"post",
endpoint="/api/v1/community/follow",
payload=payload,
response_type=CommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class DeleteCommunityRequest(T.NamedTuple):
edit_id: int
deleted: bool
auth: str
async def delete(payload: DeleteCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/community/delete',
"post",
endpoint="/api/v1/community/delete",
payload=payload,
response_type=CommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class RemoveCommunityRequest(T.NamedTuple):
edit_id: int
@ -118,28 +134,32 @@ class RemoveCommunityRequest(T.NamedTuple):
expires: T.Optional[int]
auth: str
async def remove(payload: RemoveCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/community/remove',
"post",
endpoint="/api/v1/community/remove",
payload=payload,
response_type=CommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class TransferCommunityRequest(T.NamedTuple):
community_id: int
user_id: int
auth: str
async def transfer(payload: TransferCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/community/transfer',
"post",
endpoint="/api/v1/community/transfer",
payload=payload,
response_type=CommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class BanFromCommunityRequest(T.NamedTuple):
community_id: int
@ -150,18 +170,21 @@ class BanFromCommunityRequest(T.NamedTuple):
expires: T.Optional[int]
auth: str
class BanFromCommunityResponse(T.NamedTuple):
user: UserView
banned: bool
async def ban_user(payload: BanFromCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/community/ban_user',
"post",
endpoint="/api/v1/community/ban_user",
payload=payload,
response_type=BanFromCommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class AddModToCommunityRequest(T.NamedTuple):
community_id: int
@ -169,23 +192,27 @@ class AddModToCommunityRequest(T.NamedTuple):
added: bool
auth: str
class AddModToCommunityResponse(T.NamedTuple):
moderators: list[CommunityModeratorView]
# TODO Should be called "appoint mod"?
async def add_mod(payload: AddModToCommunityRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/community/mod',
"post",
endpoint="/api/v1/community/mod",
payload=payload,
response_type=AddModToCommunityResponse,
base_url=base_url
)
base_url=base_url,
)
class GetCommunitySettingsRequest(T.NamedTuple):
community_id: int
auth: T.Optional[str] = None
class GetCommunitySettingsResponse(T.NamedTuple):
read_only: bool
private: bool
@ -194,14 +221,18 @@ class GetCommunitySettingsResponse(T.NamedTuple):
published: datetime
allow_as_default: bool
async def get_settings(payload: GetCommunitySettingsRequest, base_url: str = lib.BASE_URL):
async def get_settings(
payload: GetCommunitySettingsRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'get',
endpoint='/api/v1/community/settings',
"get",
endpoint="/api/v1/community/settings",
payload=payload,
response_type=GetCommunitySettingsResponse,
base_url=base_url
)
base_url=base_url,
)
class EditCommunitySettingsRequest(T.NamedTuple):
community_id: int
@ -212,6 +243,7 @@ class EditCommunitySettingsRequest(T.NamedTuple):
allow_as_default: bool
auth: str
class EditCommunitySettingsResponse(T.NamedTuple):
read_only: bool
private: bool
@ -220,11 +252,14 @@ class EditCommunitySettingsResponse(T.NamedTuple):
published: datetime
allow_as_default: bool
async def edit_settings(payload: EditCommunitySettingsRequest, base_url: str = lib.BASE_URL):
async def edit_settings(
payload: EditCommunitySettingsRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'put',
endpoint='/api/v1/community/settings',
"put",
endpoint="/api/v1/community/settings",
payload=payload,
response_type=EditCommunitySettingsResponse,
base_url=base_url
)
base_url=base_url,
)

116
hexbear/v1/lib.py

@ -8,71 +8,60 @@ from typing_inspect import is_optional_type, is_generic_type, get_origin, get_ar
from types import GenericAlias
from datetime import datetime
class APIError(T.NamedTuple):
message: str
SortType = T.Literal[
'Active',
'Hot',
'New',
'TopDay',
'TopWeek',
'TopMonth',
'TopYear',
'TopAll'
"Active", "Hot", "New", "TopDay", "TopWeek", "TopMonth", "TopYear", "TopAll"
]
ListingType = T.Literal[
'All',
'Local',
'Subscribed',
'Community'
]
ListingType = T.Literal["All", "Local", "Subscribed", "Community"]
SearchType = T.Literal[
'All',
'Comments',
'Posts',
'Communities',
'Users',
'Url'
]
SearchType = T.Literal["All", "Comments", "Posts", "Communities", "Users", "Url"]
ListingType = T.Literal[
'All',
'Local',
'Subscribed',
'Community',
"All",
"Local",
"Subscribed",
"Community",
]
Method = T.Literal['get', 'post', 'put']
R = T.TypeVar('R')
Method = T.Literal["get", "post", "put"]
R = T.TypeVar("R")
BASE_URL: T.Final[str] = "https://www.chapo.chat"
BASE_URL: T.Final[str] = 'https://www.chapo.chat'
async def request(
method: Method,
*,
endpoint: str,
payload: T.NamedTuple,
response_type: R,
base_url: str = BASE_URL
) -> Result[R, str]:
method: Method,
*,
endpoint: str,
payload: T.NamedTuple,
response_type: R,
base_url: str = BASE_URL,
) -> Result[R, str]:
"""Make API call and parse response"""
payload = {
key: str(value).lower() if type(value) == bool else value # TODO Requests didn't automatically make booleans lowercase. Does aiohttp do that?
key: str(value).lower()
if type(value) == bool
else value # TODO Requests didn't automatically make booleans lowercase. Does aiohttp do that?
for key, value in payload._asdict().items()
if value is not None
}
url = f'{base_url}{endpoint}'
if method not in ('get', 'put', 'post'):
}
url = f"{base_url}{endpoint}"
if method not in ("get", "put", "post"):
return Err(f'Method must be "get", "put", or "post", not "{method}"')
async with aiohttp.ClientSession(raise_for_status=True) as session:
try:
async with (session.get(url, params=payload) if method == 'get'
else session.put(url, data=payload) if method == 'put'
else session.post(url, json=payload)
) as response:
async with (
session.get(url, params=payload)
if method == "get"
else session.put(url, data=payload)
if method == "put"
else session.post(url, json=payload)
) as response:
try:
parsed = await response.json()
except ValueError as e:
@ -85,29 +74,36 @@ async def request(
"If you're sure your site's certificate is installed properly, "
"try navigating to your Python folder and running "
"'Install Certificates.command'"
)
)
return ensure_type(response_type, parsed)
def is_typed_named_tuple(t):
return hasattr(t, '_field_defaults') and hasattr(t, '__annotations__')
return hasattr(t, "_field_defaults") and hasattr(t, "__annotations__")
def ensure_type(
type_: T.Callable[..., R],
obj: T.Any
) -> Result[R, str]:
def ensure_type(type_: T.Callable[..., R], obj: T.Any) -> Result[R, str]:
"""Recursively parse json into appropriate types"""
if isinstance(obj, dict) and 'error' in obj:
return Err(obj['error'].get('message', 'Got error response, but error contained no message'))
if isinstance(obj, dict) and "error" in obj:
return Err(
obj["error"].get(
"message", "Got error response, but error contained no message"
)
)
if is_typed_named_tuple(type_):
if not isinstance(obj, dict):
return Err(f'Expected dict for NamedTuple, got {type(obj)}')
field_types = type_.__annotations__ # _type._field_types doesn't exist on NamedTuples when running tests for some reason
return Err(f"Expected dict for NamedTuple, got {type(obj)}")
field_types = (
type_.__annotations__
) # _type._field_types doesn't exist on NamedTuples when running tests for some reason
defaults = type_._field_defaults
obj_with_defaults = ChainMap(obj, defaults)
missing_fields = list(set(field_types.keys()) - set(obj_with_defaults.keys()))
if len(missing_fields) > 0:
return Err(f'Missing required field encountered while parsing: {missing_fields[0]}')
return Err(
f"Missing required field encountered while parsing: {missing_fields[0]}"
)
results = {}
for key, item in obj_with_defaults.items():
subtype = field_types[key]
@ -128,9 +124,9 @@ def ensure_type(
if isinstance(type_, GenericAlias):
origin = type_.__origin__
if origin != list:
return Err(f'Unhandled GenericAlias type: {origin}')
return Err(f"Unhandled GenericAlias type: {origin}")
elif not isinstance(obj, list):
return Err(f'Type mismatch: {type_} and {type(obj)}')
return Err(f"Type mismatch: {type_} and {type(obj)}")
else:
subtype = type_.__args__[0]
results = []
@ -142,7 +138,9 @@ def ensure_type(
results.append(r.value)
return Ok(results)
if isinstance(obj, type_): # Has to go at the end because GenericAliases can't be checked this way and we don't want to pass through dicts with 'error' elements through as Ok's
if isinstance(
obj, type_
): # Has to go at the end because GenericAliases can't be checked this way and we don't want to pass through dicts with 'error' elements through as Ok's
return Ok(obj)
if type_ == datetime:
@ -151,4 +149,4 @@ def ensure_type(
except ValueError as e:
return Err(str(e))
return Err(f'Unhandled type mismatch: {type_} and {type(obj)}')
return Err(f"Unhandled type mismatch: {type_} and {type(obj)}")

25
hexbear/v1/modlog.py

@ -2,10 +2,17 @@ import typing as T
from datetime import datetime
from . import lib
from .views import (
ModRemovePostView, ModLockPostView, ModStickyPostView,
ModRemoveCommentView, ModRemoveCommunityView, ModBanFromCommunityView,
ModBanView, ModAddCommunityView, ModAddView
)
ModRemovePostView,
ModLockPostView,
ModStickyPostView,
ModRemoveCommentView,
ModRemoveCommunityView,
ModBanFromCommunityView,
ModBanView,
ModAddCommunityView,
ModAddView,
)
class GetModlogRequest(T.NamedTuple):
mod_user_id: T.Optional[int] = None
@ -14,6 +21,7 @@ class GetModlogRequest(T.NamedTuple):
limit: T.Optional[int] = None
auth: T.Optional[str] = None # hexbear
class GetModlogResponse(T.NamedTuple):
removed_posts: list[ModRemovePostView]
locked_posts: list[ModLockPostView]
@ -25,11 +33,12 @@ class GetModlogResponse(T.NamedTuple):
added_to_community: list[ModAddCommunityView]
added: list[ModAddView]
async def get(payload: GetModlogRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/modlog',
"get",
endpoint="/api/v1/modlog",
payload=payload,
response_type=GetModlogResponse,
base_url=base_url
)
base_url=base_url,
)

136
hexbear/v1/post.py

@ -1,8 +1,13 @@
import typing as T
from . import lib
from .views import (
PostView, CommunityView, UserView, CommunityModeratorView, CommentView
)
PostView,
CommunityView,
UserView,
CommunityModeratorView,
CommentView,
)
class CreatePostRequest(T.NamedTuple):
name: str
@ -12,39 +17,45 @@ class CreatePostRequest(T.NamedTuple):
community_id: int
auth: str
class PostResponse(T.NamedTuple):
post: PostView
async def create(payload: CreatePostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/post',
"post",
endpoint="/api/v1/post",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class GetPostRequest(T.NamedTuple):
id: int
auth: T.Optional[str] = None
class GetPostResponse(T.NamedTuple):
post: PostView
comments: list[CommentView]
community: CommunityView
moderators: list[CommunityModeratorView]
online: int
admins: list[UserView] # Hexbear
sitemods: list[UserView] # Hexbear
admins: list[UserView] # Hexbear
sitemods: list[UserView] # Hexbear
async def get(payload: GetPostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/post',
"get",
endpoint="/api/v1/post",
payload=payload,
response_type=GetPostResponse,
base_url=base_url
)
base_url=base_url,
)
class EditPostRequest(T.NamedTuple):
edit_id: int
@ -54,28 +65,32 @@ class EditPostRequest(T.NamedTuple):
nsfw: bool
auth: str
async def edit(payload: EditPostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/post',
"put",
endpoint="/api/v1/post",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class DeletePostRequest(T.NamedTuple):
edit_id: int
deleted: bool
auth: str
async def delete(payload: DeletePostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/post/delete',
"post",
endpoint="/api/v1/post/delete",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class RemovePostRequest(T.NamedTuple):
edit_id: int
@ -83,42 +98,48 @@ class RemovePostRequest(T.NamedTuple):
reason: T.Optional[str]
auth: str
async def remove(payload: RemovePostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/post/remove',
"post",
endpoint="/api/v1/post/remove",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class LockPostRequest(T.NamedTuple):
edit_id: int
locked: bool
auth: str
async def lock(payload: LockPostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/post/lock',
"post",
endpoint="/api/v1/post/lock",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class StickyPostRequest(T.NamedTuple):
edit_id: int
stickied: bool
auth: str
async def sticky(payload: StickyPostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/post/sticky',
"post",
endpoint="/api/v1/post/sticky",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class GetPostsRequest(T.NamedTuple):
type_: str
@ -129,71 +150,82 @@ class GetPostsRequest(T.NamedTuple):
community_name: T.Optional[str]
auth: T.Optional[str]
class GetPostsResponse(T.NamedTuple):
posts: list[PostView]
async def feed(payload: GetPostsRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/post/list',
"get",
endpoint="/api/v1/post/list",
payload=payload,
response_type=GetPostsResponse,
base_url=base_url
)
base_url=base_url,
)
class CreatePostLikeRequest(T.NamedTuple):
post_id: int
score: int
auth: str
async def create_like(payload: CreatePostLikeRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/post/like',
"post",
endpoint="/api/v1/post/like",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class SavePostRequest(T.NamedTuple):
post_id: int
save: bool
auth: str
async def save(payload: SavePostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/post/save',
"put",
endpoint="/api/v1/post/save",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)
class GetFeaturedPostsRequest(T.NamedTuple):
auth: T.Optional[str] = None
class GetFeaturedPostsResponse(T.NamedTuple):
posts: list[PostView]
async def list_featured(payload: GetFeaturedPostsRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/post/featured',
"get",
endpoint="/api/v1/post/featured",
payload=payload,
response_type=GetFeaturedPostsResponse,
base_url=base_url
)
base_url=base_url,
)
class FeaturePostRequest(T.NamedTuple):
id: int
featured: bool
auth: str
async def feature(payload: FeaturePostRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/post/featured',
"put",
endpoint="/api/v1/post/featured",
payload=payload,
response_type=PostResponse,
base_url=base_url
)
base_url=base_url,
)

56
hexbear/v1/private_message.py

@ -2,36 +2,42 @@ import typing as T
from . import lib
from .views import PrivateMessageView
class CreatePrivateMessageRequest(T.NamedTuple):
content: str
recipient_id: int
auth: str
class PrivateMessageResponse(T.NamedTuple):
message: PrivateMessageView
async def create(payload: CreatePrivateMessageRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/private_message',
"post",
endpoint="/api/v1/private_message",
payload=payload,
response_type=PrivateMessageResponse,
base_url=base_url
)
base_url=base_url,
)
class EditPrivateMessageRequest(T.NamedTuple):
edit_id: int
content: str
auth: str
async def edit(payload: EditPrivateMessageRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/private_message',
"put",
endpoint="/api/v1/private_message",
payload=payload,
response_type=PrivateMessageResponse,
base_url=base_url
)
base_url=base_url,
)
class GetPrivateMessagesRequest(T.NamedTuple):
unread_only: bool
@ -39,42 +45,50 @@ class GetPrivateMessagesRequest(T.NamedTuple):
limit: T.Optional[int]
auth: str
class PrivateMessagesResponse(T.NamedTuple):
messages: list[PrivateMessageView]
async def get_list(payload: GetPrivateMessagesRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/private_message/list',
"get",
endpoint="/api/v1/private_message/list",
payload=payload,
response_type=PrivateMessagesResponse,
base_url=base_url
)
base_url=base_url,
)
class DeletePrivateMessageRequest(T.NamedTuple):
edit_id: int
deleted: bool
auth: str
async def delete(payload: DeletePrivateMessageRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/private_message/delete',
"post",
endpoint="/api/v1/private_message/delete",
payload=payload,
response_type=PrivateMessageResponse,
base_url=base_url
)
base_url=base_url,
)
class MarkPrivateMessageAsReadRequest(T.NamedTuple):
edit_id: int
read: bool
auth: str
async def mark_as_read(payload: MarkPrivateMessageAsReadRequest, base_url: str = lib.BASE_URL):
async def mark_as_read(
payload: MarkPrivateMessageAsReadRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'post',
endpoint='/api/v1/private_message/mark_as_read',
"post",
endpoint="/api/v1/private_message/mark_as_read",
payload=payload,
response_type=PrivateMessageResponse,
base_url=base_url
)
base_url=base_url,
)

105
hexbear/v1/reports.py

@ -3,39 +3,50 @@ from uuid import UUID
from . import lib
from .views import CommentReportView, PostReportView
class CreateCommentReportRequest(T.NamedTuple):
comment: int
auth: str
reason: T.Optional[str] = None
class CommentReportResponse(T.NamedTuple):
success: bool
async def create_comment_report(payload: CreateCommentReportRequest, base_url: str = lib.BASE_URL):
async def create_comment_report(
payload: CreateCommentReportRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'post',
endpoint='/api/v1/comment/report',
"post",
endpoint="/api/v1/comment/report",
payload=payload,
response_type=CommentReportResponse,
base_url=base_url
)
base_url=base_url,
)
class ResolveCommentReportRequest(T.NamedTuple):
report: UUID
auth: str
class ResolveCommentReportResponse(T.NamedTuple):
report: UUID
resolved: bool
async def resolve_comment_report(payload: ResolveCommentReportRequest, base_url: str = lib.BASE_URL):
async def resolve_comment_report(
payload: ResolveCommentReportRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'post',
endpoint='/api/v1/comment/resolve_report',
"post",
endpoint="/api/v1/comment/resolve_report",
payload=payload,
response_type=ResolveCommentReportResponse,
base_url=base_url
)
base_url=base_url,
)
class ListCommentReportsRequest(T.NamedTuple):
page: T.Optional[int]
@ -43,17 +54,22 @@ class ListCommentReportsRequest(T.NamedTuple):
community: int
auth: str
class ListCommentReportResponse(T.NamedTuple):
reports: list[CommentReportView]
async def list_comment_reports(payload: ListCommentReportsRequest, base_url: str = lib.BASE_URL):
async def list_comment_reports(
payload: ListCommentReportsRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'get',
endpoint='/api/v1/community/comment_reports',
"get",
endpoint="/api/v1/community/comment_reports",
payload=payload,
response_type=ListCommentReportResponse,
base_url=base_url
)
base_url=base_url,
)
class ListPostReportsRequest(T.NamedTuple):
page: T.Optional[int]
@ -61,66 +77,85 @@ class ListPostReportsRequest(T.NamedTuple):
community: int
auth: str
class ListPostReportResponse(T.NamedTuple):
reports: list[PostReportView]
async def list_post_reports(payload: ListPostReportsRequest, base_url: str = lib.BASE_URL):
async def list_post_reports(
payload: ListPostReportsRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'get',
endpoint='/api/v1/community/post_reports',
"get",
endpoint="/api/v1/community/post_reports",
payload=payload,
response_type=ListPostReportResponse,
base_url=base_url
)
base_url=base_url,
)
class GetReportCountRequest(T.NamedTuple):
community: int
auth: str
class GetReportCountResponse(T.NamedTuple):
community: int
comment_reports: int
post_reports: int
async def get_report_count(payload: GetReportCountRequest, base_url: str = lib.BASE_URL):
async def get_report_count(
payload: GetReportCountRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'get',
endpoint='/api/v1/community/reports',
"get",
endpoint="/api/v1/community/reports",
payload=payload,
response_type=GetReportCountResponse,
base_url=base_url
)
base_url=base_url,
)
class CreatePostReportRequest(T.NamedTuple):
post: int
reason: T.Optional[str]
auth: str
class PostReportResponse(T.NamedTuple):
success: bool
async def create_post_report(payload: CreatePostReportRequest, base_url: str = lib.BASE_URL):
async def create_post_report(
payload: CreatePostReportRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'post',
endpoint='/api/v1/post/report',
"post",
endpoint="/api/v1/post/report",
payload=payload,
response_type=PostReportResponse,
base_url=base_url
)
base_url=base_url,
)
class ResolvePostReportRequest(T.NamedTuple):
report: UUID
auth: str
class ResolvePostReportResponse(T.NamedTuple):
report: UUID
resolved: bool
async def resolve_post_report(payload: ResolvePostReportRequest, base_url: str = lib.BASE_URL):
async def resolve_post_report(
payload: ResolvePostReportRequest, base_url: str = lib.BASE_URL
):
return await lib.request(
'post',
endpoint='/api/v1/post/resolve_report',
"post",
endpoint="/api/v1/post/resolve_report",
payload=payload,
response_type=ResolvePostReportResponse,
base_url=base_url
)
base_url=base_url,
)

11
hexbear/v1/search.py

@ -2,6 +2,7 @@ import typing as T
from . import lib
from .views import CommentView, PostView, CommunityView, UserView
class SearchRequest(T.NamedTuple):
q: str
type_: lib.SearchType
@ -11,6 +12,7 @@ class SearchRequest(T.NamedTuple):
limit: T.Optional[int] = None
auth: T.Optional[str] = None
class SearchResponse(T.NamedTuple):
type_: str
comments: list[CommentView]
@ -18,11 +20,12 @@ class SearchResponse(T.NamedTuple):
communities: list[CommunityView]
users: list[UserView]
async def search(payload: SearchRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/search',
"get",
endpoint="/api/v1/search",
payload=payload,
response_type=SearchResponse,
base_url=base_url
)
base_url=base_url,
)

77
hexbear/v1/site.py

@ -2,9 +2,11 @@ import typing as T
from . import lib
from .views import SiteView, UserView, CommunityView, MyUserView
class GetSiteRequest(T.NamedTuple):
auth: T.Optional[str] = None
class GetSiteResponse(T.NamedTuple):
site: T.Optional[SiteView]
admins: list[UserView]
@ -13,20 +15,23 @@ class GetSiteResponse(T.NamedTuple):
version: str
my_user: T.Optional[MyUserView]
federated_instances: list[str]
sitemods: list[UserView] # hexbear
sitemods: list[UserView] # hexbear
async def get(payload: GetSiteRequest = GetSiteRequest(), base_url: str = lib.BASE_URL):
return await lib.request(
'get',
endpoint='/api/v1/site',
"get",
endpoint="/api/v1/site",
payload=payload,
response_type=GetSiteResponse,
base_url=base_url
)
base_url=base_url,
)
class SiteResponse(T.NamedTuple):
site: SiteView
class CreateSiteRequest(T.NamedTuple):
name: str
description: T.Optional[str]
@ -37,14 +42,16 @@ class CreateSiteRequest(T.NamedTuple):
enable_nsfw: bool
auth: str
async def create(payload: CreateSiteRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/site',
"post",
endpoint="/api/v1/site",
payload=payload,
response_type=SiteResponse,
base_url=base_url
)
base_url=base_url,
)
class EditSiteRequest(T.NamedTuple):
name: str
@ -57,72 +64,84 @@ class EditSiteRequest(T.NamedTuple):
auth: str
enable_create_communities: T.Optional[bool]
async def edit(payload: EditSiteRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'put',
endpoint='/api/v1/site',
"put",
endpoint="/api/v1/site",
payload=payload,
response_type=SiteResponse,
base_url=base_url
)
base_url=base_url,
)
class TransferSiteRequest(T.NamedTuple):
user_id: int
auth: str
async def transfer(payload: TransferSiteRequest, base_url: str = lib.BASE_URL):
return await lib.request(
'post',
endpoint='/api/v1/site/transfer',
"post",
endpoint="/api/v1/site/transfer",
payload=payload,
response_type=SiteResponse,
base_url=base_url
)
base_url=base_url,
)