mirror of
https://github.com/allthingslinux/tux.git
synced 2024-10-03 00:53:12 +00:00
refactor(cog_loader.py): replace aiopath with pathlib and aiofiles for better compatibility
fix(cog_loader.py): remove conditional cog_ignore_list based on DEV constant, use a single COG_IGNORE_LIST for simplicity style(cog_loader.py): rename variables for clarity, apath to path, and filepath to path
This commit is contained in:
parent
7925a3cecc
commit
a9bcf45421
1 changed files with 17 additions and 18 deletions
|
@ -1,6 +1,8 @@
|
||||||
import traceback
|
import traceback
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
from aiopath import AsyncPath # type: ignore
|
import aiofiles
|
||||||
|
import aiofiles.os
|
||||||
from discord.ext import commands
|
from discord.ext import commands
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
|
|
||||||
|
@ -10,17 +12,15 @@ from tux.utils.constants import Constants as CONST
|
||||||
class CogLoader(commands.Cog):
|
class CogLoader(commands.Cog):
|
||||||
def __init__(self, bot: commands.Bot) -> None:
|
def __init__(self, bot: commands.Bot) -> None:
|
||||||
self.bot = bot
|
self.bot = bot
|
||||||
self.cog_ignore_list: set[str] = (
|
self.cog_ignore_list: set[str] = CONST.COG_IGNORE_LIST
|
||||||
CONST.PROD_COG_IGNORE_LIST if CONST.DEV == "False" else CONST.DEV_COG_IGNORE_LIST
|
|
||||||
)
|
|
||||||
|
|
||||||
async def is_cog_eligible(self, filepath: AsyncPath) -> bool:
|
async def is_cog_eligible(self, filepath: Path) -> bool:
|
||||||
"""
|
"""
|
||||||
Checks if the specified file is a cog.
|
Checks if the specified file is a cog.
|
||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
-----------
|
-----------
|
||||||
filepath : AsyncPath
|
filepath : Path
|
||||||
The path to the file to check.
|
The path to the file to check.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
|
@ -35,16 +35,16 @@ class CogLoader(commands.Cog):
|
||||||
filepath.suffix == ".py"
|
filepath.suffix == ".py"
|
||||||
and cog_name not in self.cog_ignore_list
|
and cog_name not in self.cog_ignore_list
|
||||||
and not filepath.name.startswith("_")
|
and not filepath.name.startswith("_")
|
||||||
and await filepath.is_file()
|
and await aiofiles.os.path.isfile(filepath)
|
||||||
)
|
)
|
||||||
|
|
||||||
async def load_cogs(self, apath: AsyncPath) -> None:
|
async def load_cogs(self, path: Path) -> None:
|
||||||
"""
|
"""
|
||||||
Recursively loads eligible cogs from the specified directory.
|
Recursively loads eligible cogs from the specified directory.
|
||||||
|
|
||||||
Parameters:
|
Parameters:
|
||||||
-----------
|
-----------
|
||||||
apath : AsyncPath
|
path : Path
|
||||||
The path to the directory containing cogs.
|
The path to the directory containing cogs.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
|
@ -58,17 +58,16 @@ class CogLoader(commands.Cog):
|
||||||
"""
|
"""
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if await apath.is_dir():
|
if await aiofiles.os.path.isdir(path):
|
||||||
async for item in apath.iterdir():
|
for item in path.iterdir():
|
||||||
try:
|
try:
|
||||||
await self.load_cogs(apath=item)
|
await self.load_cogs(path=item)
|
||||||
|
|
||||||
except Exception as error:
|
except Exception as error:
|
||||||
logger.error(f"Error loading cog from {item}: {error}")
|
logger.error(f"Error loading cog from {item}: {error}")
|
||||||
|
|
||||||
elif await self.is_cog_eligible(filepath=apath):
|
elif await self.is_cog_eligible(filepath=path):
|
||||||
# TODO: Fix this type ignore
|
relative_path: Path = path.relative_to(Path(__file__).parent)
|
||||||
relative_path: AsyncPath = apath.relative_to(AsyncPath(__file__).parent) # type: ignore
|
|
||||||
module: str = str(relative_path).replace("/", ".").replace("\\", ".")[:-3]
|
module: str = str(relative_path).replace("/", ".").replace("\\", ".")[:-3]
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
@ -81,7 +80,7 @@ class CogLoader(commands.Cog):
|
||||||
)
|
)
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"An error occurred while processing {apath}: {e}")
|
logger.error(f"An error occurred while processing {path}: {e}")
|
||||||
|
|
||||||
async def load_cogs_from_folder(self, folder_name: str) -> None:
|
async def load_cogs_from_folder(self, folder_name: str) -> None:
|
||||||
"""
|
"""
|
||||||
|
@ -97,9 +96,9 @@ class CogLoader(commands.Cog):
|
||||||
None
|
None
|
||||||
"""
|
"""
|
||||||
|
|
||||||
cog_path: AsyncPath = AsyncPath(__file__).parent / folder_name
|
cog_path: Path = Path(__file__).parent / folder_name
|
||||||
|
|
||||||
await self.load_cogs(apath=cog_path)
|
await self.load_cogs(path=cog_path)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def setup(cls, bot: commands.Bot) -> None:
|
async def setup(cls, bot: commands.Bot) -> None:
|
||||||
|
|
Loading…
Reference in a new issue