sh-edraft.de/sh_cpl
1
1
Fork 0
Code Issues 4 Pull Requests 1 Actions Packages Projects 1 Releases 12 Wiki Activity

Improved cpl-discords service loading for commands and events

Browse Source
This commit is contained in:
Sven Heidemann 2023-10-12 20:41:34 +02:00
parent f357e97ce5
commit 1dd48899d7
6 changed files with 27 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/cpl_core/dependency_injection/service_provider.py
View File

@ -157,12 +157,12 @@ class ServiceProvider(ServiceProviderABC):
return implementation
def get_services(self, service_type: typing.Type[T], *args, **kwargs) -> list[Optional[T]]:
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[T]]:
implementations = []
if typing.get_origin(service_type) != list:
raise Exception(f"Invalid type {service_type}! Expected list of type")
if typing.get_origin(service_type) == list:
raise Exception(f"Invalid type {service_type}! Expected single type not list of type")
implementations.extend(self._get_services(typing.get_args(service_type)[0]))
implementations.extend(self._get_services(service_type))
return implementations

2
src/cpl_core/dependency_injection/service_provider_abc.py
View File

@ -76,7 +76,7 @@ class ServiceProviderABC(ABC):
pass
@abstractmethod
def get_services(self, service_type: Type[T], *args, **kwargs) -> list[Optional[T]]:
def get_services(self, service_type: T, *args, **kwargs) -> list[Optional[T]]:
r"""Returns instance of given type
Parameter

4
src/cpl_discord/cpl-discord.json
View File

@ -3,8 +3,8 @@
"Name": "cpl-discord",
"Version": {
"Major": "2023",
"Minor": "4",
"Micro": "0.post3"
"Minor": "10",
"Micro": "0"
},
"Author": "Sven Heidemann",
"AuthorEmail": "sven.heidemann@sh-edraft.de",

29
src/cpl_discord/service/discord_collection.py
View File

@ -1,11 +1,11 @@
from typing import Type, Optional
from typing import Type
from cpl_core.console import Console, ForegroundColorEnum
from cpl_core.dependency_injection import ServiceCollectionABC
from cpl_discord.command.discord_command_abc import DiscordCommandABC
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
from cpl_discord.service.command_error_handler_service import CommandErrorHandlerService
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from cpl_query.extension.list import List
class DiscordCollection(DiscordCollectionABC):
@ -13,26 +13,21 @@ class DiscordCollection(DiscordCollectionABC):
DiscordCollectionABC.__init__(self)
self._services = service_collection
self._events: dict[str, List] = {}
self._commands = List(type(DiscordCommandABC))
self.add_event(DiscordEventTypesEnum.on_command_error.value, CommandErrorHandlerService)
def add_command(self, _t: Type[DiscordCommandABC]):
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(
f"{type(self).__name__}.add_command is deprecated. Instead, use ServiceCollection.add_transient directly!"
)
Console.color_reset()
self._services.add_transient(DiscordCommandABC, _t)
self._commands.append(_t)
def get_commands(self) -> List[DiscordCommandABC]:
return self._commands
def add_event(self, _t_event: Type, _t: Type):
Console.set_foreground_color(ForegroundColorEnum.yellow)
Console.write_line(
f"{type(self).__name__}.add_event is deprecated. Instead, use ServiceCollection.add_transient directly!"
)
Console.color_reset()
self._services.add_transient(_t_event, _t)
if _t_event not in self._events:
self._events[_t_event] = List(type(_t_event))
self._events[_t_event].append(_t)
def get_events_by_base(self, _t_event: Type) -> Optional[List]:
if _t_event not in self._events:
return None
return self._events[_t_event]

8
src/cpl_discord/service/discord_collection_abc.py
View File

@ -13,14 +13,6 @@ class DiscordCollectionABC(ABC):
def add_command(self, _t: Type[DiscordCommandABC]):
pass
@abstractmethod
def get_commands(self) -> List[DiscordCommandABC]:
pass
@abstractmethod
def add_event(self, _t_event: Type, _t: Type):
pass
@abstractmethod
def get_events_by_base(self, _t_event: Type):
pass

23
src/cpl_discord/service/discord_service.py
View File

@ -4,11 +4,12 @@ from typing import Optional, Sequence, Union, Type
import discord
from discord import RawReactionActionEvent
from discord.ext import commands
from discord.ext.commands import Context, CommandError, Cog, Command
from discord.ext.commands import Context, CommandError, Cog
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.logging import LoggerABC
from cpl_core.utils import String
from cpl_discord.command import DiscordCommandABC
from cpl_discord.command.discord_commands_meta import DiscordCogMeta
from cpl_discord.events.on_bulk_message_delete_abc import OnBulkMessageDeleteABC
from cpl_discord.events.on_command_abc import OnCommandABC
@ -66,25 +67,17 @@ from cpl_discord.events.on_typing_abc import OnTypingABC
from cpl_discord.events.on_user_update_abc import OnUserUpdateABC
from cpl_discord.events.on_voice_state_update_abc import OnVoiceStateUpdateABC
from cpl_discord.events.on_webhooks_update_abc import OnWebhooksUpdateABC
from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
from cpl_discord.service.discord_service_abc import DiscordServiceABC
class DiscordService(DiscordServiceABC, commands.Cog, metaclass=DiscordCogMeta):
def __init__(self, logger: LoggerABC, dc_collection: DiscordCollectionABC, services: ServiceProviderABC):
def __init__(self, logger: LoggerABC, services: ServiceProviderABC):
DiscordServiceABC.__init__(self)
self._logger = logger
self._collection = dc_collection
self._services = services
async def _handle_event(self, event: Type, *args, **kwargs):
event_collection = self._collection.get_events_by_base(event)
if event_collection is None:
return
for event_type in event_collection:
event_instance = self._services.get_service(event_type)
for event_instance in self._services.get_services(event):
func_name = event.__name__
if func_name.endswith("ABC"):
func_name = func_name.replace("ABC", "")
@ -104,11 +97,11 @@ class DiscordService(DiscordServiceABC, commands.Cog, metaclass=DiscordCogMeta):
self._logger.error(__name__, f"{type(self).__name__} initialization failed", e)
try:
for command_type in self._collection.get_commands():
self._logger.trace(__name__, f"Register command {command_type.__name__}")
command: Cog = self._services.get_service(command_type)
for command in self._services.get_services(DiscordCommandABC):
self._logger.trace(__name__, f"Register command {type(command).__name__}")
command: Cog = command
if command is None:
self._logger.warn(__name__, f"Instance of {command_type.__name__} not found")
self._logger.warn(__name__, f"Instance of {type(command).__name__} not found")
continue
await bot.add_cog(command)
except Exception as e: