Formatted code

This commit is contained in:
Sven Heidemann 2023-04-04 13:30:00 +02:00
parent 01bc746092
commit c4a71de83f
12 changed files with 200 additions and 178 deletions

View File

@ -28,7 +28,12 @@ class ServiceProvider(ServiceProviderABC):
Database representation Database representation
""" """
def __init__(self, service_descriptors: list[ServiceDescriptor], config: ConfigurationABC, db_context: Optional[DatabaseContextABC]): def __init__(
self,
service_descriptors: list[ServiceDescriptor],
config: ConfigurationABC,
db_context: Optional[DatabaseContextABC],
):
ServiceProviderABC.__init__(self) ServiceProviderABC.__init__(self)
self._service_descriptors: list[ServiceDescriptor] = service_descriptors self._service_descriptors: list[ServiceDescriptor] = service_descriptors
@ -45,7 +50,9 @@ class ServiceProvider(ServiceProviderABC):
def _get_service(self, parameter: Parameter) -> Optional[object]: def _get_service(self, parameter: Parameter) -> Optional[object]:
for descriptor in self._service_descriptors: for descriptor in self._service_descriptors:
if descriptor.service_type == parameter.annotation or issubclass(descriptor.service_type, parameter.annotation): if descriptor.service_type == parameter.annotation or issubclass(
descriptor.service_type, parameter.annotation
):
if descriptor.implementation is not None: if descriptor.implementation is not None:
return descriptor.implementation return descriptor.implementation
@ -77,8 +84,7 @@ class ServiceProvider(ServiceProviderABC):
params = [] params = []
for param in sig.parameters.items(): for param in sig.parameters.items():
parameter = param[1] parameter = param[1]
if parameter.name != 'self' and parameter.annotation != Parameter.empty: if parameter.name != "self" and parameter.annotation != Parameter.empty:
if typing.get_origin(parameter.annotation) == list: if typing.get_origin(parameter.annotation) == list:
params.append(self._get_services(typing.get_args(parameter.annotation)[0])) params.append(self._get_services(typing.get_args(parameter.annotation)[0]))
@ -121,7 +127,9 @@ class ServiceProvider(ServiceProviderABC):
self._scope = scope self._scope = scope
def create_scope(self) -> ScopeABC: def create_scope(self) -> ScopeABC:
sb = ScopeBuilder(ServiceProvider(copy.deepcopy(self._service_descriptors), self._configuration, self._database_context)) sb = ScopeBuilder(
ServiceProvider(copy.deepcopy(self._service_descriptors), self._configuration, self._database_context)
)
return sb.build() return sb.build()
def get_service(self, service_type: T, *args, **kwargs) -> Optional[T]: def get_service(self, service_type: T, *args, **kwargs) -> Optional[T]:
@ -134,7 +142,11 @@ class ServiceProvider(ServiceProviderABC):
return result.implementation return result.implementation
implementation = self.build_service(service_type, *args, **kwargs) implementation = self.build_service(service_type, *args, **kwargs)
if result.lifetime == ServiceLifetimeEnum.singleton or result.lifetime == ServiceLifetimeEnum.scoped and self._scope is not None: if (
result.lifetime == ServiceLifetimeEnum.singleton
or result.lifetime == ServiceLifetimeEnum.scoped
and self._scope is not None
):
result.implementation = implementation result.implementation = implementation
return implementation return implementation
@ -143,7 +155,7 @@ class ServiceProvider(ServiceProviderABC):
implementations = [] implementations = []
if typing.get_origin(service_type) != list: if typing.get_origin(service_type) != list:
raise Exception(f'Invalid type {service_type}! Expected list of type') raise Exception(f"Invalid type {service_type}! Expected list of type")
implementations.extend(self._get_services(typing.get_args(service_type)[0])) implementations.extend(self._get_services(typing.get_args(service_type)[0]))

View File

@ -10,14 +10,14 @@ from cpl_core.type import T
class ServiceProviderABC(ABC): class ServiceProviderABC(ABC):
r"""ABC for the class :class:`cpl_core.dependency_injection.service_provider.ServiceProvider`""" r"""ABC for the class :class:`cpl_core.dependency_injection.service_provider.ServiceProvider`"""
_provider: Optional['ServiceProviderABC'] = None _provider: Optional["ServiceProviderABC"] = None
@abstractmethod @abstractmethod
def __init__(self): def __init__(self):
pass pass
@classmethod @classmethod
def set_global_provider(cls, provider: 'ServiceProviderABC'): def set_global_provider(cls, provider: "ServiceProviderABC"):
cls._provider = provider cls._provider = provider
@abstractmethod @abstractmethod
@ -108,7 +108,7 @@ class ServiceProviderABC(ABC):
@functools.wraps(f) @functools.wraps(f)
def inner(*args, **kwargs): def inner(*args, **kwargs):
if cls._provider is None: if cls._provider is None:
raise Exception(f'{cls.__name__} not build!') raise Exception(f"{cls.__name__} not build!")
injection = [x for x in cls._provider.build_by_signature(signature(f)) if x is not None] injection = [x for x in cls._provider.build_by_signature(signature(f)) if x is not None]
return f(*args, *injection, **kwargs) return f(*args, *injection, **kwargs)

View File

@ -7,7 +7,6 @@ from cpl_core.utils import String
class Event(GenerateSchematicABC): class Event(GenerateSchematicABC):
def __init__(self, name: str, schematic: str, path: str): def __init__(self, name: str, schematic: str, path: str):
GenerateSchematicABC.__init__(self, name, schematic, path) GenerateSchematicABC.__init__(self, name, schematic, path)
@ -15,8 +14,9 @@ class Event(GenerateSchematicABC):
event_class = None event_class = None
from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum from cpl_discord.discord_event_types_enum import DiscordEventTypesEnum
for event_type in DiscordEventTypesEnum: for event_type in DiscordEventTypesEnum:
event_name = event_type.value.__name__.replace("ABC", '') event_name = event_type.value.__name__.replace("ABC", "")
if name.endswith(event_name): if name.endswith(event_name):
name = name.replace(event_name, "") name = name.replace(event_name, "")
@ -25,29 +25,30 @@ class Event(GenerateSchematicABC):
break break
if event is None: if event is None:
Console.error(f'No valid event found in name {name}') Console.error(f"No valid event found in name {name}")
Console.write_line('Available events:') Console.write_line("Available events:")
for event_type in DiscordEventTypesEnum: for event_type in DiscordEventTypesEnum:
Console.write_line(f'\t{event_type.value.__name__.replace("ABC", "")}') Console.write_line(f'\t{event_type.value.__name__.replace("ABC", "")}')
sys.exit() sys.exit()
self._event_class_name = f'{event}ABC' self._event_class_name = f"{event}ABC"
event_snake_case = String.convert_to_snake_case(self._event_class_name.replace("ABC", "")) event_snake_case = String.convert_to_snake_case(self._event_class_name.replace("ABC", ""))
if event_snake_case.lower() not in dir(event_class): if event_snake_case.lower() not in dir(event_class):
Console.error(f'Error in event {event}: Function {event_snake_case} not found!') Console.error(f"Error in event {event}: Function {event_snake_case} not found!")
sys.exit() sys.exit()
self._name = f'{event_snake_case}_{schematic}.py' self._name = f"{event_snake_case}_{schematic}.py"
self._class_name = f'{self._event_class_name.replace("ABC", "")}{String.first_to_upper(schematic)}' self._class_name = f'{self._event_class_name.replace("ABC", "")}{String.first_to_upper(schematic)}'
from inspect import signature from inspect import signature
self._func_name = event_snake_case self._func_name = event_snake_case
self._signature = str(signature(getattr(event_class, event_snake_case)))[1:][:-1] self._signature = str(signature(getattr(event_class, event_snake_case)))[1:][:-1]
if name != '': if name != "":
self._name = f'{String.convert_to_snake_case(name)}_{self._name}' self._name = f"{String.convert_to_snake_case(name)}_{self._name}"
self._class_name = f'{String.first_to_upper(name)}{self._class_name}' self._class_name = f"{String.first_to_upper(name)}{self._class_name}"
def get_code(self) -> str: def get_code(self) -> str:
code = """\ code = """\
@ -77,13 +78,9 @@ class Event(GenerateSchematicABC):
Name=self._class_name, Name=self._class_name,
EventClass=self._event_class_name, EventClass=self._event_class_name,
Func=self._func_name, Func=self._func_name,
Signature=self._signature Signature=self._signature,
) )
@classmethod @classmethod
def register(cls): def register(cls):
GenerateSchematicABC.register( GenerateSchematicABC.register(cls, "event", [])
cls,
'event',
[]
)

View File

@ -3,9 +3,10 @@ import discord
class OnScheduledEventCreateABC(ABC): class OnScheduledEventCreateABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod @abstractmethod
def __init__(self): pass async def on_scheduled_event_create(self, event: discord.ScheduledEvent):
pass
@abstractmethod
async def on_scheduled_event_create(self, event: discord.ScheduledEvent): pass

View File

@ -3,9 +3,10 @@ import discord
class OnScheduledEventDeleteABC(ABC): class OnScheduledEventDeleteABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod @abstractmethod
def __init__(self): pass async def on_scheduled_event_delete(self, event: discord.ScheduledEvent):
pass
@abstractmethod
async def on_scheduled_event_delete(self, event: discord.ScheduledEvent): pass

View File

@ -3,9 +3,10 @@ import discord
class OnScheduledEventUpdateABC(ABC): class OnScheduledEventUpdateABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod @abstractmethod
def __init__(self): pass async def on_scheduled_event_update(self, before: discord.ScheduledEvent, after: discord.ScheduledEvent):
pass
@abstractmethod
async def on_scheduled_event_update(self, before: discord.ScheduledEvent, after: discord.ScheduledEvent): pass

View File

@ -3,9 +3,10 @@ import discord
class OnScheduledEventUserAddABC(ABC): class OnScheduledEventUserAddABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod @abstractmethod
def __init__(self): pass async def on_scheduled_event_user_add(self, event: discord.ScheduledEvent, user: discord.User):
pass
@abstractmethod
async def on_scheduled_event_user_add(self, event: discord.ScheduledEvent, user: discord.User): pass

View File

@ -3,9 +3,10 @@ import discord
class OnScheduledEventUserRemoveABC(ABC): class OnScheduledEventUserRemoveABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod @abstractmethod
def __init__(self): pass async def on_scheduled_event_user_remove(self, event: discord.ScheduledEvent, user: discord.User):
pass
@abstractmethod
async def on_scheduled_event_user_remove(self, event: discord.ScheduledEvent, user: discord.User): pass

View File

@ -13,7 +13,6 @@ from cpl_query.extension.list import List
class DiscordBotService(DiscordBotServiceABC): class DiscordBotService(DiscordBotServiceABC):
def __init__( def __init__(
self, self,
config: ConfigurationABC, config: ConfigurationABC,
@ -23,7 +22,7 @@ class DiscordBotService(DiscordBotServiceABC):
logging_st: LoggingSettings, logging_st: LoggingSettings,
discord_service: DiscordServiceABC, discord_service: DiscordServiceABC,
*args, *args,
**kwargs **kwargs,
): ):
# services # services
self._config = config self._config = config
@ -39,55 +38,59 @@ class DiscordBotService(DiscordBotServiceABC):
DiscordBotServiceABC.__init__( DiscordBotServiceABC.__init__(
self, self,
*args, *args,
command_prefix=self._discord_settings.prefix, help_command=None, intents=discord.Intents().all(), command_prefix=self._discord_settings.prefix,
**kwargs help_command=None,
intents=discord.Intents().all(),
**kwargs,
) )
self._base = super(DiscordBotServiceABC, self) self._base = super(DiscordBotServiceABC, self)
@staticmethod @staticmethod
def _is_string_invalid(x): def _is_string_invalid(x):
return x is None or x == '' return x is None or x == ""
def _get_settings(self, settings_from_config: DiscordBotSettings) -> DiscordBotSettings: def _get_settings(self, settings_from_config: DiscordBotSettings) -> DiscordBotSettings:
new_settings = DiscordBotSettings() new_settings = DiscordBotSettings()
token = None if settings_from_config is None else settings_from_config.token token = None if settings_from_config is None else settings_from_config.token
prefix = None if settings_from_config is None else settings_from_config.prefix prefix = None if settings_from_config is None else settings_from_config.prefix
env_token = self._config.get_configuration('TOKEN') env_token = self._config.get_configuration("TOKEN")
env_prefix = self._config.get_configuration('PREFIX') env_prefix = self._config.get_configuration("PREFIX")
new_settings.from_dict({ new_settings.from_dict(
'Token': env_token if token is None or token == '' else token, {
'Prefix': "Token": env_token if token is None or token == "" else token,
('! ' if self._is_string_invalid(env_prefix) else env_prefix) "Prefix": ("! " if self._is_string_invalid(env_prefix) else env_prefix)
if self._is_string_invalid(prefix) else prefix if self._is_string_invalid(prefix)
}) else prefix,
if new_settings.token is None or new_settings.token == '': }
raise Exception('You have to configure discord token by appsettings or environment variables') )
if new_settings.token is None or new_settings.token == "":
raise Exception("You have to configure discord token by appsettings or environment variables")
return new_settings return new_settings
async def start_async(self): async def start_async(self):
self._logger.trace(__name__, 'Try to connect to discord') self._logger.trace(__name__, "Try to connect to discord")
await self.start(self._discord_settings.token) await self.start(self._discord_settings.token)
# continue at on_ready # continue at on_ready
async def stop_async(self): async def stop_async(self):
self._logger.trace(__name__, 'Try to disconnect from discord') self._logger.trace(__name__, "Try to disconnect from discord")
try: try:
await self.close() await self.close()
except Exception as e: except Exception as e:
self._logger.error(__name__, 'Stop failed', e) self._logger.error(__name__, "Stop failed", e)
async def on_ready(self): async def on_ready(self):
self._logger.info(__name__, 'Connected to discord') self._logger.info(__name__, "Connected to discord")
self._logger.header(f'{self.user.name}:') self._logger.header(f"{self.user.name}:")
if self._logging_st.console.value >= LoggingLevelEnum.INFO.value: if self._logging_st.console.value >= LoggingLevelEnum.INFO.value:
Console.banner(self._env.application_name if self._env.application_name != '' else 'A bot') Console.banner(self._env.application_name if self._env.application_name != "" else "A bot")
await self._discord_service.init(self) await self._discord_service.init(self)
await self.wait_until_ready() await self.wait_until_ready()
await self.tree.sync() await self.tree.sync()
self._logger.debug(__name__, f'Finished syncing commands') self._logger.debug(__name__, f"Finished syncing commands")
await self._discord_service.on_ready() await self._discord_service.on_ready()

View File

@ -7,19 +7,22 @@ from cpl_query.extension.list import List
class DiscordBotServiceABC(commands.Bot): class DiscordBotServiceABC(commands.Bot):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
commands.Bot.__init__(self, *args, **kwargs) commands.Bot.__init__(self, *args, **kwargs)
@abstractmethod @abstractmethod
async def start_async(self): pass async def start_async(self):
pass
@abstractmethod @abstractmethod
async def stop_async(self): pass async def stop_async(self):
pass
@abstractmethod @abstractmethod
async def on_ready(self): pass async def on_ready(self):
pass
@property @property
@abstractmethod @abstractmethod
def guilds(self) -> List[Guild]: pass def guilds(self) -> List[Guild]:
pass

View File

@ -71,13 +71,7 @@ from cpl_discord.service.discord_service_abc import DiscordServiceABC
class DiscordService(DiscordServiceABC, commands.Cog, metaclass=DiscordCogMeta): class DiscordService(DiscordServiceABC, commands.Cog, metaclass=DiscordCogMeta):
def __init__(self, logger: LoggerABC, dc_collection: DiscordCollectionABC, services: ServiceProviderABC):
def __init__(
self,
logger: LoggerABC,
dc_collection: DiscordCollectionABC,
services: ServiceProviderABC
):
DiscordServiceABC.__init__(self) DiscordServiceABC.__init__(self)
self._logger = logger self._logger = logger
self._collection = dc_collection self._collection = dc_collection
@ -92,8 +86,8 @@ class DiscordService(DiscordServiceABC, commands.Cog, metaclass=DiscordCogMeta):
event_instance = self._services.get_service(event_type) event_instance = self._services.get_service(event_type)
func_name = event.__name__ func_name = event.__name__
if func_name.endswith('ABC'): if func_name.endswith("ABC"):
func_name = func_name.replace('ABC', '') func_name = func_name.replace("ABC", "")
func_name = String.convert_to_snake_case(func_name) func_name = String.convert_to_snake_case(func_name)
@ -101,305 +95,311 @@ class DiscordService(DiscordServiceABC, commands.Cog, metaclass=DiscordCogMeta):
func = getattr(event_instance, func_name) func = getattr(event_instance, func_name)
await func(*args, **kwargs) await func(*args, **kwargs)
except Exception as e: except Exception as e:
self._logger.error(__name__, f'Cannot execute {func_name} of {type(event_instance).__name__}', e) self._logger.error(__name__, f"Cannot execute {func_name} of {type(event_instance).__name__}", e)
async def init(self, bot: commands.Bot): async def init(self, bot: commands.Bot):
try: try:
await bot.add_cog(self) await bot.add_cog(self)
except Exception as e: except Exception as e:
self._logger.error(__name__, f'{type(self).__name__} initialization failed', e) self._logger.error(__name__, f"{type(self).__name__} initialization failed", e)
try: try:
for command_type in self._collection.get_commands(): for command_type in self._collection.get_commands():
self._logger.trace(__name__, f'Register command {command_type.__name__}') self._logger.trace(__name__, f"Register command {command_type.__name__}")
command: Cog = self._services.get_service(command_type) command: Cog = self._services.get_service(command_type)
if command is None: if command is None:
self._logger.warn(__name__, f'Instance of {command_type.__name__} not found') self._logger.warn(__name__, f"Instance of {command_type.__name__} not found")
continue continue
await bot.add_cog(command) await bot.add_cog(command)
except Exception as e: except Exception as e:
self._logger.error(__name__, f'Registration of commands failed', e) self._logger.error(__name__, f"Registration of commands failed", e)
@commands.Cog.listener() @commands.Cog.listener()
async def on_connect(self): async def on_connect(self):
self._logger.trace(__name__, f'Received on_connect') self._logger.trace(__name__, f"Received on_connect")
await self._handle_event(OnConnectABC) await self._handle_event(OnConnectABC)
@commands.Cog.listener() @commands.Cog.listener()
async def on_command(self, ctx: Context): async def on_command(self, ctx: Context):
self._logger.trace(__name__, f'Received on_command') self._logger.trace(__name__, f"Received on_command")
await self._handle_event(OnCommandABC, ctx) await self._handle_event(OnCommandABC, ctx)
@commands.Cog.listener() @commands.Cog.listener()
async def on_command_error(self, ctx: Context, error: CommandError): async def on_command_error(self, ctx: Context, error: CommandError):
self._logger.trace(__name__, f'Received on_command_error') self._logger.trace(__name__, f"Received on_command_error")
await self._handle_event(OnCommandErrorABC, ctx, error) await self._handle_event(OnCommandErrorABC, ctx, error)
@commands.Cog.listener() @commands.Cog.listener()
async def on_command_completion(self, ctx: Context): async def on_command_completion(self, ctx: Context):
self._logger.trace(__name__, f'Received on_command_completion') self._logger.trace(__name__, f"Received on_command_completion")
await self._handle_event(OnCommandCompletionABC, ctx) await self._handle_event(OnCommandCompletionABC, ctx)
@commands.Cog.listener() @commands.Cog.listener()
async def on_disconnect(self): async def on_disconnect(self):
self._logger.trace(__name__, f'Received on_disconnect') self._logger.trace(__name__, f"Received on_disconnect")
await self._handle_event(OnDisconnectABC) await self._handle_event(OnDisconnectABC)
@commands.Cog.listener() @commands.Cog.listener()
async def on_error(self, event: str, *args, **kwargs): async def on_error(self, event: str, *args, **kwargs):
self._logger.trace(__name__, f'Received on_error') self._logger.trace(__name__, f"Received on_error")
await self._handle_event(OnErrorABC, event, *args, **kwargs) await self._handle_event(OnErrorABC, event, *args, **kwargs)
async def on_ready(self): async def on_ready(self):
self._logger.trace(__name__, f'Received on_ready') self._logger.trace(__name__, f"Received on_ready")
await self._handle_event(OnReadyABC) await self._handle_event(OnReadyABC)
@commands.Cog.listener() @commands.Cog.listener()
async def on_resume(self): async def on_resume(self):
self._logger.trace(__name__, f'Received on_resume') self._logger.trace(__name__, f"Received on_resume")
await self._handle_event(OnResumeABC) await self._handle_event(OnResumeABC)
@commands.Cog.listener() @commands.Cog.listener()
async def on_error(self, event: str, *args, **kwargs): async def on_error(self, event: str, *args, **kwargs):
self._logger.trace(__name__, f'Received on_error:\n\t{event}\n\t{args}\n\t{kwargs}') self._logger.trace(__name__, f"Received on_error:\n\t{event}\n\t{args}\n\t{kwargs}")
await self._handle_event(OnReadyABC, event, *args, **kwargs) await self._handle_event(OnReadyABC, event, *args, **kwargs)
@commands.Cog.listener() @commands.Cog.listener()
async def on_typing(self, channel: discord.abc.Messageable, user: Union[discord.User, discord.Member], when: datetime): async def on_typing(
self._logger.trace(__name__, f'Received on_typing:\n\t{channel}\n\t{user}\n\t{when}') self, channel: discord.abc.Messageable, user: Union[discord.User, discord.Member], when: datetime
):
self._logger.trace(__name__, f"Received on_typing:\n\t{channel}\n\t{user}\n\t{when}")
await self._handle_event(OnTypingABC, channel, user, when) await self._handle_event(OnTypingABC, channel, user, when)
@commands.Cog.listener() @commands.Cog.listener()
async def on_message(self, message: discord.Message): async def on_message(self, message: discord.Message):
self._logger.trace(__name__, f'Received on_message:\n\t{message}') self._logger.trace(__name__, f"Received on_message:\n\t{message}")
await self._handle_event(OnMessageABC, message) await self._handle_event(OnMessageABC, message)
@commands.Cog.listener() @commands.Cog.listener()
async def on_message_delete(self, message: discord.Message): async def on_message_delete(self, message: discord.Message):
self._logger.trace(__name__, f'Received on_message_delete:\n\t{message}') self._logger.trace(__name__, f"Received on_message_delete:\n\t{message}")
await self._handle_event(OnMessageDeleteABC, message) await self._handle_event(OnMessageDeleteABC, message)
@commands.Cog.listener() @commands.Cog.listener()
async def on_bulk_message_delete(self, messages: list[discord.Message]): async def on_bulk_message_delete(self, messages: list[discord.Message]):
self._logger.trace(__name__, f'Received on_bulk_message_delete:\n\t{len(messages)}') self._logger.trace(__name__, f"Received on_bulk_message_delete:\n\t{len(messages)}")
await self._handle_event(OnBulkMessageDeleteABC, messages) await self._handle_event(OnBulkMessageDeleteABC, messages)
@commands.Cog.listener() @commands.Cog.listener()
async def on_message_edit(self, before: discord.Message, after: discord.Message): async def on_message_edit(self, before: discord.Message, after: discord.Message):
self._logger.trace(__name__, f'Received on_message_edit:\n\t{before}\n\t{after}') self._logger.trace(__name__, f"Received on_message_edit:\n\t{before}\n\t{after}")
await self._handle_event(OnMessageEditABC, before, after) await self._handle_event(OnMessageEditABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_raw_reaction_add(self, payload: RawReactionActionEvent): async def on_raw_reaction_add(self, payload: RawReactionActionEvent):
self._logger.trace(__name__, f'Received on_raw_reaction_add') self._logger.trace(__name__, f"Received on_raw_reaction_add")
await self._handle_event(OnRawReactionAddABC, payload) await self._handle_event(OnRawReactionAddABC, payload)
@commands.Cog.listener() @commands.Cog.listener()
async def on_raw_reaction_remove(self, payload: RawReactionActionEvent): async def on_raw_reaction_remove(self, payload: RawReactionActionEvent):
self._logger.trace(__name__, f'Received on_raw_reaction_remove') self._logger.trace(__name__, f"Received on_raw_reaction_remove")
await self._handle_event(OnRawReactionRemoveABC, payload) await self._handle_event(OnRawReactionRemoveABC, payload)
@commands.Cog.listener() @commands.Cog.listener()
async def on_raw_reaction_clear(self, payload: RawReactionActionEvent): async def on_raw_reaction_clear(self, payload: RawReactionActionEvent):
self._logger.trace(__name__, f'Received on_raw_reaction_clear') self._logger.trace(__name__, f"Received on_raw_reaction_clear")
await self._handle_event(OnRawReactionClearABC, payload) await self._handle_event(OnRawReactionClearABC, payload)
@commands.Cog.listener() @commands.Cog.listener()
async def on_raw_reaction_clear_emoji(self, payload: RawReactionActionEvent): async def on_raw_reaction_clear_emoji(self, payload: RawReactionActionEvent):
self._logger.trace(__name__, f'Received on_raw_reaction_clear_emoji') self._logger.trace(__name__, f"Received on_raw_reaction_clear_emoji")
await self._handle_event(OnRawReactionClearEmojiABC, payload) await self._handle_event(OnRawReactionClearEmojiABC, payload)
@commands.Cog.listener() @commands.Cog.listener()
async def on_reaction_add(self, reaction: discord.Reaction, user: discord.User): async def on_reaction_add(self, reaction: discord.Reaction, user: discord.User):
self._logger.trace(__name__, f'Received on_reaction_add:\n\t{reaction}\n\t{user}') self._logger.trace(__name__, f"Received on_reaction_add:\n\t{reaction}\n\t{user}")
await self._handle_event(OnReactionAddABC, reaction, user) await self._handle_event(OnReactionAddABC, reaction, user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_reaction_remove(self, reaction: discord.Reaction, user: discord.User): async def on_reaction_remove(self, reaction: discord.Reaction, user: discord.User):
self._logger.trace(__name__, f'Received on_reaction_remove:\n\t{reaction}\n\t{user}') self._logger.trace(__name__, f"Received on_reaction_remove:\n\t{reaction}\n\t{user}")
await self._handle_event(OnReactionRemoveABC, reaction, user) await self._handle_event(OnReactionRemoveABC, reaction, user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_reaction_clear(self, message: discord.Message, reactions: list[discord.Reaction]): async def on_reaction_clear(self, message: discord.Message, reactions: list[discord.Reaction]):
self._logger.trace(__name__, f'Received on_reaction_reon_reaction_clearmove:\n\t{message}\n\t{len(reactions)}') self._logger.trace(__name__, f"Received on_reaction_reon_reaction_clearmove:\n\t{message}\n\t{len(reactions)}")
await self._handle_event(OnReactionClearABC, message, reactions) await self._handle_event(OnReactionClearABC, message, reactions)
@commands.Cog.listener() @commands.Cog.listener()
async def on_reaction_clear_emoji(self, reaction: discord.Reaction): async def on_reaction_clear_emoji(self, reaction: discord.Reaction):
self._logger.trace(__name__, f'Received on_reaction_clear_emoji:\n\t{reaction}') self._logger.trace(__name__, f"Received on_reaction_clear_emoji:\n\t{reaction}")
await self._handle_event(OnReactionClearEmojiABC, reaction) await self._handle_event(OnReactionClearEmojiABC, reaction)
@commands.Cog.listener() @commands.Cog.listener()
async def on_private_channel_delete(self, channel: discord.abc.PrivateChannel): async def on_private_channel_delete(self, channel: discord.abc.PrivateChannel):
self._logger.trace(__name__, f'Received on_private_channel_delete:\n\t{channel}') self._logger.trace(__name__, f"Received on_private_channel_delete:\n\t{channel}")
await self._handle_event(OnPrivateChannelDeleteABC, channel) await self._handle_event(OnPrivateChannelDeleteABC, channel)
@commands.Cog.listener() @commands.Cog.listener()
async def on_private_channel_create(self, channel: discord.abc.PrivateChannel): async def on_private_channel_create(self, channel: discord.abc.PrivateChannel):
self._logger.trace(__name__, f'Received on_private_channel_create:\n\t{channel}') self._logger.trace(__name__, f"Received on_private_channel_create:\n\t{channel}")
await self._handle_event(OnPrivateChannelCreateABC, channel) await self._handle_event(OnPrivateChannelCreateABC, channel)
@commands.Cog.listener() @commands.Cog.listener()
async def on_private_channel_update(self, before: discord.GroupChannel, after: discord.GroupChannel): async def on_private_channel_update(self, before: discord.GroupChannel, after: discord.GroupChannel):
self._logger.trace(__name__, f'Received on_private_channel_update:\n\t{before}\n\t{after}') self._logger.trace(__name__, f"Received on_private_channel_update:\n\t{before}\n\t{after}")
await self._handle_event(OnPrivateChannelUpdateABC, before, after) await self._handle_event(OnPrivateChannelUpdateABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_private_channel_pins_update(self, channel: discord.abc.PrivateChannel, list_pin: Optional[datetime]): async def on_private_channel_pins_update(self, channel: discord.abc.PrivateChannel, list_pin: Optional[datetime]):
self._logger.trace(__name__, f'Received on_private_channel_pins_update:\n\t{channel}\n\t{list_pin}') self._logger.trace(__name__, f"Received on_private_channel_pins_update:\n\t{channel}\n\t{list_pin}")
await self._handle_event(OnPrivateChannelPinsUpdateABC, channel, list_pin) await self._handle_event(OnPrivateChannelPinsUpdateABC, channel, list_pin)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_channel_delete(self, channel: discord.abc.GuildChannel): async def on_guild_channel_delete(self, channel: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_guild_channel_delete:\n\t{channel}') self._logger.trace(__name__, f"Received on_guild_channel_delete:\n\t{channel}")
await self._handle_event(OnGuildChannelDeleteABC, channel) await self._handle_event(OnGuildChannelDeleteABC, channel)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_channel_create(self, channel: discord.abc.GuildChannel): async def on_guild_channel_create(self, channel: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_guild_channel_create:\n\t{channel}') self._logger.trace(__name__, f"Received on_guild_channel_create:\n\t{channel}")
await self._handle_event(OnGuildChannelCreateABC, channel) await self._handle_event(OnGuildChannelCreateABC, channel)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_channel_update(self, before: discord.abc.GuildChannel, after: discord.abc.GuildChannel): async def on_guild_channel_update(self, before: discord.abc.GuildChannel, after: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_guild_channel_update:\n\t{before}\n\t{after}') self._logger.trace(__name__, f"Received on_guild_channel_update:\n\t{before}\n\t{after}")
await self._handle_event(OnGuildChannelUpdateABC, before, after) await self._handle_event(OnGuildChannelUpdateABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_channel_pins_update(self, channel: discord.abc.GuildChannel, list_pin: Optional[datetime]): async def on_guild_channel_pins_update(self, channel: discord.abc.GuildChannel, list_pin: Optional[datetime]):
self._logger.trace(__name__, f'Received on_guild_channel_pins_update:\n\t{channel}\n\t{list_pin}') self._logger.trace(__name__, f"Received on_guild_channel_pins_update:\n\t{channel}\n\t{list_pin}")
await self._handle_event(OnGuildChannelPinsUpdateABC, channel, list_pin) await self._handle_event(OnGuildChannelPinsUpdateABC, channel, list_pin)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_integrations_update(self, guild: discord.Guild): async def on_guild_integrations_update(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_integrations_update:\n\t{guild}') self._logger.trace(__name__, f"Received on_guild_integrations_update:\n\t{guild}")
await self._handle_event(OnGuildIntegrationsUpdateABC, guild) await self._handle_event(OnGuildIntegrationsUpdateABC, guild)
@commands.Cog.listener() @commands.Cog.listener()
async def on_webhooks_update(self, channel: discord.abc.GuildChannel): async def on_webhooks_update(self, channel: discord.abc.GuildChannel):
self._logger.trace(__name__, f'Received on_webhooks_update:\n\t{channel}') self._logger.trace(__name__, f"Received on_webhooks_update:\n\t{channel}")
await self._handle_event(OnWebhooksUpdateABC, channel) await self._handle_event(OnWebhooksUpdateABC, channel)
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_join(self, member: discord.Member): async def on_member_join(self, member: discord.Member):
self._logger.trace(__name__, f'Received on_member_join:\n\t{member}') self._logger.trace(__name__, f"Received on_member_join:\n\t{member}")
await self._handle_event(OnMemberJoinABC, member) await self._handle_event(OnMemberJoinABC, member)
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_remove(self, member: discord.Member): async def on_member_remove(self, member: discord.Member):
self._logger.trace(__name__, f'Received on_member_remove:\n\t{member}') self._logger.trace(__name__, f"Received on_member_remove:\n\t{member}")
await self._handle_event(OnMemberRemoveABC, member) await self._handle_event(OnMemberRemoveABC, member)
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_update(self, before: discord.Member, after: discord.Member): async def on_member_update(self, before: discord.Member, after: discord.Member):
self._logger.trace(__name__, f'Received on_member_update:\n\t{before}\n\t{after}') self._logger.trace(__name__, f"Received on_member_update:\n\t{before}\n\t{after}")
await self._handle_event(OnMemberUpdateABC, before, after) await self._handle_event(OnMemberUpdateABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_user_update(self, before: discord.User, after: discord.User): async def on_user_update(self, before: discord.User, after: discord.User):
self._logger.trace(__name__, f'Received on_user_update:\n\t{before}\n\t{after}') self._logger.trace(__name__, f"Received on_user_update:\n\t{before}\n\t{after}")
await self._handle_event(OnUserUpdateABC, before, after) await self._handle_event(OnUserUpdateABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_join(self, guild: discord.Guild): async def on_guild_join(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_join:\n\t{guild}') self._logger.trace(__name__, f"Received on_guild_join:\n\t{guild}")
await self._handle_event(OnGuildJoinABC, guild) await self._handle_event(OnGuildJoinABC, guild)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_remove(self, guild: discord.Guild): async def on_guild_remove(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_remove:\n\t{guild}') self._logger.trace(__name__, f"Received on_guild_remove:\n\t{guild}")
await self._handle_event(OnGuildRemoveABC, guild) await self._handle_event(OnGuildRemoveABC, guild)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_update(self, before: discord.Guild, after: discord.Guild): async def on_guild_update(self, before: discord.Guild, after: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_update:\n\t{before}\n\t{after}') self._logger.trace(__name__, f"Received on_guild_update:\n\t{before}\n\t{after}")
await self._handle_event(OnGuildUpdateABC, before, after) await self._handle_event(OnGuildUpdateABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_role_create(self, role: discord.Role): async def on_guild_role_create(self, role: discord.Role):
self._logger.trace(__name__, f'Received on_guild_role_create:\n\t{role}') self._logger.trace(__name__, f"Received on_guild_role_create:\n\t{role}")
await self._handle_event(OnGuildRoleCreateABC, role) await self._handle_event(OnGuildRoleCreateABC, role)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_role_delete(self, role: discord.Role): async def on_guild_role_delete(self, role: discord.Role):
self._logger.trace(__name__, f'Received on_guild_role_delete:\n\t{role}') self._logger.trace(__name__, f"Received on_guild_role_delete:\n\t{role}")
await self._handle_event(OnGuildRoleDeleteABC, role) await self._handle_event(OnGuildRoleDeleteABC, role)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_role_update(self, before: discord.Role, after: discord.Role): async def on_guild_role_update(self, before: discord.Role, after: discord.Role):
self._logger.trace(__name__, f'Received on_guild_role_update:\n\t{before}\n\t{after}') self._logger.trace(__name__, f"Received on_guild_role_update:\n\t{before}\n\t{after}")
await self._handle_event(OnGuildRoleUpdateABC, before, after) await self._handle_event(OnGuildRoleUpdateABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_emojis_update(self, guild: discord.Guild, before: Sequence[discord.Emoji], after: Sequence[discord.Emoji]): async def on_guild_emojis_update(
self._logger.trace(__name__, f'Received on_guild_emojis_update:\n\t{guild}\n\t{before}\n\t{after}') self, guild: discord.Guild, before: Sequence[discord.Emoji], after: Sequence[discord.Emoji]
):
self._logger.trace(__name__, f"Received on_guild_emojis_update:\n\t{guild}\n\t{before}\n\t{after}")
await self._handle_event(OnGuildEmojisUpdateABC, guild, before, after) await self._handle_event(OnGuildEmojisUpdateABC, guild, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_available(self, guild: discord.Guild): async def on_guild_available(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_available:\n\t{guild}') self._logger.trace(__name__, f"Received on_guild_available:\n\t{guild}")
await self._handle_event(OnGuildAvailableABC, guild) await self._handle_event(OnGuildAvailableABC, guild)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_unavailable(self, guild: discord.Guild): async def on_guild_unavailable(self, guild: discord.Guild):
self._logger.trace(__name__, f'Received on_guild_unavailable:\n\t{guild}') self._logger.trace(__name__, f"Received on_guild_unavailable:\n\t{guild}")
await self._handle_event(OnGuildUnavailableABC, guild) await self._handle_event(OnGuildUnavailableABC, guild)
@commands.Cog.listener() @commands.Cog.listener()
async def on_scheduled_event_create(self, event: discord.ScheduledEvent): async def on_scheduled_event_create(self, event: discord.ScheduledEvent):
self._logger.trace(__name__, f'Received on_scheduled_event_create:\n\t{event}') self._logger.trace(__name__, f"Received on_scheduled_event_create:\n\t{event}")
await self._handle_event(OnScheduledEventCreateABC, event) await self._handle_event(OnScheduledEventCreateABC, event)
@commands.Cog.listener() @commands.Cog.listener()
async def on_scheduled_event_delete(self, event: discord.ScheduledEvent): async def on_scheduled_event_delete(self, event: discord.ScheduledEvent):
self._logger.trace(__name__, f'Received on_scheduled_event_delete:\n\t{event}') self._logger.trace(__name__, f"Received on_scheduled_event_delete:\n\t{event}")
await self._handle_event(OnScheduledEventDeleteABC, event) await self._handle_event(OnScheduledEventDeleteABC, event)
@commands.Cog.listener() @commands.Cog.listener()
async def on_scheduled_event_update(self, before: discord.ScheduledEvent, after: discord.ScheduledEvent): async def on_scheduled_event_update(self, before: discord.ScheduledEvent, after: discord.ScheduledEvent):
self._logger.trace(__name__, f'Received on_scheduled_event_update:\n\t{before}, {after}') self._logger.trace(__name__, f"Received on_scheduled_event_update:\n\t{before}, {after}")
await self._handle_event(OnScheduledEventUpdateABC, before, after) await self._handle_event(OnScheduledEventUpdateABC, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_scheduled_event_user_add(self, event: discord.ScheduledEvent, user: discord.User): async def on_scheduled_event_user_add(self, event: discord.ScheduledEvent, user: discord.User):
self._logger.trace(__name__, f'Received on_scheduled_event_user_add:\n\t{event}, {user}') self._logger.trace(__name__, f"Received on_scheduled_event_user_add:\n\t{event}, {user}")
await self._handle_event(OnScheduledEventUserAddABC, event, user) await self._handle_event(OnScheduledEventUserAddABC, event, user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_scheduled_event_user_remove(self, event: discord.ScheduledEvent, user: discord.User): async def on_scheduled_event_user_remove(self, event: discord.ScheduledEvent, user: discord.User):
self._logger.trace(__name__, f'Received on_scheduled_event_user_remove:\n\t{event}, {user}') self._logger.trace(__name__, f"Received on_scheduled_event_user_remove:\n\t{event}, {user}")
await self._handle_event(OnScheduledEventUserRemoveABC, event, user) await self._handle_event(OnScheduledEventUserRemoveABC, event, user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_voice_state_update(self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState): async def on_voice_state_update(
self._logger.trace(__name__, f'Received on_voice_state_update:\n\t{member}\n\t{before}\n\t{after}') self, member: discord.Member, before: discord.VoiceState, after: discord.VoiceState
):
self._logger.trace(__name__, f"Received on_voice_state_update:\n\t{member}\n\t{before}\n\t{after}")
await self._handle_event(OnVoiceStateUpdateABC, member, before, after) await self._handle_event(OnVoiceStateUpdateABC, member, before, after)
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_ban(self, guild: discord.Guild, user: discord.User): async def on_member_ban(self, guild: discord.Guild, user: discord.User):
self._logger.trace(__name__, f'Received on_member_ban:\n\t{guild}\n\t{user}') self._logger.trace(__name__, f"Received on_member_ban:\n\t{guild}\n\t{user}")
await self._handle_event(OnMemberBanABC, guild, user) await self._handle_event(OnMemberBanABC, guild, user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_unban(self, guild: discord.Guild, user: discord.User): async def on_member_unban(self, guild: discord.Guild, user: discord.User):
self._logger.trace(__name__, f'Received on_member_unban:\n\t{guild}\n\t{user}') self._logger.trace(__name__, f"Received on_member_unban:\n\t{guild}\n\t{user}")
await self._handle_event(OnMemberUnbanABC, guild, user) await self._handle_event(OnMemberUnbanABC, guild, user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_invite_create(self, invite: discord.Invite): async def on_invite_create(self, invite: discord.Invite):
self._logger.trace(__name__, f'Received on_invite_create:\n\t{invite}') self._logger.trace(__name__, f"Received on_invite_create:\n\t{invite}")
await self._handle_event(OnInviteCreateABC, invite) await self._handle_event(OnInviteCreateABC, invite)
@commands.Cog.listener() @commands.Cog.listener()
async def on_invite_delete(self, invite: discord.Invite): async def on_invite_delete(self, invite: discord.Invite):
self._logger.trace(__name__, f'Received on_invite_create:\n\t{invite}') self._logger.trace(__name__, f"Received on_invite_create:\n\t{invite}")
await self._handle_event(OnInviteDeleteABC, invite) await self._handle_event(OnInviteDeleteABC, invite)
@commands.Cog.listener() @commands.Cog.listener()
async def on_group_join(self, channel: discord.GroupChannel, user: discord.User): async def on_group_join(self, channel: discord.GroupChannel, user: discord.User):
self._logger.trace(__name__, f'Received on_group_join:\n\t{channel}\n\t{user}') self._logger.trace(__name__, f"Received on_group_join:\n\t{channel}\n\t{user}")
await self._handle_event(OnGroupJoinABC, channel, user) await self._handle_event(OnGroupJoinABC, channel, user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_group_remove(self, channel: discord.GroupChannel, user: discord.User): async def on_group_remove(self, channel: discord.GroupChannel, user: discord.User):
self._logger.trace(__name__, f'Received on_group_remove:\n\t{channel}\n\t{user}') self._logger.trace(__name__, f"Received on_group_remove:\n\t{channel}\n\t{user}")
await self._handle_event(OnGroupRemoveABC, channel, user) await self._handle_event(OnGroupRemoveABC, channel, user)

View File

@ -2,7 +2,12 @@ from typing import Optional, Callable, Union, Iterable
from cpl_query._helper import is_number from cpl_query._helper import is_number
from cpl_query.base.sequence import Sequence from cpl_query.base.sequence import Sequence
from cpl_query.exceptions import InvalidTypeException, ArgumentNoneException, ExceptionArgument, IndexOutOfRangeException from cpl_query.exceptions import (
InvalidTypeException,
ArgumentNoneException,
ExceptionArgument,
IndexOutOfRangeException,
)
def _default_lambda(x: object) -> object: def _default_lambda(x: object) -> object:
@ -10,7 +15,6 @@ def _default_lambda(x: object) -> object:
class QueryableABC(Sequence): class QueryableABC(Sequence):
def __init__(self, t: type, values: Iterable = None): def __init__(self, t: type, values: Iterable = None):
Sequence.__init__(self, t, values) Sequence.__init__(self, t, values)
@ -99,7 +103,7 @@ class QueryableABC(Sequence):
return self.where(_func).count() return self.where(_func).count()
def distinct(self, _func: Callable = None) -> 'QueryableABC': def distinct(self, _func: Callable = None) -> "QueryableABC":
r"""Returns list without redundancies r"""Returns list without redundancies
Parameter Parameter
@ -208,7 +212,7 @@ class QueryableABC(Sequence):
return self return self
def group_by(self, _func: Callable = None) -> 'QueryableABC': def group_by(self, _func: Callable = None) -> "QueryableABC":
r"""Groups by func r"""Groups by func
Returns Returns
@ -289,11 +293,7 @@ class QueryableABC(Sequence):
result = self.order_by(_func).select(_func).to_list() result = self.order_by(_func).select(_func).to_list()
length = len(result) length = len(result)
i = int(length / 2) i = int(length / 2)
return ( return result[i] if length % 2 == 1 else (float(result[i - 1]) + float(result[i])) / float(2)
result[i]
if length % 2 == 1
else (float(result[i - 1]) + float(result[i])) / float(2)
)
def min(self, _func: Callable = None) -> object: def min(self, _func: Callable = None) -> object:
r"""Returns the lowest value r"""Returns the lowest value
@ -315,7 +315,7 @@ class QueryableABC(Sequence):
return _func(min(self, key=_func)) return _func(min(self, key=_func))
def order_by(self, _func: Callable = None) -> 'OrderedQueryableABC': def order_by(self, _func: Callable = None) -> "OrderedQueryableABC":
r"""Sorts elements by function in ascending order r"""Sorts elements by function in ascending order
Parameter Parameter
@ -331,9 +331,10 @@ class QueryableABC(Sequence):
_func = _default_lambda _func = _default_lambda
from cpl_query.base.ordered_queryable import OrderedQueryable from cpl_query.base.ordered_queryable import OrderedQueryable
return OrderedQueryable(self.type, sorted(self, key=_func), _func) return OrderedQueryable(self.type, sorted(self, key=_func), _func)
def order_by_descending(self, _func: Callable = None) -> 'OrderedQueryableABC': def order_by_descending(self, _func: Callable = None) -> "OrderedQueryableABC":
r"""Sorts elements by function in descending order r"""Sorts elements by function in descending order
Parameter Parameter
@ -349,9 +350,10 @@ class QueryableABC(Sequence):
_func = _default_lambda _func = _default_lambda
from cpl_query.base.ordered_queryable import OrderedQueryable from cpl_query.base.ordered_queryable import OrderedQueryable
return OrderedQueryable(self.type, sorted(self, key=_func, reverse=True), _func) return OrderedQueryable(self.type, sorted(self, key=_func, reverse=True), _func)
def reverse(self) -> 'QueryableABC': def reverse(self) -> "QueryableABC":
r"""Reverses list r"""Reverses list
Returns Returns
@ -360,7 +362,7 @@ class QueryableABC(Sequence):
""" """
return type(self)(self._type, reversed(self._values)) return type(self)(self._type, reversed(self._values))
def select(self, _func: Callable) -> 'QueryableABC': def select(self, _func: Callable) -> "QueryableABC":
r"""Formats each element of list to a given format r"""Formats each element of list to a given format
Returns Returns
@ -372,7 +374,7 @@ class QueryableABC(Sequence):
return type(self)(object, [_func(_o) for _o in self]) return type(self)(object, [_func(_o) for _o in self])
def select_many(self, _func: Callable) -> 'QueryableABC': def select_many(self, _func: Callable) -> "QueryableABC":
r"""Flattens resulting lists to one r"""Flattens resulting lists to one
Returns Returns
@ -396,9 +398,9 @@ class QueryableABC(Sequence):
Exception: when argument is None or found more than one element Exception: when argument is None or found more than one element
""" """
if self.count() > 1: if self.count() > 1:
raise Exception('Found more than one element') raise Exception("Found more than one element")
elif self.count() == 0: elif self.count() == 0:
raise Exception('Found no element') raise Exception("Found no element")
return self._values[0] return self._values[0]
@ -410,13 +412,13 @@ class QueryableABC(Sequence):
Found value: Optional[any] Found value: Optional[any]
""" """
if self.count() > 1: if self.count() > 1:
raise Exception('Index out of range') raise Exception("Index out of range")
elif self.count() == 0: elif self.count() == 0:
return None return None
return self._values[0] return self._values[0]
def skip(self, _index: int) -> 'QueryableABC': def skip(self, _index: int) -> "QueryableABC":
r"""Skips all elements from index r"""Skips all elements from index
Parameter Parameter
@ -433,7 +435,7 @@ class QueryableABC(Sequence):
return type(self)(self.type, self._values[_index:]) return type(self)(self.type, self._values[_index:])
def skip_last(self, _index: int) -> 'QueryableABC': def skip_last(self, _index: int) -> "QueryableABC":
r"""Skips all elements after index r"""Skips all elements after index
Parameter Parameter
@ -475,7 +477,7 @@ class QueryableABC(Sequence):
return result return result
def split(self, _func: Callable) -> 'QueryableABC': def split(self, _func: Callable) -> "QueryableABC":
r"""Splits the list by given function r"""Splits the list by given function
@ -508,7 +510,7 @@ class QueryableABC(Sequence):
return type(self)(self._type, query_groups) return type(self)(self._type, query_groups)
def take(self, _index: int) -> 'QueryableABC': def take(self, _index: int) -> "QueryableABC":
r"""Takes all elements from index r"""Takes all elements from index
Parameter Parameter
@ -525,7 +527,7 @@ class QueryableABC(Sequence):
return type(self)(self._type, self._values[:_index]) return type(self)(self._type, self._values[:_index])
def take_last(self, _index: int) -> 'QueryableABC': def take_last(self, _index: int) -> "QueryableABC":
r"""Takes all elements after index r"""Takes all elements after index
Parameter Parameter
@ -544,7 +546,7 @@ class QueryableABC(Sequence):
return type(self)(self._type, self._values[index:]) return type(self)(self._type, self._values[index:])
def where(self, _func: Callable = None) -> 'QueryableABC': def where(self, _func: Callable = None) -> "QueryableABC":
r"""Select element by function r"""Select element by function
Parameter Parameter