Optimized json_load usage, imports and typing
All checks were successful
Analysis / SonarCloud (push) Successful in 48s
Analysis / SonarCloud (pull_request) Successful in 39s
Tests / Build and Test (3.11) (pull_request) Successful in 1m22s
Tests / Build and Test (3.12) (pull_request) Successful in 1m27s
Tests / Build and Test (3.13) (pull_request) Successful in 1m29s

This commit is contained in:
kku 2025-01-01 22:34:38 +01:00
parent ae54bd5cce
commit d5e390fe66
2 changed files with 45 additions and 54 deletions

View File

@ -1,9 +1,13 @@
import logging import logging
from logging import Logger
from pathlib import Path from pathlib import Path
from typing import Any, Dict from typing import Any, Dict
from typing_extensions import override from typing_extensions import override
from ...i18n.classes import BotLocale
from ...utils import json_read
try: try:
from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.schedulers.background import BackgroundScheduler
@ -11,32 +15,21 @@ try:
except ImportError as exc: 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: logger: Logger = logging.getLogger(__name__)
from ujson import loads
except ImportError:
from json import loads
from ...i18n.classes import BotLocale
logger = logging.getLogger(__name__)
class PycordBot(Bot): class PycordBot(Bot):
@override @override
def __init__( def __init__(
self, self,
*args, *args,
config: Dict[str, Any] | None = None, config: Dict[str, Any] | None = None,
config_path: str | Path = Path("config.json"), config_path: str | Path = Path("config.json"),
locales_root: str | Path | None = None, locales_root: str | Path | None = None,
scheduler: AsyncIOScheduler | BackgroundScheduler | None = None, scheduler: AsyncIOScheduler | BackgroundScheduler | None = None,
**kwargs, **kwargs,
): ):
if config is None: self.config: Dict[str, Any] = config if config is not None else json_read(config_path)
with open(config_path, "r", encoding="utf-8") as f:
self.config: dict = loads(f.read())
else:
self.config = config
super().__init__( 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),

View File

@ -2,6 +2,7 @@ import asyncio
import logging import logging
import sys import sys
from datetime import datetime, timedelta from datetime import datetime, timedelta
from logging import Logger
from os import cpu_count, getpid from os import cpu_count, getpid
from pathlib import Path from pathlib import Path
from time import time from time import time
@ -9,6 +10,12 @@ from typing import Any, Dict, List
from typing_extensions import override from typing_extensions import override
from .command import PyroCommand
from .commandset import CommandSet
from ...i18n import _
from ...i18n.classes import BotLocale
from ...utils import json_read
try: try:
import pyrogram import pyrogram
from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.schedulers.asyncio import AsyncIOScheduler
@ -35,42 +42,33 @@ try:
except ImportError: except ImportError:
from json import dumps, loads from json import dumps, loads
from ...i18n.classes import BotLocale logger: Logger = logging.getLogger(__name__)
from ...i18n import _
from .command import PyroCommand
from .commandset import CommandSet
logger = logging.getLogger(__name__)
class PyroClient(Client): class PyroClient(Client):
@override @override
def __init__( def __init__(
self, self,
name: str = "bot_client", name: str = "bot_client",
owner: int | None = None, owner: int | None = None,
config: Dict[str, Any] | None = None, config: Dict[str, Any] | None = None,
config_path: str | Path = Path("config.json"), config_path: str | Path = Path("config.json"),
api_id: int | None = None, api_id: int | None = None,
api_hash: str | None = None, api_hash: str | None = None,
bot_token: str | None = None, bot_token: str | None = None,
workers: int = min(32, cpu_count() + 4), workers: int = min(32, cpu_count() + 4),
locales_root: str | Path | None = None, locales_root: str | Path | None = None,
plugins_root: str = "plugins", plugins_root: str = "plugins",
plugins_exclude: List[str] | None = None, plugins_exclude: List[str] | None = None,
sleep_threshold: int = 120, sleep_threshold: int = 120,
max_concurrent_transmissions: int = 1, max_concurrent_transmissions: int = 1,
commands_source: Dict[str, dict] | None = None, commands_source: Dict[str, dict] | None = None,
scoped_commands: bool | None = None, scoped_commands: bool | None = None,
i18n_bot_info: bool = False, i18n_bot_info: bool = False,
scheduler: AsyncIOScheduler | BackgroundScheduler | None = None, scheduler: AsyncIOScheduler | BackgroundScheduler | None = None,
**kwargs, **kwargs,
): ):
if config is None: self.config: Dict[str, Any] = config if config is not None else json_read(config_path)
with open(config_path, "r", encoding="utf-8") as f:
self.config: dict = loads(f.read())
else:
self.config = config
super().__init__( super().__init__(
name=name, name=name,
@ -211,7 +209,7 @@ class PyroClient(Client):
@override @override
async def stop( async def stop(
self, exit_completely: bool = True, scheduler_shutdown: bool = True, scheduler_wait: bool = True self, exit_completely: bool = True, scheduler_shutdown: bool = True, scheduler_wait: bool = True
) -> None: ) -> None:
try: try:
await self.send_message( await self.send_message(
@ -310,7 +308,7 @@ class PyroClient(Client):
# in it, if there are any. Then adds them to self.commands # in it, if there are any. Then adds them to self.commands
for handler in self.dispatcher.groups[0]: for handler in self.dispatcher.groups[0]:
if isinstance(handler, MessageHandler) and ( if isinstance(handler, MessageHandler) and (
hasattr(handler.filters, "base") or hasattr(handler.filters, "other") hasattr(handler.filters, "base") or hasattr(handler.filters, "other")
): ):
for entry in [handler.filters.base, handler.filters.other]: for entry in [handler.filters.base, handler.filters.other]:
if hasattr(entry, "commands"): if hasattr(entry, "commands"):
@ -321,8 +319,8 @@ class PyroClient(Client):
return command_sets return command_sets
def add_command( def add_command(
self, self,
command: str, command: str,
) -> None: ) -> None:
"""Add command to the bot's internal commands list """Add command to the bot's internal commands list