Merge pull request #14 from boticord/v3

V3
This commit is contained in:
Виктор 2023-06-27 18:44:52 +03:00 committed by GitHub
commit 549b671d36
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
32 changed files with 1780 additions and 920 deletions

1
.gitignore vendored
View file

@ -10,3 +10,4 @@ boticordpy.egg-info
_testing.py _testing.py
/.pytest_cache /.pytest_cache
/docs/build/ /docs/build/
.vscode

8
ACKNOWLEDGEMENTS.txt Normal file
View file

@ -0,0 +1,8 @@
These are the open source libraries we use:
- aiohttp (https://github.com/aio-libs/aiohttp)
- typing_extensions (https://github.com/python/typing_extensions)
- sphinx (https://github.com/sphinx-doc/sphinx)
- furo (https://github.com/pradyunsg/furo)
It is also important to note that some developments of Melisa (https://github.com/MelisaDev/melisa) were used in the development of the project.
I would also like to express my gratitude to the former admin staff of the BotiCord service (until 06/02/2023) and the entire project community.

View file

@ -1,4 +1,4 @@
Copyright 2021 Victor Kotlin Copyright 2021 - 2023 Viktor K (Marakarka)
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

View file

@ -22,12 +22,12 @@
* Object-oriented * Object-oriented
* Full BotiCord API Coverage * Full BotiCord API Coverage
* Modern Pythonic API using `async`/`await` syntax * Modern Pythonic API using `async`/`await` syntax
* BotiCord Webhooks * BotiCord Websocket
* It is not necessary to use any particular library used to interact with the Discord API. * It is not necessary to use any particular library used to interact with the Discord API.
<h2>Installation</h2> <h2>Installation</h2>
<b>Python 3.6 or newer is required.</b> <b>Python 3.8 or newer is required.</b>
Enter one of these commands to install the library: Enter one of these commands to install the library:
@ -49,28 +49,33 @@ You can find other examples in an examples folder.
```py ```py
from discord.ext import commands from discord.ext import commands
from boticordpy import BoticordClient from boticordpy import BoticordClient
bot = commands.Bot(command_prefix="!") bot = commands.Bot(command_prefix="!")
# Function that will return the current bot's stats.
async def get_stats(): async def get_stats():
return {"servers": len(bot.guilds), "shards": 0, "users": len(bot.users)} return {"servers": len(bot.guilds), "shards": 0, "users": len(bot.users)}
# Function that will be called if stats are posted successfully.
async def on_success_posting(): async def on_success_posting():
print("stats posting successfully") print("wow stats posting works")
boticord_client = BoticordClient("your_api_token")
boticord_client = BoticordClient(
"your_boticord_api_token", version=3
) # <--- BotiCord API token
autopost = ( autopost = (
boticord_client.autopost() boticord_client.autopost()
.init_stats(get_stats) .init_stats(get_stats)
.on_success(on_success_posting) .on_success(on_success_posting)
.start() .start("id_of_your_bot") # <--- ID of your bot
) )
bot.run("bot token") bot.run("bot token") # <--- Discord bot's token
``` ```
<h2>Links</h2> <h2>Links</h2>

View file

@ -2,17 +2,16 @@
Boticord API Wrapper Boticord API Wrapper
~~~~~~~~~~~~~~~~~~~ ~~~~~~~~~~~~~~~~~~~
A basic wrapper for the BotiCord API. A basic wrapper for the BotiCord API.
:copyright: (c) 2022 Marakarka :copyright: (c) 2023 Marakarka
:license: MIT, see LICENSE for more details. :license: MIT, see LICENSE for more details.
""" """
__title__ = "boticordpy" __title__ = "boticordpy"
__author__ = "Marakarka" __author__ = "Marakarka"
__license__ = "MIT" __license__ = "MIT"
__copyright__ = "Copyright 2022 Marakarka" __copyright__ = "Copyright 2021 - 2023 Marakarka"
__version__ = "2.2.2" __version__ = "3.0.0"
from .client import BoticordClient from .client import BoticordClient
from .webhook import Webhook
from .types import * from .types import *
from .websocket import BotiCordWebsocket

View file

@ -1,8 +1,11 @@
import asyncio import asyncio
import typing import typing
import logging
from . import exceptions as bexc from . import exceptions as bexc
_logger = logging.getLogger("boticord.autopost")
class AutoPost: class AutoPost:
""" """
@ -28,12 +31,16 @@ class AutoPost:
_stats: typing.Any _stats: typing.Any
_task: typing.Optional["asyncio.Task[None]"] _task: typing.Optional["asyncio.Task[None]"]
bot_id: str
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
self._stopped: bool = False self._stopped: bool = False
self._interval: int = 900 self._interval: int = 900
self._task: typing.Optional["asyncio.Task[None]"] = None self._task: typing.Optional["asyncio.Task[None]"] = None
self.bot_id = None
@property @property
def is_running(self) -> bool: def is_running(self) -> bool:
""" """
@ -65,6 +72,8 @@ class AutoPost:
self._success = callback self._success = callback
return func return func
_logger.info("Registering success callback")
return inner return inner
def on_error(self, callback: typing.Any = None): def on_error(self, callback: typing.Any = None):
@ -91,6 +100,8 @@ class AutoPost:
self._error = callback self._error = callback
return func return func
_logger.info("Registering error callback")
return inner return inner
def init_stats(self, callback: typing.Any = None): def init_stats(self, callback: typing.Any = None):
@ -116,6 +127,8 @@ class AutoPost:
self._stats = callback self._stats = callback
return func return func
_logger.info("Registered stats initialization function")
return inner return inner
@property @property
@ -145,7 +158,8 @@ class AutoPost:
while True: while True:
stats = await self._stats() stats = await self._stats()
try: try:
await self.client.http.post_bot_stats(stats) await self.client.http.post_bot_stats(self.bot_id, stats)
_logger.info("Tried to post bot stats")
except Exception as err: except Exception as err:
on_error = getattr(self, "_error", None) on_error = getattr(self, "_error", None)
if on_error: if on_error:
@ -160,14 +174,21 @@ class AutoPost:
await asyncio.sleep(self._interval) await asyncio.sleep(self._interval)
def start(self): def start(self, bot_id: typing.Union[str, int]):
""" """
Starts the loop. Starts the loop.
Args:
bot_id ( Union[:obj:`int`, :obj:`str`] )
Id of the bot to send stats of.
Raises: Raises:
:obj:`~.exceptions.InternalException` :obj:`~.exceptions.InternalException`
If there's no callback (for getting stats) provided or the autopost is already running. If there's no callback (for getting stats) provided or the autopost is already running.
""" """
self.bot_id = bot_id
if not hasattr(self, "_stats"): if not hasattr(self, "_stats"):
raise bexc.InternalException("You must provide stats") raise bexc.InternalException("You must provide stats")
@ -178,6 +199,9 @@ class AutoPost:
task = asyncio.ensure_future(self._internal_loop()) task = asyncio.ensure_future(self._internal_loop())
self._task = task self._task = task
_logger.info("Started autoposting")
return task return task
def stop(self) -> None: def stop(self) -> None:
@ -188,3 +212,5 @@ class AutoPost:
return None return None
self._stopped = True self._stopped = True
_logger.info("Stopped autoposting")

View file

@ -1,18 +1,17 @@
import typing import typing
import logging
from . import types as boticord_types from . import types as boticord_types
from .http import HttpClient from .http import HttpClient
from .autopost import AutoPost from .autopost import AutoPost
from .exceptions import MeilisearchException
_logger = logging.getLogger("boticord")
class BoticordClient: class BoticordClient:
"""Represents a client that can be used to interact with the BotiCord API. """Represents a client that can be used to interact with the BotiCord API.
.. warning::
In BotiCord API v2 there are some changes with token.
`Read more here <https://docs.boticord.top/topics/v1vsv2/>`_
Note: Note:
Remember that every http method can return an http exception. Remember that every http method can return an http exception.
@ -20,210 +19,179 @@ class BoticordClient:
token (:obj:`str`) token (:obj:`str`)
Your bot's Boticord API Token. Your bot's Boticord API Token.
version (:obj:`int`) version (:obj:`int`)
BotiCord API version (Default: 2) BotiCord API version (Default: 3)
""" """
__slots__ = ("http", "_autopost", "_token") __slots__ = ("http", "_autopost", "_token", "_meilisearch_api_key")
http: HttpClient http: HttpClient
def __init__(self, token: str = None, version: int = 2): def __init__(self, token: str = None, version: int = 3):
self._token = token self._token = token
self._meilisearch_api_key = None
self._autopost: typing.Optional[AutoPost] = None self._autopost: typing.Optional[AutoPost] = None
self.http = HttpClient(token, version) self.http = HttpClient(token, version)
async def get_bot_info(self, bot_id: int) -> boticord_types.Bot: async def get_bot_info(
self, bot_id: typing.Union[str, int]
) -> boticord_types.ResourceBot:
"""Gets information about specified bot. """Gets information about specified bot.
Args: Args:
bot_id (:obj:`int`) bot_id (Union[:obj:`str`, :obj:`int`])
Id of the bot Id of the bot
Returns: Returns:
:obj:`~.types.Bot`: :obj:`~.types.ResourceBot`:
Bot object. ResourceBot object.
""" """
_logger.info("Requesting information about bot")
response = await self.http.get_bot_info(bot_id) response = await self.http.get_bot_info(bot_id)
return boticord_types.Bot(**response) return boticord_types.ResourceBot.from_dict(response)
async def get_bot_comments(self, bot_id: int) -> list:
"""Gets list of comments of specified bot.
Args:
bot_id (:obj:`int`)
Id of the bot
Returns:
:obj:`list` [ :obj:`~.types.SingleComment` ]:
List of comments.
"""
response = await self.http.get_bot_comments(bot_id)
return [boticord_types.SingleComment(**comment) for comment in response]
async def post_bot_stats( async def post_bot_stats(
self, servers: int = 0, shards: int = 0, users: int = 0 self,
) -> dict: bot_id: typing.Union[str, int],
*,
servers: int = 0,
shards: int = 0,
users: int = 0,
) -> boticord_types.ResourceBot:
"""Post Bot's stats. """Post Bot's stats.
Args: Args:
bot_id (Union[:obj:`str`, :obj:`int`])
Id of the bot to post stats of.
servers ( :obj:`int` ) servers ( :obj:`int` )
Bot's servers count Bot's servers count
shards ( :obj:`int` ) shards ( :obj:`int` )
Bot's shards count Bot's shards count
users ( :obj:`int` ) users ( :obj:`int` )
Bot's users count Bot's users count
Returns:
:obj:`dict`:
Boticord API Response status
"""
response = await self.http.post_bot_stats(
{"servers": servers, "shards": shards, "users": users}
)
return response
async def get_server_info(self, server_id: int) -> boticord_types.Server: Returns:
:obj:`~.types.ResourceBot`:
ResourceBot object.
"""
_logger.info("Posting bot stats")
response = await self.http.post_bot_stats(
bot_id, {"servers": servers, "shards": shards, "users": users}
)
return boticord_types.ResourceBot.from_dict(response)
async def get_server_info(
self, server_id: typing.Union[str, int]
) -> boticord_types.ResourceServer:
"""Gets information about specified server. """Gets information about specified server.
Args: Args:
server_id (:obj:`int`) server_id (Union[:obj:`str`, :obj:`int`])
Id of the server Id of the server
Returns: Returns:
:obj:`~.types.Server`: :obj:`~.types.ResourceServer`:
Server object. ResourceServer object.
""" """
_logger.info("Requesting information about server")
response = await self.http.get_server_info(server_id) response = await self.http.get_server_info(server_id)
return boticord_types.Server(**response) return boticord_types.ResourceServer.from_dict(response)
async def get_server_comments(self, server_id: int) -> list: async def get_user_info(
"""Gets list of comments of specified server. self, user_id: typing.Union[str, int]
) -> boticord_types.UserProfile:
Args:
server_id (:obj:`int`)
Id of the server
Returns:
:obj:`list` [ :obj:`~.types.SingleComment` ]:
List of comments.
"""
response = await self.http.get_server_comments(server_id)
return [boticord_types.SingleComment(**comment) for comment in response]
async def post_server_stats(self, payload: dict) -> dict:
"""Post Server's stats. You must be Boticord-Service bot.
Payload is raw, because if you use it - you know what you are doing.
You can find more information about payload `in BotiCord API Docs <https://docs.boticord.top/methods/servers/>`_
Args:
payload (:obj:`dict`)
Custom data (Use Boticord API docs.)
Returns:
:obj:`dict`:
Boticord API Response.
"""
response = await self.http.post_server_stats(payload)
return response
async def get_user_info(self, user_id: int) -> boticord_types.UserProfile:
"""Gets information about specified user. """Gets information about specified user.
Args: Args:
user_id (:obj:`int`) user_id (Union[:obj:`str`, :obj:`int`])
Id of the user Id of the user
Returns: Returns:
:obj:`~.types.UserProfile`: :obj:`~.types.UserProfile`:
UserProfile object. UserProfile object.
""" """
_logger.info("Requesting information about user")
response = await self.http.get_user_info(user_id) response = await self.http.get_user_info(user_id)
return boticord_types.UserProfile(**response) return boticord_types.UserProfile.from_dict(response)
async def get_user_comments(self, user_id: int) -> boticord_types.UserComments: async def __search_for(self, index, data):
"""Gets comments of specified user. """Search for something on BotiCord"""
if self._meilisearch_api_key is None:
token_response = await self.http.get_search_key()
self._meilisearch_api_key = token_response["key"]
Args: try:
user_id (:obj:`int`) response = await self.http.search_for(
Id of the user index, self._meilisearch_api_key, data
)
except MeilisearchException:
token_response = await self.http.get_search_key()
self._meilisearch_api_key = token_response["key"]
Returns: response = await self.http.search_for(
:obj:`~.types.UserComments`: index, self._meilisearch_api_key, data
User comments on Bots and Servers pages.
"""
response = await self.http.get_user_comments(user_id)
return boticord_types.UserComments(**response)
async def get_user_bots(self, user_id: int) -> list:
"""Gets list of bots of specified user.
Args:
user_id (:obj:`int`)
Id of the user
Returns:
:obj:`list` [ :obj:`~.types.SimpleBot` ]:
List of simple information about users bots.
"""
response = await self.http.get_user_bots(user_id)
return [boticord_types.SimpleBot(**bot) for bot in response]
async def get_my_shorted_links(self, *, code: str = None):
"""Gets shorted links of an authorized user
Args:
code (:obj:`str`)
Code of shorted link. Could be None.
Returns:
Union[:obj:`list` [ :obj:`~.types.ShortedLink` ], :obj:`~types.ShortedLink`]:
List of shorted links if none else shorted link
"""
response = await self.http.get_my_shorted_links(code)
return (
[boticord_types.ShortedLink(**link) for link in response]
if code is None
else boticord_types.ShortedLink(**response[0])
) )
async def create_shorted_link( return response["hits"]
self, *, code: str, link: str, domain: boticord_types.LinkDomain = 1
):
"""Creates new shorted link
Args: async def search_for_bots(
code (:obj:`str`) self, **kwargs
Code of link to short. ) -> typing.List[boticord_types.MeiliIndexedBot]:
link (:obj:`str`) """Search for bots on BotiCord.
Link to short.
domain (:obj:`~.types.LinkDomain`) Note:
Domain to use in shorted link You can find every keyword argument `here <https://www.meilisearch.com/docs/reference/api/search#search-parameters>`_.
Returns: Returns:
:obj:`~types.ShortedLink`: List[:obj:`~.types.MeiliIndexedBot`]:
Shorted Link List of found bots
""" """
response = await self.http.create_shorted_link(code, link, domain=domain) _logger.info("Searching for bots on BotiCord")
return boticord_types.ShortedLink(**response) response = await self.__search_for("bots", kwargs)
return [boticord_types.MeiliIndexedBot.from_dict(bot) for bot in response]
async def delete_shorted_link( async def search_for_servers(
self, code: str, domain: boticord_types.LinkDomain = 1 self, **kwargs
): ) -> typing.List[boticord_types.MeiliIndexedServer]:
"""Deletes shorted link """Search for servers on BotiCord.
Args: Note:
code (:obj:`str`) You can find every keyword argument `here <https://www.meilisearch.com/docs/reference/api/search#search-parameters>`_.
Code of link to delete.
domain (:obj:`~.types.LinkDomain`)
Domain that is used in shorted link
Returns: Returns:
:obj:`bool`: List[:obj:`~.types.MeiliIndexedServer`]:
Is link deleted successfully? List of found servers
""" """
response = await self.http.delete_shorted_link(code, domain) _logger.info("Searching for servers on BotiCord")
return response.get("ok", False) response = await self.__search_for("servers", kwargs)
return [
boticord_types.MeiliIndexedServer.from_dict(server) for server in response
]
async def search_for_comments(
self, **kwargs
) -> typing.List[boticord_types.MeiliIndexedComment]:
"""Search for comments on BotiCord.
Note:
You can find every keyword argument `here <https://www.meilisearch.com/docs/reference/api/search#search-parameters>`_.
Returns:
List[:obj:`~.types.MeiliIndexedComment`]:
List of found comments
"""
_logger.info("Searching for comments on BotiCord")
response = await self.__search_for("comments", kwargs)
return [
boticord_types.MeiliIndexedComment.from_dict(comment)
for comment in response
]
def autopost(self) -> AutoPost: def autopost(self) -> AutoPost:
"""Returns a helper instance for auto-posting. """Returns a helper instance for auto-posting.

View file

@ -1,3 +1,6 @@
from enum import IntEnum
class BoticordException(Exception): class BoticordException(Exception):
"""Base exception class for boticordpy. """Base exception class for boticordpy.
This could be caught to handle any exceptions thrown from this library. This could be caught to handle any exceptions thrown from this library.
@ -18,7 +21,7 @@ class InternalException(BoticordException):
class HTTPException(BoticordException): class HTTPException(BoticordException):
"""Exception that's thrown when an HTTP request operation fails. """Exception that's thrown when request to BotiCord API operation fails.
Attributes Attributes
---------- ----------
@ -29,26 +32,148 @@ class HTTPException(BoticordException):
def __init__(self, response): def __init__(self, response):
self.response = response self.response = response
fmt = f"{self.response.reason} (Status code: {self.response.status})" fmt = f"{HTTPErrors(self.response['error']).name} (Status code: {StatusCodes(self.response['status']).name})"
super().__init__(fmt) super().__init__(fmt)
class Unauthorized(HTTPException): class MeilisearchException(BoticordException):
"""Exception that's thrown when status code 401 occurs.""" """Exception that's thrown when request to Meilisearch API operation fails.
Attributes
----------
response:
The response of the failed HTTP request.
"""
def __init__(self, response):
self.response = response
fmt = f"{self.response['code']} ({self.response['message']})"
super().__init__(fmt)
class Forbidden(HTTPException): class StatusCodes(IntEnum):
"""Exception that's thrown when status code 403 occurs.""" """Status codes of response"""
SERVER_ERROR = 500
"""Server Error (>500)"""
TOO_MANY_REQUESTS = 429
"""Too Many Requests"""
NOT_FOUND = 404
"""Requested resource was not found"""
FORBIDDEN = 403
"""You don't have access to this resource"""
UNAUTHORIZED = 401
"""Authorization is required to access this resource"""
BAD_REQUEST = 400
"""Bad Request"""
class NotFound(HTTPException): class HTTPErrors(IntEnum):
"""Exception that's thrown when status code 404 occurs.""" """Errors which BotiCord may return"""
UNKNOWN_ERROR = 0
"""Unknown error"""
class ToManyRequests(HTTPException): INTERNAL_SERVER_ERROR = 1
"""Exception that's thrown when status code 429 occurs.""" """Server error (>500)"""
RATE_LIMITED = 2
"""Too many requests"""
class ServerError(HTTPException): NOT_FOUND = 3
"""Exception that's thrown when status code 500 or 503 occurs.""" """Not found"""
FORBIDDEN = 4
"""Access denied"""
BAD_REQUEST = 5
"""Bad request"""
UNAUTHORIZED = 6
"""Unauthorized. Authorization required"""
RPC_ERROR = 7
"""Server error (RPC)"""
WS_ERROR = 8
"""Server error (WS)"""
THIRD_PARTY_FAIL = 9
"""Third-party service error"""
UNKNOWN_USER = 10
"""Unknown user"""
SHORT_DOMAIN_TAKEN = 11
"""Short link already taken"""
UNKNOWN_SHORT_DOMAIN = 12
"""Unknown short link"""
UNKNOWN_LIBRARY = 13
"""Unknown library"""
TOKEN_INVALID = 14
"""Invalid token"""
UNKNOWN_RESOURCE = 15
"""Unknown resource"""
UNKNOWN_TAG = 16
"""Unknown tag"""
PERMISSION_DENIED = 17
"""Insufficient permissions"""
UNKNOWN_COMMENT = 18
"""Unknown comment"""
UNKNOWN_BOT = 19
"""Unknown bot"""
UNKNOWN_SERVER = 20
"""Unknown server"""
UNKNOWN_BADGE = 21
"""Unknown badge"""
USER_ALREADY_HAS_A_BADGE = 22
"""User already has a badge"""
INVALID_INVITE_CODE = 23
"""Invalid invite code"""
SERVER_ALREADY_EXISTS = 24
"""Server already exists"""
BOT_NOT_PRESENT_ON_QUEUE_SERVER = 25
"""Bot not present on queue server"""
UNKNOWN_UP = 26
"""Unknown up"""
TOO_MANY_UPS = 27
"""Too many ups"""
INVALID_STATUS = 28
"""Invalid resource status"""
UNKNOWN_REPORT = 29
"""Unknown report"""
UNSUPPORTED_MEDIA_TYPE = 30
"""Unsupported media type. Should be one of"""
UNKNOWN_APPLICATION = 31
"""Unknown application"""
AUTOMATED_REQUESTS_NOT_ALLOWED = 32
"""Please confirm that you are not a robot by refreshing the page"""

View file

@ -1,9 +1,9 @@
import asyncio import asyncio
import typing
import aiohttp import aiohttp
from . import exceptions from . import exceptions
from .types import LinkDomain
class HttpClient: class HttpClient:
@ -20,96 +20,66 @@ class HttpClient:
loop: `asyncio loop` loop: `asyncio loop`
""" """
def __init__(self, auth_token: str, version: int = 1, **kwargs): def __init__(self, auth_token: str = None, version: int = 3, **kwargs):
self.token = auth_token self.token = auth_token
self.API_URL = f"https://api.boticord.top/v{version}/" self.API_URL = f"https://api.boticord.top/v{version}"
loop = kwargs.get("loop") or asyncio.get_event_loop() loop = kwargs.get("loop") or asyncio.get_event_loop()
self.session = kwargs.get("session") or aiohttp.ClientSession(loop=loop) self.session = kwargs.get("session") or aiohttp.ClientSession(loop=loop)
async def make_request(self, method: str, endpoint: str, **kwargs): async def make_request(
self, method: str, endpoint: str, *, meilisearch_token: str = None, **kwargs
) -> dict:
"""Send requests to the API""" """Send requests to the API"""
kwargs["headers"] = { kwargs["headers"] = {"Content-Type": "application/json"}
"Content-Type": "application/json",
"Authorization": self.token, if self.token is not None:
} kwargs["headers"]["Authorization"] = self.token
if meilisearch_token is not None:
kwargs["headers"]["Authorization"] = f"Bearer {meilisearch_token}"
url = f"{self.API_URL}{endpoint}" url = f"{self.API_URL}{endpoint}"
async with self.session.request(method, url, **kwargs) as response: async with self.session.request(method, url, **kwargs) as response:
data = await response.json() data = await response.json()
if response.status == 200: if (200, 201).__contains__(response.status):
return data return data["result"] if not meilisearch_token else data
elif response.status == 401: else:
raise exceptions.Unauthorized(response) if not meilisearch_token:
elif response.status == 403: raise exceptions.HTTPException(
raise exceptions.Forbidden(response) {"status": response.status, "error": data["errors"][0]["code"]}
elif response.status == 404: )
raise exceptions.NotFound(response) else:
elif response.status == 429: raise exceptions.MeilisearchException(data)
raise exceptions.ToManyRequests(response)
elif response.status == 500:
raise exceptions.ServerError(response)
elif response.status == 503:
raise exceptions.ServerError(response)
raise exceptions.HTTPException(response) def get_bot_info(self, bot_id: typing.Union[str, int]):
def get_bot_info(self, bot_id: int):
"""Get information about the specified bot""" """Get information about the specified bot"""
return self.make_request("GET", f"bot/{bot_id}") return self.make_request("GET", f"bots/{bot_id}")
def get_bot_comments(self, bot_id: int): def post_bot_stats(self, bot_id: typing.Union[str, int], stats: dict):
"""Get list of specified bot comments"""
return self.make_request("GET", f"bot/{bot_id}/comments")
def post_bot_stats(self, stats: dict):
"""Post bot's stats""" """Post bot's stats"""
return self.make_request("POST", "stats", json=stats) return self.make_request("POST", f"bots/{bot_id}/stats", json=stats)
def get_server_info(self, server_id: int): def get_server_info(self, server_id: typing.Union[str, int]):
"""Get information about specified server""" """Get information about specified server"""
return self.make_request("GET", f"server/{server_id}") return self.make_request("GET", f"servers/{server_id}")
def get_server_comments(self, server_id: int): def get_user_info(self, user_id: typing.Union[str, int]):
"""Get list of specified server comments""" """Get information about specified user"""
return self.make_request("GET", f"server/{server_id}/comments") return self.make_request("GET", f"users/{user_id}")
def post_server_stats(self, payload: dict): def get_search_key(self):
"""Post server's stats""" """Get API key for Meilisearch"""
return self.make_request("POST", "server", json=payload) return self.make_request("GET", f"search-key")
def get_user_info(self, user_id: int): def search_for(self, index: str, api_key: str, data: dict):
"""Get information about the user""" """Search for something on BotiCord."""
return self.make_request("GET", f"profile/{user_id}")
def get_user_comments(self, user_id: int):
"""Get specified user's comments"""
return self.make_request("GET", f"user/{user_id}/comments")
def get_user_bots(self, user_id: int):
"""Get bots of specified user"""
return self.make_request("GET", f"bots/{user_id}")
def get_my_shorted_links(self, code: str = None):
"""Get shorted links of an authorized user"""
body = {"code": code} if code is not None else {}
return self.make_request("POST", "links/get", json=body)
def create_shorted_link(self, code: str, link: str, *, domain: LinkDomain = 1):
"""Create new shorted link"""
return self.make_request( return self.make_request(
"POST", "POST",
"links/create", f"search/indexes/{index}/search",
json={"code": code, "link": link, "domain": int(domain)}, meilisearch_token=api_key,
) json=data,
def delete_shorted_link(self, code: str, domain: LinkDomain = 1):
"""Delete shorted link"""
return self.make_request(
"POST", "links/delete", json={"code": code, "domain": int(domain)}
) )

File diff suppressed because it is too large Load diff

View file

@ -1,135 +0,0 @@
import asyncio
import typing
from .types import BumpResponse, CommentResponse
from aiohttp import web
import aiohttp
class Webhook:
"""Represents a client that can be used to work with BotiCord Webhooks.
IP of the server - your machine IP. (`0.0.0.0`)
Args:
x_hook_key (:obj:`str`)
X-hook-key to check the auth of incoming request.
endpoint_name (:obj:`str`)
Name of endpoint (for example: `/bot`)
Keyword Arguments:
loop: `asyncio loop`
"""
__slots__ = (
"_webserver",
"_listeners",
"_is_running",
"__app",
"_endpoint_name",
"_x_hook_key",
"_loop",
)
__app: web.Application
_webserver: web.TCPSite
def __init__(self, x_hook_key: str, endpoint_name: str, **kwargs) -> None:
self._x_hook_key = x_hook_key
self._endpoint_name = endpoint_name
self._listeners = {}
self.__app = web.Application()
self._is_running = False
self._loop = kwargs.get("loop") or asyncio.get_event_loop()
def listener(self, response_type: str):
"""Decorator to set the listener.
Args:
response_type (:obj:`str`)
Type of response (Check reference page)
"""
def inner(func):
if not asyncio.iscoroutinefunction(func):
raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
self._listeners[response_type] = func
return func
return inner
def register_listener(self, response_type: str, callback: typing.Any):
"""Method to set the listener.
Args:
response_type (:obj:`str`)
Type of response (Check reference page)
callback (:obj:`function`)
Coroutine Callback Function
"""
if not asyncio.iscoroutinefunction(callback):
raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
self._listeners[response_type] = callback
return self
async def _interaction_handler(self, request: aiohttp.web.Request) -> web.Response:
"""Interaction handler"""
auth = request.headers.get("X-Hook-Key")
if auth == self._x_hook_key:
data = await request.json()
responder = self._listeners.get(data["type"])
if responder is not None:
await responder(
(
BumpResponse
if data["type"].endswith("_bump")
else CommentResponse
)(**data)
)
return web.Response(status=200)
return web.Response(status=401)
async def _run(self, port):
self.__app.router.add_post("/" + self._endpoint_name, self._interaction_handler)
runner = web.AppRunner(self.__app)
await runner.setup()
self._webserver = web.TCPSite(runner, "0.0.0.0", port)
await self._webserver.start()
self._is_running = True
def start(self, port: int) -> None:
"""Method to start the webhook server
Args:
port (:obj:`int`)
Port to start the webserver
"""
self._loop.create_task(self._run(port))
@property
def is_running(self) -> bool:
"""If the server running?"""
return self._is_running
@property
def listeners(self) -> dict:
"""Dictionary of listeners (`type`: `callback function`)"""
return self._listeners
@property
def app(self) -> web.Application:
"""Web application that handles incoming requests"""
return self.__app
async def close(self) -> None:
"""Stop the webhooks server"""
await self._webserver.stop()
self._is_running = False

151
boticordpy/websocket.py Normal file
View file

@ -0,0 +1,151 @@
# Copyright Marakarka (Viktor K) 2021 - Present
# Full MIT License can be found in `LICENSE.txt` at the project root.
import logging
import json
import asyncio
import typing
import aiohttp
_logger = logging.getLogger("boticord.websocket")
class BotiCordWebsocket:
"""Represents a client that can be used to interact with the BotiCord by websocket connection."""
def __init__(self, token: str):
self.__session = None
self.loop = asyncio.get_event_loop()
self.ws = None
self._listeners = {}
self.not_closed = True
self._token = token
def listener(self):
"""Decorator to set the listener.
.. warning::
Callback functions must be a **coroutine**. If they aren't, then you might get unexpected
errors. In order to turn a function into a coroutine they must be ``async def``
functions.
For example:
.. code-block:: python
@websocket.listener()
async def comment_removed(data):
pass
"""
def inner(func):
if not asyncio.iscoroutinefunction(func):
raise TypeError(f"<{func.__qualname__}> must be a coroutine function")
self._listeners[func.__qualname__] = func
_logger.debug(f"Listener {func.__qualname__} added successfully!")
return func
return inner
def register_listener(self, notification_type: str, callback: typing.Any):
"""Method to set the listener.
Args:
notify_type (:obj:`str`)
Type of notification (Check reference page)
callback (:obj:`function`)
Coroutine Callback Function
.. warning::
Callback functions must be a **coroutine**. If they aren't, then you might get unexpected
errors. In order to turn a function into a coroutine they must be ``async def``
functions.
"""
if not asyncio.iscoroutinefunction(callback):
raise TypeError(f"<{callback.__qualname__}> must be a coroutine function")
self._listeners[notification_type] = callback
_logger.debug(f"Listener {callback.__qualname__} added successfully!")
return self
async def connect(self) -> None:
"""Connect to BotiCord."""
try:
self.__session = aiohttp.ClientSession()
self.ws = await self.__session.ws_connect(
"wss://gateway.boticord.top/websocket/",
timeout=30.0,
)
_logger.info("Connected to BotiCord.")
self.not_closed = True
self.loop.create_task(self._receive())
await self._send_identify()
except Exception as exc:
_logger.error("Connecting failed!")
raise exc
async def _send_identify(self) -> None:
await self.ws.send_json({"event": "auth", "data": {"token": self._token}})
async def _receive(self) -> None:
while self.not_closed:
async for msg in self.ws:
if msg.type == aiohttp.WSMsgType.TEXT:
await self._handle_data(msg.data)
else:
raise RuntimeError
close_code = self.ws.close_code
if close_code is not None:
await self._handle_close(close_code)
async def _handle_data(self, data):
data = json.loads(data)
if data["event"] == "hello":
_logger.info("Authorized successfully.")
self.loop.create_task(self._send_ping())
elif data["event"] == "notify":
listener = self._listeners.get(data["data"]["type"])
if listener:
self.loop.create_task(listener(data["data"]))
elif data["event"] == "pong":
_logger.info("Received pong-response.")
self.loop.create_task(self._send_ping())
else:
_logger.error("An error has occurred.")
async def _handle_close(self, code: int) -> None:
self.not_closed = False
await self.__session.close()
if code == 4000:
_logger.info("Closed connection successfully.")
return
elif code == 1006:
_logger.error("Token is invalid.")
return
_logger.info("Disconnected from BotiCord. Reconnecting...")
await self.connect()
async def _send_ping(self) -> None:
if not self.ws.closed:
await asyncio.sleep(45)
await self.ws.send_json({"event": "ping"})
async def close(self) -> None:
"""Close websocket connection with BotiCord"""
if self.ws:
self.not_closed = False
await self.ws.close(code=4000)

View file

@ -1,2 +1,4 @@
sphinxawesome_theme furo
sphinxcontrib_trio
sphinx_design
sphinx sphinx

View file

@ -15,4 +15,3 @@ API Reference for the boticordpy Module
api/autopost api/autopost
api/exceptions api/exceptions
api/types api/types
api/webhook

View file

@ -1,6 +1,6 @@
#################### ###########################
AutoPost API Reference AutoPost API Reference
#################### ###########################
.. automodule:: boticordpy.autopost .. automodule:: boticordpy.autopost
:members: :members:

View file

@ -1,7 +1,11 @@
.. currentmodule:: boticordpy
#################### ####################
Client API Reference Client API Reference
#################### ####################
.. automodule:: boticordpy.client BoticordClient
:members: -----------------
.. autoclass:: BoticordClient
:inherited-members: :inherited-members:

View file

@ -1,7 +1,23 @@
#################### .. currentmodule:: boticordpy.exceptions
Exceptions API Reference
####################
.. automodule:: boticordpy.exceptions ##########################
Exceptions API Reference
##########################
.. autoclass:: BoticordException
:members:
.. autoclass:: InternalException
:members:
.. autoclass:: HTTPException
:members:
.. autoclass:: MeilisearchException
:members:
.. autoclass:: StatusCodes
:members:
.. autoclass:: HTTPErrors
:members: :members:
:inherited-members:

View file

@ -1,8 +1,71 @@
.. currentmodule:: boticordpy.types
#################### ####################
Models API Reference Models API Reference
#################### ####################
We recommend you to read the `boticordpy/types.py <https://github.com/boticord/boticordpy/blob/master/boticordpy/types.py>`_ file, because it is much easier to read than here. .. autoclass:: ResourceRating
:members:
.. automodule:: boticordpy.types
.. autoclass:: ResourceUp
:members:
Enums
-------
.. autoclass:: BotLibrary
:members:
.. autoclass:: BotTag
:members:
.. autoclass:: ServerTag
:members:
.. autoclass:: ResourceStatus
:members:
Bots
------
.. autoclass:: ResourceBot
:members:
Servers
---------
.. autoclass:: ResourceServer
:members:
Users
------
.. autoclass:: UserLinks
:members:
.. autoclass:: UserBadge
:members:
.. autoclass:: PartialUser
:members:
.. autoclass:: UserProfile
:members:
:exclude-members: to_dict
:inherited-members:
MeiliSearch
------------
.. autoclass:: MeiliIndexedBot
:members:
.. autoclass:: MeiliIndexedServer
:members:
.. autoclass:: MeiliIndexedComment
:members: :members:

View file

@ -1,7 +0,0 @@
####################
Webhook API Reference
####################
.. automodule:: boticordpy.webhook
:members:
:inherited-members:

View file

@ -23,11 +23,11 @@ import os
sys.path.insert(0, os.path.abspath("../..")) sys.path.insert(0, os.path.abspath("../.."))
project = "BoticordPY" project = "BoticordPY"
copyright = "2022, Victor Kotlin (Marakarka)" copyright = "2022 - 2023, Viktor K (Marakarka)"
author = "Victor Kotlin (Marakarka)" author = "Viktor K (Marakarka)"
# The full version, including alpha/beta/rc tags # The full version, including alpha/beta/rc tags
release = "2.2.2" release = "3.0.0a"
# -- General configuration --------------------------------------------------- # -- General configuration ---------------------------------------------------
@ -36,41 +36,47 @@ release = "2.2.2"
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom # extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones. # ones.
extensions = [ extensions = [
"sphinx_design",
"sphinx.ext.napoleon", "sphinx.ext.napoleon",
"sphinx.ext.autodoc", "sphinx.ext.autodoc",
"sphinx.ext.viewcode", "sphinx.ext.viewcode",
"sphinx.ext.autosectionlabel", "sphinx.ext.autosectionlabel",
"sphinx.ext.extlinks", "sphinx.ext.extlinks",
"sphinxcontrib_trio",
] ]
autodoc_default_options = {
"members": True,
"show-inheritance": True,
"member-order": "bysource",
}
# Add any paths that contain templates here, relative to this directory. # Add any paths that contain templates here, relative to this directory.
templates_path = ["_templates"] templates_path = ["_templates"]
# List of patterns, relative to source directory, that match files and add_module_names = False
# directories to ignore when looking for source files.
# This pattern also affects html_static_path and html_extra_path.
exclude_patterns = [] exclude_patterns = []
intersphinx_mapping = { intersphinx_mapping = {
"py": ("https://docs.python.org/3", None), "py": ("https://docs.python.org/3", None),
"discord": ("https://discordpy.readthedocs.io/en/latest/", None),
"aiohttp": ("https://docs.aiohttp.org/en/stable/", None), "aiohttp": ("https://docs.aiohttp.org/en/stable/", None),
} }
# -- Options for HTML output ------------------------------------------------- # -- Options for HTML output -------------------------------------------------
# The theme to use for HTML and HTML Help pages. See the documentation for html_theme = "furo"
# a list of builtin themes. html_theme_options = {
# "sidebar_hide_name": True,
html_theme = "sphinxawesome_theme" }
pygments_style = "monokai"
html_theme_options = {} default_dark_mode = True
# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "custom.css" will overwrite the builtin "custom.css".
html_static_path = ["_static"] html_static_path = ["_static"]
html_css_files = ["custom.css"]
rst_prolog = """
def setup(app): .. |coro| replace:: This function is a |coroutine_link|_.
app.add_css_file("custom.css") .. |maybecoro| replace:: This function *could be a* |coroutine_link|_.
.. |coroutine_link| replace:: *coroutine*
.. _coroutine_link: https://docs.python.org/3/library/asyncio-task.html#coroutine
"""

View file

@ -11,7 +11,7 @@ This is a documentation for wrapper for BotiCord API.
quickstart quickstart
api api
other websocket
Links Links
===== =====

View file

@ -1,38 +0,0 @@
.. currentmodule:: boticordpy
.. other:
Other Information
=================
##########
Listeners
##########
When you work with BotiCord Webhooks you may receive a lot of events.
To make it easier to handle them there is a list of the events you can receive:
.. csv-table::
:header: "BotiCord Events", "Meaning"
:widths: 20, 20
"test_webhook_message", "Test message."
"new_bot_comment", "On new bot comment"
"edit_bot_comment", "On bot comment edit"
"delete_bot_comment", "On bot comment delete"
"new_bot_bump", "On new bot bump"
"new_server_comment", "On new server comment"
"edit_server_comment", "On server comment edit"
"delete_server_comment", "On server comment delete"
"new_server_bump", "On new server bump"
##################
Callback functions
##################
.. warning::
Callback functions must be a **coroutine**. If they aren't, then you might get unexpected
errors. In order to turn a function into a coroutine they must be ``async def``
functions.

22
docs/source/websocket.rst Normal file
View file

@ -0,0 +1,22 @@
.. currentmodule:: boticordpy.websocket
###########
WebSocket
###########
BotiCord Websocket
-------------------
.. autoclass:: BotiCordWebsocket
:exclude-members: listener
:inherited-members:
.. automethod:: BotiCordWebsocket.listener()
:decorator:
Notification types
-------------------
.. function:: comment_removed(data)
Called when comment is deleted.

View file

@ -15,15 +15,17 @@ async def get_stats():
# Function that will be called if stats are posted successfully. # Function that will be called if stats are posted successfully.
async def on_success_posting(): async def on_success_posting():
print("stats posting successfully") print("wow stats posting works")
boticord_client = BoticordClient("Bot your_api_token", version=2) boticord_client = BoticordClient(
"your_boticord_api_token", version=3
) # <--- BotiCord API token
autopost = ( autopost = (
boticord_client.autopost() boticord_client.autopost()
.init_stats(get_stats) .init_stats(get_stats)
.on_success(on_success_posting) .on_success(on_success_posting)
.start() .start("id_of_your_bot") # <--- ID of your bot
) )
bot.run("bot token") bot.run("bot token") # <--- Discord bot's token

16
examples/stats_melisa.py Normal file
View file

@ -0,0 +1,16 @@
import melisa
from boticordpy import BoticordClient
bot = melisa.Bot("your_discord_bot_token")
boticord = BoticordClient("your_boticord_api_token")
@bot.listen
async def on_message_create(message):
if message.content.startswith("!guilds"):
data = await boticord.get_bot_info(bot.user.id)
await bot.rest.create_message(message.channel.id, data.guilds)
bot.run_autosharded()

View file

@ -1,20 +0,0 @@
# You can use any library to interact with the Discord API.
# This example uses discord.py.
# You can install it with `pip install discord.py`.
from discord.ext import commands
from boticordpy import webhook
bot = commands.Bot(command_prefix="!")
async def edit_bot_comment(data):
print(data.comment.new)
boticord_webhook = webhook.Webhook("x-hook-key", "bot").register_listener(
"edit_bot_comment", edit_bot_comment
)
boticord_webhook.start(5000)
bot.run("bot_token")

23
examples/websocket.py Normal file
View file

@ -0,0 +1,23 @@
# You can use any library to interact with the Discord API.
# This example uses discord.py.
# You can install it with `pip install discord.py`.
from discord.ext import commands
from boticordpy import BotiCordWebsocket
bot = commands.Bot(command_prefix="!")
websocket = BotiCordWebsocket("your_boticord_api_token") # <--- BotiCord API token
@websocket.listener()
async def comment_removed(data):
print(data["payload"])
@bot.event
async def on_ready():
await websocket.connect()
bot.run("bot token") # <--- Discord bot's token

View file

@ -1 +1,2 @@
aiohttp aiohttp
typing_extensions

View file

@ -44,10 +44,11 @@ setup(
}, },
packages=find_packages(), packages=find_packages(),
version=version, version=version,
python_requires=">= 3.6", python_requires=">= 3.8",
description="A Python wrapper for BotiCord API", description="A Python wrapper for BotiCord API",
long_description=README, long_description=README,
long_description_content_type="text/markdown", long_description_content_type="text/markdown",
include_package_data=True,
url="https://github.com/boticord/boticordpy", url="https://github.com/boticord/boticordpy",
author="Marakarka", author="Marakarka",
author_email="support@kerdoku.top", author_email="support@kerdoku.top",
@ -55,7 +56,9 @@ setup(
classifiers=[ classifiers=[
"License :: OSI Approved :: MIT License", "License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3", "Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.7", "Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
], ],
install_requires=["aiohttp"], install_requires=["aiohttp", "typing_extensions"],
) )

View file

@ -0,0 +1,74 @@
from boticordpy import types
resource_up_dict = {"id": "arbuz123", "expires": "1685262170000"}
resource_rating_dict = {"count": 15, "rating": 5}
resource_bot_dict = {
"id": "947141336451153931",
"name": "BumpBot",
"status": 1,
"createdDate": "2023-05-22T22:29:23.264Z",
"premium": {},
}
resource_server_dict = {
"id": "722424773233213460",
"name": "BotiCord.top",
"tags": [134, 132],
"status": 1,
"createdDate": "2023-05-23T15:16:45.387Z",
"premium": {},
}
user_profile_dict = {
"id": "585766846268047370",
"username": "Marakarka",
"bots": [resource_bot_dict],
"shortDescription": None,
}
def test_resource_up_convertation():
model_from_dict = types.ResourceUp.from_dict(resource_up_dict)
assert model_from_dict.id == "arbuz123"
assert (
model_from_dict.expires.strftime("%Y.%m.%d %H:%M:%S") == "2023.05.28 08:22:50"
)
dict_from_model = model_from_dict.to_dict()
assert dict_from_model == resource_up_dict
def test_resource_rating_convertation():
model_from_dict = types.ResourceRating.from_dict(resource_rating_dict)
assert model_from_dict.count == 15
assert model_from_dict.rating == 5
dict_from_model = model_from_dict.to_dict()
assert dict_from_model == resource_rating_dict
def test_resource_bot_convertation():
model_from_dict = types.ResourceBot.from_dict(resource_bot_dict)
assert int(model_from_dict.created_date.timestamp()) == 1684794563
assert model_from_dict.status.name == "PUBLIC"
def test_resource_server_convertation():
model_from_dict = types.ResourceServer.from_dict(resource_server_dict)
assert int(model_from_dict.created_date.timestamp()) == 1684855005
assert model_from_dict.name == "BotiCord.top"
assert model_from_dict.tags[1].name == "GAMES"
def test_user_profile_convertation():
model_from_dict = types.UserProfile.from_dict(user_profile_dict)
assert model_from_dict.id == "585766846268047370"
assert model_from_dict.username == "Marakarka"
assert model_from_dict.short_description == None
assert model_from_dict.bots[0].id == "947141336451153931"

View file

@ -1,141 +0,0 @@
import pytest
from boticordpy import types
single_comment_dict = {
"userID": "525366699969478676",
"text": "aboba",
"vote": 1,
"isUpdated": False,
"createdAt": 1644388399,
}
bot_data_dict = {
"id": "724663360934772797",
"shortCode": "kerdoku",
"links": [
"https://boticord.top/bot/724663360934772797",
"https://bcord.cc/b/724663360934772797",
"https://myservers.me/b/724663360934772797",
"https://boticord.top/bot/kerdoku",
"https://bcord.cc/b/kerdoku",
"https://myservers.me/b/kerdoku",
],
"server": {"id": "724668798874943529", "approved": True},
"information": {
"bumps": 37,
"added": 1091,
"prefix": "?",
"permissions": 1544023111,
"tags": ["комбайн", "экономика", "модерация", "приветствия"],
"developers": ["585766846268047370"],
"links": {"discord": "5qXgJvr", "github": None, "site": "https://kerdoku.top"},
"library": "discordpy",
"shortDescription": "Удобный и дружелюбный бот, который имеет крутой функционал!",
"longDescription": "wow",
"badge": None,
"stats": {"servers": 2558, "shards": 3, "users": 348986},
"status": "APPROVED",
},
}
server_data_dict = {
"id": "722424773233213460",
"shortCode": "boticord",
"status": "ACCEPT_MEMBERS",
"links": [
"https://boticord.top/server/722424773233213460",
"https://bcord.cc/s/722424773233213460",
"https://myservers.me/s/722424773233213460",
"https://boticord.top/server/boticord",
"https://bcord.cc/s/boticord",
"https://myservers.me/s/boticord",
],
"bot": {"id": None, "approved": False},
"information": {
"name": "BotiCord Community",
"avatar": "https://cdn.discordapp.com/icons/722424773233213460/060188f770836697846710b109272e4c.webp",
"members": [438, 0],
"bumps": 62,
"tags": [
"аниме",
"игры",
"поддержка",
"комьюнити",
"сообщество",
"discord",
"дискорд сервера",
"дискорд боты",
],
"links": {
"invite": "hkHjW8a",
"site": "https://boticord.top/",
"youtube": None,
"twitch": None,
"steam": None,
"vk": None,
},
"shortDescription": "short text",
"longDescription": "long text",
"badge": "STAFF",
},
}
user_profile_dict = {
"id": "178404926869733376",
"status": '"Если вы не разделяете мою точку зрения, поздравляю — вам больше достанется." © Артемий Лебедев',
"badge": "STAFF",
"shortCode": "cipherka",
"site": "https://sqdsh.top/",
"vk": None,
"steam": "sadlycipherka",
"youtube": None,
"twitch": None,
"git": "https://git.sqdsh.top/me",
}
@pytest.fixture
def single_comment() -> types.SingleComment:
return types.SingleComment(**single_comment_dict)
@pytest.fixture
def bot_data() -> types.Bot:
return types.Bot(**bot_data_dict)
@pytest.fixture
def server_data() -> types.Server:
return types.Bot(**server_data_dict)
@pytest.fixture
def user_profile_data() -> types.UserProfile:
return types.UserProfile(**user_profile_dict)
def test_comment_dict_fields(single_comment: types.SingleComment) -> None:
for attr in single_comment:
assert single_comment.get(attr) == getattr(single_comment, attr)
def test_user_profile_dict_fields(user_profile_data: types.UserProfile) -> None:
for attr in user_profile_data:
assert user_profile_data.get(attr) == getattr(user_profile_data, attr)
def test_bot_dict_fields(bot_data: types.Bot) -> None:
for attr in bot_data:
if attr.lower() == "information":
assert bot_data["information"].get(attr) == getattr(bot_data, attr)
else:
assert bot_data[attr] == getattr(bot_data, attr)
def test_server_dict_fields(server_data: types.Server) -> None:
for attr in server_data:
if attr.lower() == "information":
assert server_data["information"].get(attr) == getattr(bot_data, attr)
else:
assert server_data[attr] == getattr(server_data, attr)