profitroll/LibBotUniversal
profitroll/LibBotUniversal
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Releases 32 Activity

WIP: Overhaul for 4.0.0

Browse Source
This commit is contained in:
kku 2024-12-18 14:16:37 +01:00
parent 5e479ddc79
commit 95d04308bd
32 changed files with 718 additions and 688 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
pyproject.toml
View File

@ -8,17 +8,16 @@ dynamic = ["version", "dependencies", "optional-dependencies"]
authors = [{ name = "Profitroll" }]
description = "Universal bot library with functions needed for basic Discord/Telegram bot development."
readme = "README.md"
requires-python = ">=3.9"
requires-python = ">=3.11"
license = { text = "GPLv3" }
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Topic :: Software Development :: Libraries :: Python Modules",
"Topic :: Utilities",
]
@ -42,7 +41,8 @@ Tracker = "https://git.end-play.xyz/profitroll/LibBotUniversal/issues"
where = ["src"]
[tool.black]
target-version = ['py39', 'py310', 'py311' ,'py312']
line-length = 108
target-version = ["py311", "py312", "py313"]
[tool.isort]
profile = "black"
@ -52,6 +52,8 @@ minversion = "6.0"
python_files = ["test_*.py"]
pythonpath = "."
testpaths = ["tests"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
[tool.mypy]
namespace_packages = true
@ -59,9 +61,12 @@ install_types = true
strict = true
show_error_codes = true
[tool.pylint]
disable = ["line-too-long"]
[tool.pylint.main]
extension-pkg-whitelist = ["ujson"]
py-version = 3.9
py-version = 3.11
[tool.coverage.run]
source = ["libbot"]

5
src/libbot/__init__.py
View File

@ -1,6 +1,5 @@
__version__ = "3.3.1"
__version__ = "4.0.0"
__license__ = "GPL3"
__author__ = "Profitroll"
from . import errors, i18n, pycord, pyrogram, sync
from .__main__ import config_delete, config_get, config_set, json_read, json_write
from . import utils, errors, i18n, pycord, pyrogram

131
src/libbot/__main__.py
View File

@ -1,131 +0,0 @@
from pathlib import Path
from typing import Any, Union
import aiofiles
try:
from ujson import dumps, loads
except ImportError:
from json import dumps, loads
from ._utils import supports_argument
from .sync._nested import nested_delete, nested_set
DEFAULT_CONFIG_LOCATION: str = "config.json"
async def json_read(path: Union[str, Path]) -> Any:
"""Read contents of a JSON file
### Args:
* path (`Union[str, Path]`): Path-like object or path as a string
### Returns:
* `Any`: File contents
"""
async with aiofiles.open(str(path), mode="r", encoding="utf-8") as f:
data = await f.read()
return loads(data)
async def json_write(data: Any, path: Union[str, Path]) -> None:
"""Write contents to a JSON file
### Args:
* data (`Any`): Contents to write. Must be a JSON serializable
* path (`Union[str, Path]`): Path-like object or path as a string of a destination
"""
async with aiofiles.open(str(path), mode="w", encoding="utf-8") as f:
await f.write(
dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4)
if supports_argument(dumps, "escape_forward_slashes")
else dumps(data, ensure_ascii=False, indent=4)
)
async def config_get(
key: str, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION
) -> Any:
"""Get a value of the config key by its path provided
For example, `foo.bar.key` has a path of `"foo", "bar"` and the key `"key"`
### Args:
* key (`str`): Key that contains the value
* *path (`str`): Path to the key that contains the value
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Returns:
* `Any`: Key's value
### Example:
Get the "salary" of "Pete" from this JSON structure:
```json
{
"users": {
"Pete": {
"salary": 10.0
}
}
}
```
This can be easily done with the following code:
```python
import libbot
salary = await libbot.config_get("salary", "users", "Pete")
```
"""
this_key = await json_read(config_file)
for dict_key in path:
this_key = this_key[dict_key]
return this_key[key]
async def config_set(
key: str, value: Any, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION
) -> None:
"""Set config's key by its path to the value
### Args:
* key (`str`): Key that leads to the value
* value (`Any`): Any JSON serializable data
* *path (`str`): Path to the key of the target
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided
"""
await json_write(
nested_set(await json_read(config_file), value, *(*path, key)), config_file
)
async def config_delete(
key: str,
*path: str,
missing_ok: bool = False,
config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION,
) -> None:
"""Set config's key by its path
### Args:
* key (`str`): Key to delete
* *path (`str`): Path to the key of the target
* missing_ok (`bool`): Do not raise an exception if the key is missing. Defaults to `False`
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided and `missing_ok` is `False`
"""
config_data = await json_read(config_file)
try:
nested_delete(config_data, *(*path, key))
except KeyError as exc:
if not missing_ok:
raise exc from exc
await json_write(config_data, config_file)

22
src/libbot/_utils.py
View File

@ -1,22 +0,0 @@
import inspect
from typing import Callable
def supports_argument(func: Callable, arg_name: str) -> bool:
"""Check whether a function has a specific argument
### Args:
* func (`Callable`): Function to be inspected
* arg_name (`str`): Argument to be checked
### Returns:
* `bool`: `True` if argument is supported and `False` if not
"""
if hasattr(func, "__code__"):
return arg_name in inspect.signature(func).parameters
elif hasattr(func, "__doc__"):
if doc := func.__doc__:
first_line = doc.splitlines()[0]
return arg_name in first_line
return False

120
src/libbot/i18n/__init__.py
View File

@ -1,118 +1,2 @@
from os import listdir
from pathlib import Path
from typing import Any, Dict, Union
import libbot
from libbot.i18n import sync
from libbot.i18n.classes.bot_locale import BotLocale
async def _(
key: str,
*args: str,
locale: Union[str, None] = "en",
locales_root: Union[str, Path] = Path("locale"),
) -> Any:
"""Get value of locale string
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locale (`Union[str, None]`): Locale to looked up in. Defaults to `"en"`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Any`: Value of provided locale key. Is usually `str`, `dict` or `list`
"""
locale = libbot.sync.config_get("locale") if locale is None else locale
try:
this_dict = await libbot.json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
try:
this_dict = await libbot.json_read(
Path(f'{locales_root}/{await libbot.config_get("locale")}.json')
)
except FileNotFoundError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
return this_key[key]
except KeyError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
async def in_all_locales(
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> list:
"""Get value of the provided key and path in all available locales
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `list`: List of values in all locales
"""
output = []
files_locales = listdir(locales_root)
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales:
try:
this_dict = await libbot.json_read(Path(f"{locales_root}/{lc}.json"))
except FileNotFoundError:
continue
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output.append(this_key[key])
except KeyError:
continue
return output
async def in_every_locale(
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> Dict[str, Any]:
"""Get value of the provided key and path in every available locale with locale tag
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Dict[str, Any]`: Locale is a key and it's value from locale file is a value
"""
output = {}
files_locales = listdir(locales_root)
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales:
try:
this_dict = await libbot.json_read(Path(f"{locales_root}/{lc}.json"))
except FileNotFoundError:
continue
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output[lc] = this_key[key]
except KeyError:
continue
return output
from ._functions import _, in_all_locales, in_every_locale
from .classes import BotLocale

232
src/libbot/i18n/_functions.py Normal file
View File

@ -0,0 +1,232 @@
from os import listdir, PathLike
from pathlib import Path
from typing import Any, Dict, Union, List
from ..utils.config import config_get
from ..utils.json import json_read
from ..utils.syncs import asyncable
def _get_valid_locales(locales_root: Union[str, PathLike[str]]) -> List[str]:
return [".".join(entry.split(".")[:-1]) for entry in listdir(locales_root)]
@asyncable
def _(
key: str,
*args: str,
locale: Union[str, None] = "en",
locales_root: Union[str, Path] = Path("locale"),
) -> Any:
"""Get value of locale string
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locale (`Union[str, None]`): Locale to looked up in. Defaults to `"en"`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Any`: Value of provided locale key. Is usually `str`, `dict` or `list`
"""
if locale is None:
locale: str = config_get("locale")
try:
this_dict: Dict[str, Any] = json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
try:
this_dict: Dict[str, Any] = json_read(Path(f'{locales_root}/{config_get("locale")}.json'))
except FileNotFoundError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
return this_key[key]
except KeyError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
@_.asynchronous
async def _(
key: str,
*args: str,
locale: Union[str, None] = "en",
locales_root: Union[str, Path] = Path("locale"),
) -> Any:
"""Get value of locale string
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locale (`Union[str, None]`): Locale to looked up in. Defaults to `"en"`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Any`: Value of provided locale key. Is usually `str`, `dict` or `list`
"""
locale: str = config_get("locale") if locale is None else locale
try:
this_dict: Dict[str, Any] = await json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
try:
this_dict: Dict[str, Any] = await json_read(
Path(f'{locales_root}/{await config_get("locale")}.json')
)
except FileNotFoundError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
return this_key[key]
except KeyError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
@asyncable
def in_all_locales(key: str, *args: str, locales_root: Union[str, Path] = Path("locale")) -> list:
"""Get value of the provided key and path in all available locales
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `list`: List of values in all locales
"""
output: List[str] = []
for locale in _get_valid_locales(locales_root):
try:
this_dict: Dict[str, Any] = json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
continue
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output.append(this_key[key])
except KeyError:
continue
return output
@in_all_locales.asynchronous
async def in_all_locales(key: str, *args: str, locales_root: Union[str, Path] = Path("locale")) -> list:
"""Get value of the provided key and path in all available locales
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `list`: List of values in all locales
"""
output = []
for locale in _get_valid_locales(locales_root):
try:
this_dict: Dict[str, Any] = await json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
continue
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output.append(this_key[key])
except KeyError:
continue
return output
@asyncable
def in_every_locale(
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> Dict[str, Any]:
"""Get value of the provided key and path in every available locale with locale tag
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Dict[str, Any]`: Locale is a key and it's value from locale file is a value
"""
output: Dict[str, Any] = {}
for locale in _get_valid_locales(locales_root):
try:
this_dict: Dict[str, Any] = json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
continue
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output[locale] = this_key[key]
except KeyError:
continue
return output
@in_every_locale.asynchronous
async def in_every_locale(
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> Dict[str, Any]:
"""Get value of the provided key and path in every available locale with locale tag
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Dict[str, Any]`: Locale is a key and it's value from locale file is a value
"""
output: Dict[str, Any] = {}
for locale in _get_valid_locales(locales_root):
try:
this_dict: Dict[str, Any] = await json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
continue
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output[locale] = this_key[key]
except KeyError:
continue
return output

1
src/libbot/i18n/classes/__init__.py Normal file
View File

@ -0,0 +1 @@
from .bot_locale import BotLocale

55
src/libbot/i18n/classes/bot_locale.py
View File

@ -1,8 +1,9 @@
from os import listdir
from pathlib import Path
from typing import Any, Dict, Union
from typing import Any, Dict, Union, List
from libbot import sync
from ...utils.config import config_get
from ...utils.json import json_read
class BotLocale:
@ -18,19 +19,15 @@ class BotLocale:
elif not isinstance(locales_root, Path):
raise TypeError("'locales_root' must be a valid path or path-like object")
files_locales: list = listdir(locales_root)
files_locales: List[str] = listdir(locales_root)
valid_locales: list = [
".".join(entry.split(".")[:-1]) for entry in files_locales
]
valid_locales: List[str] = [".".join(entry.split(".")[:-1]) for entry in files_locales]
self.default: str = (
sync.config_get("locale") if default_locale is None else default_locale
)
self.locales: dict = {}
self.default: str = config_get("locale") if default_locale is None else default_locale
self.locales: Dict[str, Any] = {}
for lc in valid_locales:
self.locales[lc] = sync.json_read(Path(f"{locales_root}/{lc}.json"))
for locale in valid_locales:
self.locales[locale] = json_read(Path(f"{locales_root}/{locale}.json"))
def _(self, key: str, *args: str, locale: Union[str, None] = None) -> Any:
"""Get value of locale string
@ -43,19 +40,21 @@ class BotLocale:
### Returns:
* `Any`: Value of provided locale key. Is usually `str`, `dict` or `list`
"""
if locale is None:
locale = self.default
locale: str = self.default
try:
this_dict = self.locales[locale]
this_dict: Dict[str, Any] = self.locales[locale]
except KeyError:
try:
this_dict = self.locales[self.default]
this_dict: Dict[str, Any] = self.locales[self.default]
except KeyError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
return (
f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
)
this_key: Dict[str, Any] = this_dict
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
@ -74,16 +73,16 @@ class BotLocale:
### Returns:
* `list`: List of values in all locales
"""
output: List[str] = []
output = []
for name, lc in self.locales.items():
for name, locale in self.locales.items():
try:
this_dict = lc
this_dict: Dict[str, Any] = locale
except KeyError:
continue
this_key = this_dict
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]
@ -104,16 +103,16 @@ class BotLocale:
### Returns:
* `Dict[str, Any]`: Locale is a key and it's value from locale file is a value
"""
output: Dict[str, Any] = {}
output = {}
for name, lc in self.locales.items():
for name, locale in self.locales.items():
try:
this_dict = lc
this_dict: Dict[str, Any] = locale
except KeyError:
continue
this_key = this_dict
this_key: Dict[str, Any] = this_dict
for dict_key in args:
this_key = this_key[dict_key]

117
src/libbot/i18n/sync/__init__.py
View File

@ -1,117 +0,0 @@
from os import listdir
from pathlib import Path
from typing import Any, Dict, Union
import libbot
def _(
key: str,
*args: str,
locale: Union[str, None] = "en",
locales_root: Union[str, Path] = Path("locale"),
) -> Any:
"""Get value of locale string
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locale (`Union[str, None]`): Locale to looked up in. Defaults to `"en"`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Any`: Value of provided locale key. Is usually `str`, `dict` or `list`
"""
if locale is None:
locale = libbot.sync.config_get("locale")
try:
this_dict = libbot.sync.json_read(Path(f"{locales_root}/{locale}.json"))
except FileNotFoundError:
try:
this_dict = libbot.sync.json_read(
Path(f'{locales_root}/{libbot.sync.config_get("locale")}.json')
)
except FileNotFoundError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
return this_key[key]
except KeyError:
return f'⚠️ Locale in config is invalid: could not get "{key}" in {args} from locale "{locale}"'
def in_all_locales(
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> list:
"""Get value of the provided key and path in all available locales
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `list`: List of values in all locales
"""
output = []
files_locales = listdir(locales_root)
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales:
try:
this_dict = libbot.sync.json_read(Path(f"{locales_root}/{lc}.json"))
except FileNotFoundError:
continue
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output.append(this_key[key])
except KeyError:
continue
return output
def in_every_locale(
key: str, *args: str, locales_root: Union[str, Path] = Path("locale")
) -> Dict[str, Any]:
"""Get value of the provided key and path in every available locale with locale tag
### Args:
* key (`str`): The last key of the locale's keys path.
* *args (`list`): Path to key like: `dict[args][key]`.
* locales_root (`Union[str, Path]`, *optional*): Folder where locales are located. Defaults to `Path("locale")`.
### Returns:
* `Dict[str, Any]`: Locale is a key and it's value from locale file is a value
"""
output = {}
files_locales = listdir(locales_root)
valid_locales = [".".join(entry.split(".")[:-1]) for entry in files_locales]
for lc in valid_locales:
try:
this_dict = libbot.sync.json_read(Path(f"{locales_root}/{lc}.json"))
except FileNotFoundError:
continue
this_key = this_dict
for dict_key in args:
this_key = this_key[dict_key]
try:
output[lc] = this_key[key]
except KeyError:
continue
return output

0
src/libbot/pycord/__init__.py Normal file
View File

20
src/libbot/pycord/classes/bot.py
View File

@ -1,35 +1,35 @@
import logging
from pathlib import Path
from typing import Any, Dict, List, Union
from typing import Any, Dict, Union
from typing_extensions import override
try:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from discord import Bot
except ImportError as exc:
raise ImportError(
"You need to install libbot[pycord] in order to use this class."
) from exc
raise ImportError("You need to install libbot[pycord] in order to use this class.") from exc
try:
from ujson import loads
except ImportError:
from json import loads
from libbot.i18n import BotLocale
from libbot.i18n.sync import _
from ...i18n.classes import BotLocale
logger = logging.getLogger(__name__)
class PycordBot(Bot):
@override
def __init__(
self,
*args,
config: Union[Dict[str, Any], None] = None,
config_path: Union[str, Path] = Path("config.json"),
locales_root: Union[str, Path, None] = None,
scheduler: Union[AsyncIOScheduler, BackgroundScheduler, None] = None,
*args,
**kwargs,
):
if config is None:
@ -39,9 +39,7 @@ class PycordBot(Bot):
self.config = config
super().__init__(
debug_guilds=(
self.config["bot"]["debug_guilds"] if self.config["debug"] else None
),
debug_guilds=(self.config["bot"]["debug_guilds"] if self.config["debug"] else None),
owner_ids=self.config["bot"]["owners"],
*args,
**kwargs,
@ -60,12 +58,14 @@ class PycordBot(Bot):
self.scheduler: Union[AsyncIOScheduler, BackgroundScheduler, None] = scheduler
@override
async def start(self, token: str, reconnect: bool = True, scheduler_start: bool = True) -> None:
if self.scheduler is not None and scheduler_start:
self.scheduler.start()
await super().start(token, reconnect=reconnect)
@override
async def close(self, scheduler_shutdown: bool = True, scheduler_wait: bool = True) -> None:
if self.scheduler is not None and scheduler_shutdown:
self.scheduler.shutdown(scheduler_wait)

0
src/libbot/pyrogram/__init__.py Normal file
View File

105
src/libbot/pyrogram/classes/client.py
View File

@ -1,11 +1,14 @@
import asyncio
import logging
import sys
from datetime import datetime, timedelta
from os import cpu_count, getpid
from pathlib import Path
from time import time
from typing import Any, Dict, List, Union
from typing_extensions import override
try:
import pyrogram
from apscheduler.schedulers.asyncio import AsyncIOScheduler
@ -25,24 +28,23 @@ try:
BotCommandScopeDefault,
)
except ImportError as exc:
raise ImportError(
"You need to install libbot[pyrogram] in order to use this class."
) from exc
raise ImportError("You need to install libbot[pyrogram] in order to use this class.") from exc
try:
from ujson import dumps, loads
except ImportError:
from json import dumps, loads
from libbot.i18n import BotLocale
from libbot.i18n.sync import _
from libbot.pyrogram.classes.command import PyroCommand
from libbot.pyrogram.classes.commandset import CommandSet
from ...i18n.classes import BotLocale
from ...i18n import _
from .command import PyroCommand
from .commandset import CommandSet
logger = logging.getLogger(__name__)
class PyroClient(Client):
@override
def __init__(
self,
name: str = "bot_client",
@ -74,26 +76,20 @@ class PyroClient(Client):
name=name,
api_id=self.config["bot"]["api_id"] if api_id is None else api_id,
api_hash=self.config["bot"]["api_hash"] if api_hash is None else api_hash,
bot_token=self.config["bot"]["bot_token"]
if bot_token is None
else bot_token,
bot_token=self.config["bot"]["bot_token"] if bot_token is None else bot_token,
# Workers should be `min(32, cpu_count() + 4)`, otherwise
# handlers land in another event loop and you won't see them
workers=self.config["bot"]["workers"]
if "workers" in self.config["bot"]
else workers,
workers=self.config["bot"]["workers"] if "workers" in self.config["bot"] else workers,
plugins=dict(
root=plugins_root,
exclude=self.config["disabled_plugins"]
if plugins_exclude is None
else plugins_exclude,
exclude=self.config["disabled_plugins"] if plugins_exclude is None else plugins_exclude,
),
sleep_threshold=sleep_threshold,
max_concurrent_transmissions=self.config["bot"][
"max_concurrent_transmissions"
]
if "max_concurrent_transmissions" in self.config["bot"]
else max_concurrent_transmissions,
max_concurrent_transmissions=(
self.config["bot"]["max_concurrent_transmissions"]
if "max_concurrent_transmissions" in self.config["bot"]
else max_concurrent_transmissions
),
**kwargs,
)
self.owner: int = self.config["bot"]["owner"] if owner is None else owner
@ -102,9 +98,7 @@ class PyroClient(Client):
self.config["commands"] if commands_source is None else commands_source
)
self.scoped_commands: bool = (
self.config["bot"]["scoped_commands"]
if scoped_commands is None
else scoped_commands
self.config["bot"]["scoped_commands"] if scoped_commands is None else scoped_commands
)
self.start_time: float = 0
@ -125,6 +119,7 @@ class PyroClient(Client):
self.i18n_bot_info: bool = i18n_bot_info
@override
async def start(self, register_commands: bool = True, scheduler_start: bool = True) -> None:
await super().start()
@ -189,9 +184,11 @@ class PyroClient(Client):
# Send a message to the bot's reports chat about the startup
try:
await self.send_message(
chat_id=self.owner
if self.config["reports"]["chat_id"] == "owner"
else self.config["reports"]["chat_id"],
chat_id=(
self.owner
if self.config["reports"]["chat_id"] == "owner"
else self.config["reports"]["chat_id"]
),
text=f"Bot started PID `{getpid()}`",
)
except BadRequest:
@ -212,14 +209,17 @@ class PyroClient(Client):
if scheduler_start:
self.scheduler.start()
@override
async def stop(
self, exit_completely: bool = True, scheduler_shutdown: bool = True, scheduler_wait: bool = True
) -> None:
try:
await self.send_message(
chat_id=self.owner
if self.config["reports"]["chat_id"] == "owner"
else self.config["reports"]["chat_id"],
chat_id=(
self.owner
if self.config["reports"]["chat_id"] == "owner"
else self.config["reports"]["chat_id"]
),
text=f"Bot stopped with PID `{getpid()}`",
)
await asyncio.sleep(0.5)
@ -234,11 +234,9 @@ class PyroClient(Client):
if exit_completely:
try:
exit()
sys.exit()
except SystemExit as exc:
raise SystemExit(
"Bot has been shut down, this is not an application error!"
) from exc
raise SystemExit("Bot has been shut down, this is not an application error!") from exc
async def collect_commands(self) -> Union[List[CommandSet], None]:
"""Gather list of the bot's commands
@ -262,13 +260,9 @@ class PyroClient(Client):
scopes[dumps(scope)] = {"_": []}
# Add command to the scope's flattened key in scopes dict
scopes[dumps(scope)]["_"].append(
BotCommand(command, _(command, "commands"))
)
scopes[dumps(scope)]["_"].append(BotCommand(command, _(command, "commands")))
for locale, string in (
self.in_every_locale(command, "commands")
).items():
for locale, string in (self.in_every_locale(command, "commands")).items():
if locale not in scopes[dumps(scope)]:
scopes[dumps(scope)][locale] = []
@ -287,11 +281,7 @@ class PyroClient(Client):
# Create object with the same name and args from the dict
try:
scope_obj = globals()[scope_dict["name"]](
**{
key: value
for key, value in scope_dict.items()
if key != "name"
}
**{key: value for key, value in scope_dict.items() if key != "name"}
)
except NameError:
logger.error(
@ -309,13 +299,9 @@ class PyroClient(Client):
# Add set of commands to the list of the command sets
for locale, commands in locales.items():
if locale == "_":
command_sets.append(
CommandSet(commands, scope=scope_obj, language_code="")
)
command_sets.append(CommandSet(commands, scope=scope_obj, language_code=""))
continue
command_sets.append(
CommandSet(commands, scope=scope_obj, language_code=locale)
)
command_sets.append(CommandSet(commands, scope=scope_obj, language_code=locale))
logger.info("Registering the following command sets: %s", command_sets)
@ -352,9 +338,7 @@ class PyroClient(Client):
command,
)
async def register_commands(
self, command_sets: Union[List[CommandSet], None] = None
) -> None:
async def register_commands(self, command_sets: Union[List[CommandSet], None] = None) -> None:
"""Register commands stored in bot's 'commands' attribute"""
if command_sets is None:
@ -363,10 +347,7 @@ class PyroClient(Client):
for command in self.commands
]
logger.info(
"Registering commands %s with a default scope 'BotCommandScopeDefault'",
commands
)
logger.info("Registering commands %s with a default scope 'BotCommandScopeDefault'", commands)
await self.set_bot_commands(commands)
return
@ -384,15 +365,11 @@ class PyroClient(Client):
language_code=command_set.language_code,
)
async def remove_commands(
self, command_sets: Union[List[CommandSet], None] = None
) -> None:
async def remove_commands(self, command_sets: Union[List[CommandSet], None] = None) -> None:
"""Remove commands stored in bot's 'commands' attribute"""
if command_sets is None:
logger.info(
"Removing commands with a default scope 'BotCommandScopeDefault'"
)
logger.info("Removing commands with a default scope 'BotCommandScopeDefault'")
await self.delete_bot_commands(BotCommandScopeDefault())
return

1
src/libbot/sync/__init__.py
View File

@ -1 +0,0 @@
from .__main__ import config_delete, config_get, config_set, json_read, json_write

127
src/libbot/sync/__main__.py
View File

@ -1,127 +0,0 @@
from pathlib import Path
from typing import Any, Union
from .._utils import supports_argument
from ._nested import nested_delete, nested_set
try:
from ujson import dumps, loads
except ImportError:
from json import dumps, loads
DEFAULT_CONFIG_LOCATION: str = "config.json"
def json_read(path: Union[str, Path]) -> Any:
"""Read contents of a JSON file
### Args:
* path (`Union[str, Path]`): Path-like object or path as a string
### Returns:
* `Any`: File contents
"""
with open(str(path), mode="r", encoding="utf-8") as f:
data = f.read()
return loads(data)
def json_write(data: Any, path: Union[str, Path]) -> None:
"""Write contents to a JSON file
### Args:
* data (`Any`): Contents to write. Must be a JSON serializable
* path (`Union[str, Path]`): Path-like object or path as a string of a destination
"""
with open(str(path), mode="w", encoding="utf-8") as f:
f.write(
dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4)
if supports_argument(dumps, "escape_forward_slashes")
else dumps(data, ensure_ascii=False, indent=4)
)
def config_get(
key: str, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION
) -> Any:
"""Get a value of the config key by its path provided
For example, `foo.bar.key` has a path of `"foo", "bar"` and the key `"key"`
### Args:
* key (`str`): Key that contains the value
* *path (`str`): Path to the key that contains the value
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Returns:
* `Any`: Key's value
### Example:
Get the "salary" of "Pete" from this JSON structure:
```json
{
"users": {
"Pete": {
"salary": 10.0
}
}
}
```
This can be easily done with the following code:
```python
import libbot
salary = libbot.sync.config_get("salary", "users", "Pete")
```
"""
this_key = json_read(config_file)
for dict_key in path:
this_key = this_key[dict_key]
return this_key[key]
def config_set(
key: str, value: Any, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION
) -> None:
"""Set config's key by its path to the value
### Args:
* key (`str`): Key that leads to the value
* value (`Any`): Any JSON serializable data
* *path (`str`): Path to the key of the target
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided
"""
json_write(nested_set(json_read(config_file), value, *(*path, key)), config_file)
def config_delete(
key: str,
*path: str,
missing_ok: bool = False,
config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION,
) -> None:
"""Set config's key by its path
### Args:
* key (`str`): Key to delete
* *path (`str`): Path to the key of the target
* missing_ok (`bool`): Do not raise an exception if the key is missing. Defaults to `False`
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided and `missing_ok` is `False`
"""
config_data = json_read(config_file)
try:
nested_delete(config_data, *(*path, key))
except KeyError as exc:
if not missing_ok:
raise exc from exc
json_write(config_data, config_file)

3
src/libbot/utils/__init__.py Normal file
View File

@ -0,0 +1,3 @@
from . import misc
from .config import config_get, config_set, config_delete
from .json import json_read, json_write

1
src/libbot/utils/config/__init__.py Normal file
View File

@ -0,0 +1 @@
from ._functions import config_get, config_set, config_delete

185
src/libbot/utils/config/_functions.py Normal file
View File

@ -0,0 +1,185 @@
from pathlib import Path
from typing import Any, Union, Dict
from ..json import json_read, json_write
from ..misc import nested_delete, nested_set
from ..syncs import asyncable
try:
from ujson import dumps, loads
except ImportError:
from json import dumps, loads
DEFAULT_CONFIG_LOCATION: str = "config.json"
@asyncable
def config_get(key: str, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION) -> Any:
"""Get a value of the config key by its path provided
For example, `foo.bar.key` has a path of `"foo", "bar"` and the key `"key"`
### Args:
* key (`str`): Key that contains the value
* *path (`str`): Path to the key that contains the value
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Returns:
* `Any`: Key's value
### Example:
Get the "salary" of "Pete" from this JSON structure:
```json
{
"users": {
"Pete": {
"salary": 10.0
}
}
}
```
This can be easily done with the following code:
```python
import libbot
salary = libbot.sync.config_get("salary", "users", "Pete")
```
"""
this_key: Dict[str, Any] = json_read(config_file)
for dict_key in path:
this_key = this_key[dict_key]
return this_key[key]
@config_get.asynchronous
async def config_get(key: str, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION) -> Any:
"""Get a value of the config key by its path provided
For example, `foo.bar.key` has a path of `"foo", "bar"` and the key `"key"`
### Args:
* key (`str`): Key that contains the value
* *path (`str`): Path to the key that contains the value
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Returns:
* `Any`: Key's value
### Example:
Get the "salary" of "Pete" from this JSON structure:
```json
{
"users": {
"Pete": {
"salary": 10.0
}
}
}
```
This can be easily done with the following code:
```python
import libbot
salary = await libbot.config_get("salary", "users", "Pete")
```
"""
this_key: Dict[str, Any] = await json_read(config_file)
for dict_key in path:
this_key = this_key[dict_key]
return this_key[key]
@asyncable
def config_set(
key: str, value: Any, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION
) -> None:
"""Set config's key by its path to the value
### Args:
* key (`str`): Key that leads to the value
* value (`Any`): Any JSON serializable data
* *path (`str`): Path to the key of the target
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided
"""
json_write(nested_set(json_read(config_file), value, *(*path, key)), config_file)
@config_set.asynchronous
async def config_set(
key: str, value: Any, *path: str, config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION
) -> None:
"""Set config's key by its path to the value
### Args:
* key (`str`): Key that leads to the value
* value (`Any`): Any JSON serializable data
* *path (`str`): Path to the key of the target
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided
"""
await json_write(nested_set(await json_read(config_file), value, *(*path, key)), config_file)
@asyncable
def config_delete(
key: str,
*path: str,
missing_ok: bool = False,
config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION,
) -> None:
"""Set config's key by its path
### Args:
* key (`str`): Key to delete
* *path (`str`): Path to the key of the target
* missing_ok (`bool`): Do not raise an exception if the key is missing. Defaults to `False`
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided and `missing_ok` is `False`
"""
config_data: Dict[str, Any] = json_read(config_file)
try:
nested_delete(config_data, *(*path, key))
except KeyError as exc:
if not missing_ok:
raise exc from exc
json_write(config_data, config_file)
@config_delete.asynchronous
async def config_delete(
key: str,
*path: str,
missing_ok: bool = False,
config_file: Union[str, Path] = DEFAULT_CONFIG_LOCATION,
) -> None:
"""Set config's key by its path
### Args:
* key (`str`): Key to delete
* *path (`str`): Path to the key of the target
* missing_ok (`bool`): Do not raise an exception if the key is missing. Defaults to `False`
* config_file (`Union[str, Path]`, *optional*): Path-like object or path as a string of a location of the config file. Defaults to `"config.json"`
### Raises:
* `KeyError`: Key is not found under path provided and `missing_ok` is `False`
"""
config_data: Dict[str, Any] = await json_read(config_file)
try:
nested_delete(config_data, *(*path, key))
except KeyError as exc:
if not missing_ok:
raise exc from exc
await json_write(config_data, config_file)

1
src/libbot/utils/json/__init__.py Normal file
View File

@ -0,0 +1 @@
from ._functions import json_read, json_write

76
src/libbot/utils/json/_functions.py Normal file
View File

@ -0,0 +1,76 @@
from pathlib import Path
from typing import Any, Union
import aiofiles
from ..misc import supports_argument
from ..syncs import asyncable
try:
from ujson import dumps, loads
except ImportError:
from json import dumps, loads
@asyncable
def json_read(path: Union[str, Path]) -> Any:
"""Read contents of a JSON file
### Args:
* path (`Union[str, Path]`): Path-like object or path as a string
### Returns:
* `Any`: File contents
"""
with open(str(path), mode="r", encoding="utf-8") as f:
data = f.read()
return loads(data)
@json_read.asynchronous
async def json_read(path: Union[str, Path]) -> Any:
"""Read contents of a JSON file
### Args:
* path (`Union[str, Path]`): Path-like object or path as a string
### Returns:
* `Any`: File contents
"""
async with aiofiles.open(str(path), mode="r", encoding="utf-8") as f:
data = await f.read()
return loads(data)
@asyncable
def json_write(data: Any, path: Union[str, Path]) -> None:
"""Write contents to a JSON file
### Args:
* data (`Any`): Contents to write. Must be a JSON serializable
* path (`Union[str, Path]`): Path-like object or path as a string of a destination
"""
with open(str(path), mode="w", encoding="utf-8") as f:
f.write(
dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4)
if supports_argument(dumps, "escape_forward_slashes")
else dumps(data, ensure_ascii=False, indent=4)
)
@json_write.asynchronous
async def json_write(data: Any, path: Union[str, Path]) -> None:
"""Write contents to a JSON file
### Args:
* data (`Any`): Contents to write. Must be a JSON serializable
* path (`Union[str, Path]`): Path-like object or path as a string of a destination
"""
async with aiofiles.open(str(path), mode="w", encoding="utf-8") as f:
await f.write(
dumps(data, ensure_ascii=False, escape_forward_slashes=False, indent=4)
if supports_argument(dumps, "escape_forward_slashes")
else dumps(data, ensure_ascii=False, indent=4)
)

1
src/libbot/utils/misc/__init__.py Normal file
View File

@ -0,0 +1 @@
from ._functions import supports_argument, nested_set, nested_delete

27
src/libbot/sync/_nested.py → src/libbot/utils/misc/_functions.py
View File

@ -1,9 +1,30 @@
import inspect
from typing import Any, Dict
from typing import Callable
def nested_set(
target: dict, value: Any, *path: str, create_missing=True
) -> Dict[str, Any]:
def supports_argument(func: Callable, arg_name: str) -> bool:
"""Check whether a function has a specific argument
### Args:
* func (`Callable`): Function to be inspected
* arg_name (`str`): Argument to be checked
### Returns:
* `bool`: `True` if argument is supported and `False` if not
"""
if hasattr(func, "__code__"):
return arg_name in inspect.signature(func).parameters
if hasattr(func, "__doc__"):
if doc := func.__doc__:
first_line = doc.splitlines()[0]
return arg_name in first_line
return False
def nested_set(target: dict, value: Any, *path: str, create_missing=True) -> Dict[str, Any]:
"""Set the key by its path to the value
### Args:

1
src/libbot/utils/syncs/__init__.py Normal file
View File

@ -0,0 +1 @@
from ._syncs import asyncable

69
src/libbot/utils/syncs/_syncs.py Normal file
View File

@ -0,0 +1,69 @@
import asyncio
import inspect
from inspect import FrameInfo
from typing import Any, Callable, Optional, Type
class asyncable:
"""Allows to mark a callable able to be async.
Source: https://itsjohannawren.medium.com/single-call-sync-and-async-in-python-2acadd07c9d6"""
def __init__(self, method: Callable):
self.__sync = method
self.__async = None
def asynchronous(self, method: Callable) -> "asyncable":
if not isinstance(method, Callable):
raise RuntimeError("NOT CALLABLE!!!")
self.__async = method
return self
@staticmethod
def __is_awaited() -> bool:
frame: FrameInfo = inspect.stack()[2]
if not hasattr(frame, "positions"):
return False
return (
frame.positions.col_offset >= 6
and frame.code_context[frame.index][frame.positions.col_offset - 6 : frame.positions.col_offset]
== "await "
)
def __get__(
self,
instance: Type,
*args,
owner_class: Optional[Type[Type]] = None,
**kwargs,
) -> Callable:
if self.__is_awaited():
if self.__async is None:
raise RuntimeError(
"Attempting to call asyncable with await, but no asynchronous call has been defined"
)
bound_method = self.__async.__get__(instance, owner_class)
if isinstance(self.__sync, classmethod):
return lambda: asyncio.ensure_future(bound_method(owner_class, *args, **kwargs))
return lambda: asyncio.ensure_future(bound_method(*args, **kwargs))
bound_method = self.__sync.__get__(instance, owner_class)
return lambda: bound_method(*args, **kwargs)
def __call__(self, *args, **kwargs) -> Any:
if self.__is_awaited():
if self.__async is None:
raise RuntimeError(
"Attempting to call asyncable with await, but no asynchronous call has been defined"
)
return asyncio.ensure_future(self.__async(*args, **kwargs))
return self.__sync(*args, **kwargs)

17
tests/test_config_async.py
View File

@ -2,8 +2,7 @@ from pathlib import Path
from typing import Any, List
import pytest
from libbot import config_delete, config_get, config_set
from libbot.utils import config_delete, config_get, config_set
@pytest.mark.asyncio
@ -25,14 +24,9 @@ async def test_config_get_valid(args: List[str], expected: str, location_config:
(["bot_stonks", "bot"], pytest.raises(KeyError)),
],
)
async def test_config_get_invalid(
args: List[str], expected: Any, location_config: Path
):
async def test_config_get_invalid(args: List[str], expected: Any, location_config: Path):
with expected:
assert (
await config_get(args[0], *args[1:], config_file=location_config)
== expected
)
assert await config_get(args[0], *args[1:], config_file=location_config) == expected
@pytest.mark.asyncio
@ -68,7 +62,4 @@ async def test_config_delete(key: str, path: List[str], location_config: Path):
],
)
async def test_config_delete_missing(key: str, path: List[str], location_config: Path):
assert (
await config_delete(key, *path, missing_ok=True, config_file=location_config)
is None
)
assert await config_delete(key, *path, missing_ok=True, config_file=location_config) is None

24
tests/test_config_sync.py
View File

@ -2,8 +2,7 @@ from pathlib import Path
from typing import Any, List
import pytest
from libbot import sync
from libbot.utils import config_delete, config_get, config_set
@pytest.mark.parametrize(
@ -13,8 +12,8 @@ from libbot import sync
(["bot_token", "bot"], "sample_token"),
],
)
def test_config_get_valid(args: List[str], expected: str, location_config: Path):
assert sync.config_get(args[0], *args[1:], config_file=location_config) == expected
async def test_config_get_valid(args: List[str], expected: str, location_config: Path):
assert config_get(args[0], *args[1:], config_file=location_config) == expected
@pytest.mark.parametrize(
@ -25,9 +24,7 @@ def test_config_get_valid(args: List[str], expected: str, location_config: Path)
)
def test_config_get_invalid(args: List[str], expected: Any, location_config: Path):
with expected:
assert (
sync.config_get(args[0], *args[1:], config_file=location_config) == expected
)
assert config_get(args[0], *args[1:], config_file=location_config) == expected
@pytest.mark.parametrize(
@ -38,8 +35,8 @@ def test_config_get_invalid(args: List[str], expected: Any, location_config: Pat
],
)
def test_config_set(key: str, path: List[str], value: Any, location_config: Path):
sync.config_set(key, value, *path, config_file=location_config)
assert sync.config_get(key, *path, config_file=location_config) == value
config_set(key, value, *path, config_file=location_config)
assert config_get(key, *path, config_file=location_config) == value
@pytest.mark.parametrize(
@ -49,8 +46,8 @@ def test_config_set(key: str, path: List[str], value: Any, location_config: Path
],
)
def test_config_delete(key: str, path: List[str], location_config: Path):
sync.config_delete(key, *path, config_file=location_config)
assert key not in sync.config_get(*path, config_file=location_config)
config_delete(key, *path, config_file=location_config)
assert key not in config_get(*path, config_file=location_config)
@pytest.mark.parametrize(
@ -60,7 +57,4 @@ def test_config_delete(key: str, path: List[str], location_config: Path):
],
)
def test_config_delete_missing(key: str, path: List[str], location_config: Path):
assert (
sync.config_delete(key, *path, missing_ok=True, config_file=location_config)
is None
)
assert config_delete(key, *path, missing_ok=True, config_file=location_config) is None

19
tests/test_i18n_sync.py
View File

@ -2,8 +2,7 @@ from pathlib import Path
from typing import Any, List, Union
import pytest
from libbot.i18n import sync
from libbot.i18n import _, in_all_locales, in_every_locale
@pytest.mark.parametrize(
@ -25,9 +24,9 @@ def test_i18n_get(
location_locale: Path,
):
assert (
sync._(key, *args, locale=locale, locales_root=location_locale)
_(key, *args, locale=locale, locales_root=location_locale)
if locale is not None
else sync._(key, *args, locales_root=location_locale)
else _(key, *args, locales_root=location_locale)
) == expected
@ -39,10 +38,8 @@ def test_i18n_get(
("nested", ["callbacks", "default"], ["sure", "авжеж"]),
],
)
def test_i18n_in_all_locales(
key: str, args: List[str], expected: Any, location_locale: Path
):
assert (sync.in_all_locales(key, *args, locales_root=location_locale)) == expected
def test_i18n_in_all_locales(key: str, args: List[str], expected: Any, location_locale: Path):
assert (in_all_locales(key, *args, locales_root=location_locale)) == expected
@pytest.mark.parametrize(
@ -53,7 +50,5 @@ def test_i18n_in_all_locales(
("nested", ["callbacks", "default"], {"en": "sure", "uk": "авжеж"}),
],
)
def test_i18n_in_every_locale(
key: str, args: List[str], expected: Any, location_locale: Path
):
assert (sync.in_every_locale(key, *args, locales_root=location_locale)) == expected
def test_i18n_in_every_locale(key: str, args: List[str], expected: Any, location_locale: Path):
assert (in_every_locale(key, *args, locales_root=location_locale)) == expected

3
tests/test_json_async.py
View File

@ -7,8 +7,7 @@ from pathlib import Path
from typing import Any, Union
import pytest
from libbot import json_read, json_write
from libbot.utils import json_read, json_write
@pytest.mark.asyncio

9
tests/test_json_sync.py
View File

@ -7,8 +7,7 @@ from pathlib import Path
from typing import Any, Union
import pytest
from libbot import sync
from libbot.utils import json_read, json_write
@pytest.mark.parametrize(
@ -25,7 +24,7 @@ from libbot import sync
],
)
def test_json_read_valid(path: Union[str, Path], expected: Any):
assert sync.json_read(path) == expected
assert json_read(path) == expected
@pytest.mark.parametrize(
@ -37,7 +36,7 @@ def test_json_read_valid(path: Union[str, Path], expected: Any):
)
def test_json_read_invalid(path: Union[str, Path], expected: Any):
with pytest.raises(expected):
assert sync.json_read(path) == expected
assert json_read(path) == expected
@pytest.mark.parametrize(
@ -54,7 +53,7 @@ def test_json_read_invalid(path: Union[str, Path], expected: Any):
],
)
def test_json_write(data: Any, path: Union[str, Path]):
sync.json_write(data, path)
json_write(data, path)
assert Path(path).is_file()
with open(path, "r", encoding="utf-8") as f:

7
tests/test_nested_set.py
View File

@ -1,8 +1,7 @@
from typing import Any, Dict, List
import pytest
from libbot.sync._nested import nested_delete, nested_set
from libbot.utils.misc import nested_delete, nested_set
@pytest.mark.parametrize(
@ -56,9 +55,7 @@ def test_nested_set_invalid(
expected: Any,
):
with pytest.raises(expected):
assert (
nested_set(target, value, *path, create_missing=create_missing)
) == expected
assert (nested_set(target, value, *path, create_missing=create_missing)) == expected
@pytest.mark.parametrize(

3
tests/test_utils.py
View File

@ -1,8 +1,7 @@
from typing import Callable
import pytest
from libbot._utils import supports_argument
from libbot.utils.misc import supports_argument
def func1(foo: str, bar: str):

7
tox.ini
View File

@ -1,14 +1,13 @@
[tox]
minversion = 3.9.0
envlist = py39, py310, py311, py312
minversion = 3.11.0
envlist = py311, py312, py313
isolated_build = true
[gh-actions]
python =
3.9: py39
3.10: py310
3.11: py311
3.12: py312
3.13: py313
[testenv]
setenv =