From d5e390fe6620bbd3ab3a2dbdef05d761cbd14ab2 Mon Sep 17 00:00:00 2001 From: kku Date: Wed, 1 Jan 2025 22:34:38 +0100 Subject: [PATCH] Optimized json_load usage, imports and typing --- src/libbot/pycord/classes/bot.py | 33 ++++++-------- src/libbot/pyrogram/classes/client.py | 66 +++++++++++++-------------- 2 files changed, 45 insertions(+), 54 deletions(-) diff --git a/src/libbot/pycord/classes/bot.py b/src/libbot/pycord/classes/bot.py index d5986a1..6390813 100644 --- a/src/libbot/pycord/classes/bot.py +++ b/src/libbot/pycord/classes/bot.py @@ -1,9 +1,13 @@ import logging +from logging import Logger from pathlib import Path from typing import Any, Dict from typing_extensions import override +from ...i18n.classes import BotLocale +from ...utils import json_read + try: from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.schedulers.background import BackgroundScheduler @@ -11,32 +15,21 @@ try: except ImportError as 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 ...i18n.classes import BotLocale - -logger = logging.getLogger(__name__) +logger: Logger = logging.getLogger(__name__) class PycordBot(Bot): @override def __init__( - self, - *args, - config: Dict[str, Any] | None = None, - config_path: str | Path = Path("config.json"), - locales_root: str | Path | None = None, - scheduler: AsyncIOScheduler | BackgroundScheduler | None = None, - **kwargs, + self, + *args, + config: Dict[str, Any] | None = None, + config_path: str | Path = Path("config.json"), + locales_root: str | Path | None = None, + scheduler: AsyncIOScheduler | BackgroundScheduler | None = None, + **kwargs, ): - if config is None: - with open(config_path, "r", encoding="utf-8") as f: - self.config: dict = loads(f.read()) - else: - self.config = config + self.config: Dict[str, Any] = config if config is not None else json_read(config_path) super().__init__( debug_guilds=(self.config["bot"]["debug_guilds"] if self.config["debug"] else None), diff --git a/src/libbot/pyrogram/classes/client.py b/src/libbot/pyrogram/classes/client.py index 0e8a204..2b01db7 100644 --- a/src/libbot/pyrogram/classes/client.py +++ b/src/libbot/pyrogram/classes/client.py @@ -2,6 +2,7 @@ import asyncio import logging import sys from datetime import datetime, timedelta +from logging import Logger from os import cpu_count, getpid from pathlib import Path from time import time @@ -9,6 +10,12 @@ from typing import Any, Dict, List 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: import pyrogram from apscheduler.schedulers.asyncio import AsyncIOScheduler @@ -35,42 +42,33 @@ try: except ImportError: from json import dumps, loads -from ...i18n.classes import BotLocale -from ...i18n import _ -from .command import PyroCommand -from .commandset import CommandSet - -logger = logging.getLogger(__name__) +logger: Logger = logging.getLogger(__name__) class PyroClient(Client): @override def __init__( - self, - name: str = "bot_client", - owner: int | None = None, - config: Dict[str, Any] | None = None, - config_path: str | Path = Path("config.json"), - api_id: int | None = None, - api_hash: str | None = None, - bot_token: str | None = None, - workers: int = min(32, cpu_count() + 4), - locales_root: str | Path | None = None, - plugins_root: str = "plugins", - plugins_exclude: List[str] | None = None, - sleep_threshold: int = 120, - max_concurrent_transmissions: int = 1, - commands_source: Dict[str, dict] | None = None, - scoped_commands: bool | None = None, - i18n_bot_info: bool = False, - scheduler: AsyncIOScheduler | BackgroundScheduler | None = None, - **kwargs, + self, + name: str = "bot_client", + owner: int | None = None, + config: Dict[str, Any] | None = None, + config_path: str | Path = Path("config.json"), + api_id: int | None = None, + api_hash: str | None = None, + bot_token: str | None = None, + workers: int = min(32, cpu_count() + 4), + locales_root: str | Path | None = None, + plugins_root: str = "plugins", + plugins_exclude: List[str] | None = None, + sleep_threshold: int = 120, + max_concurrent_transmissions: int = 1, + commands_source: Dict[str, dict] | None = None, + scoped_commands: bool | None = None, + i18n_bot_info: bool = False, + scheduler: AsyncIOScheduler | BackgroundScheduler | None = None, + **kwargs, ): - if config is None: - with open(config_path, "r", encoding="utf-8") as f: - self.config: dict = loads(f.read()) - else: - self.config = config + self.config: Dict[str, Any] = config if config is not None else json_read(config_path) super().__init__( name=name, @@ -211,7 +209,7 @@ class PyroClient(Client): @override 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: try: await self.send_message( @@ -310,7 +308,7 @@ class PyroClient(Client): # in it, if there are any. Then adds them to self.commands for handler in self.dispatcher.groups[0]: 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]: if hasattr(entry, "commands"): @@ -321,8 +319,8 @@ class PyroClient(Client): return command_sets def add_command( - self, - command: str, + self, + command: str, ) -> None: """Add command to the bot's internal commands list