Refactor and optimize commands

Author: JrooTJunior

aiogram_cli/__init__.py

@@ -1 +1 @@
-__version__ = "1.0.0"
+__version__ = "1.0.1"

aiogram_cli/develop/_dispatcher.py

@@ -0,0 +1,46 @@
+import inspect
+from typing import Any
+
+from aiogram import Bot, Dispatcher
+
+from aiogram_cli.develop._resolver import logger, resolve_import
+
+
+async def do_skip_updates(bot: Bot):
+    logger.warning("Skipping updates")
+    await bot.delete_webhook(drop_pending_updates=True)
+
+
+def resolve_dispatcher(*, target: str) -> Dispatcher:
+    return resolve_import(target=target, possible_names={"dp", "dispatcher"})
+
+
+async def prepare_dispatcher(*, dispatcher: Any, skip_updates: bool) -> Dispatcher:
+    if callable(dispatcher):
+        dispatcher = dispatcher()
+
+    if inspect.isawaitable(dispatcher):
+        dispatcher = await dispatcher
+
+    if not isinstance(dispatcher, Dispatcher):
+        msg = f"Dispatcher must be instance of aiogram.Dispatcher, got: {type(dispatcher)}"
+        raise ValueError(msg)
+
+    if skip_updates:
+        dispatcher.startup.register(do_skip_updates)
+
+    return dispatcher
+
+
+def simplified_prepare_dispatcher(*, dispatcher: Any, skip_updates: bool) -> Dispatcher:
+    if callable(dispatcher):
+        dispatcher = dispatcher()
+
+    if not isinstance(dispatcher, Dispatcher):
+        msg = f"Dispatcher must be instance of aiogram.Dispatcher, got: {type(dispatcher)}"
+        raise ValueError(msg)
+
+    if skip_updates:
+        dispatcher.startup.register(do_skip_updates)
+
+    return dispatcher

aiogram_cli/develop/_polling.py

@@ -1,13 +1,15 @@
 import asyncio
 import sys
 from logging import basicConfig
+from typing import Any
 
 import aiogram
 import click
-from aiogram import Bot, Dispatcher
+from aiogram import Bot
 from aiogram.enums import ParseMode
 
-from aiogram_cli.develop._resolver import LoadError, resolve_dispatcher
+from aiogram_cli.develop._resolver import LoadError
+from aiogram_cli.develop._dispatcher import prepare_dispatcher, resolve_dispatcher
 
 
 def start_polling(
@@ -29,26 +31,29 @@ def start_polling(
 
     click.echo("Loading application...")
     try:
-        dispatcher = resolve_dispatcher(target=target, skip_updates=skip_updates)
+        dispatcher = resolve_dispatcher(target=target)
     except LoadError as e:
         click.echo(str(e), err=True)
         sys.exit(2)
 
+    # TODO: Add possibility to specify bot instance
     bot = Bot(
         token=token,
         parse_mode=parse_mode,
         disable_web_page_preview=disable_web_page_preview,
         protect_content=protect_content,
     )
-    click.echo("Start polling")
-    asyncio.run(_polling(bot=bot, dispatcher=dispatcher))
+
+    asyncio.run(_polling(bot=bot, dispatcher=dispatcher, skip_updates=skip_updates))
     return 0
 
 
 async def _polling(
     *,
-    dispatcher: Dispatcher,
+    dispatcher: Any,
     bot: Bot,
+    skip_updates: bool = False,
 ):
+    dispatcher = await prepare_dispatcher(dispatcher=dispatcher, skip_updates=skip_updates)
     async with bot.context() as bot:
         await dispatcher.start_polling(bot, close_bot_session=False)

aiogram_cli/develop/_resolver.py

@@ -1,7 +1,6 @@
 import importlib
 import logging
-
-from aiogram import Bot, Dispatcher
+from typing import Any
 
 logger = logging.getLogger(__name__)
 
@@ -10,42 +9,24 @@ class LoadError(Exception):
     pass
 
 
-def resolve_dispatcher(*, target: str, skip_updates: bool) -> Dispatcher:
+def resolve_import(*, target: str, possible_names: set[str] | None = None) -> Any:
     module_name, _, target_name = target.partition(":")
-
     try:
         module = importlib.import_module(module_name)
-    except ImportError:
-        msg = f"Can't import module: {module_name!r}"
-        raise LoadError(msg) from None
-
+    except ImportError as e:
+        msg = f"Can't import module: {module_name!r} ({type(e).__name__}: {e})"
+        raise LoadError(msg) from e
     if not target_name:
-        for name in ["dp", "dispatcher"]:
+        for name in possible_names:
             if hasattr(module, name):
                 target_name = name
                 break
         else:
-            msg = f"Can't find dispatcher in module: {module_name!r}"
+            msg = f"Can't find target name in module: {module_name!r} (checked names: {possible_names})"
             raise LoadError(msg)
 
-    dispatcher = getattr(module, target_name, None)
-    if not dispatcher:
-        msg = f"Can't find dispatcher in module: {module_name!r} by name: {target_name!r}"
-        raise LoadError(msg)
-
-    if callable(dispatcher):
-        dispatcher = dispatcher()
-
-    if not isinstance(dispatcher, Dispatcher):
-        msg = f"Dispatcher must be instance of aiogram.Dispatcher, got: {type(dispatcher)}"
-        raise LoadError(msg)
-
-    if skip_updates:
-        dispatcher.startup.register(_skip_updates)
-
-    return dispatcher
-
-
-async def _skip_updates(bot: Bot):
-    logger.warning("Skipping updates")
-    await bot.delete_webhook(drop_pending_updates=True)
+    try:
+        return getattr(module, target_name, None)
+    except AttributeError as e:
+        msg = f"Can't find target name in module: {module_name!r} by name: {target_name!r}"
+        raise LoadError(msg) from e

aiogram_cli/develop/_webhook.py

@@ -9,7 +9,8 @@
 from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
 from aiohttp.web import Application, run_app
 
-from aiogram_cli.develop._resolver import LoadError, resolve_dispatcher
+from aiogram_cli.develop._resolver import LoadError
+from aiogram_cli.develop._dispatcher import resolve_dispatcher, simplified_prepare_dispatcher
 
 logger = logging.getLogger(__name__)
 
@@ -38,11 +39,12 @@ def start_webhook(
 
     click.echo("Loading application...")
     try:
-        dispatcher = resolve_dispatcher(target=target, skip_updates=skip_updates)
+        dispatcher = resolve_dispatcher(target=target)
     except LoadError as e:
         click.echo(str(e), err=True)
         sys.exit(2)
 
+    dispatcher = simplified_prepare_dispatcher(dispatcher=dispatcher, skip_updates=skip_updates)
     bot = Bot(
         token=token,
         parse_mode=parse_mode,

aiogram_cli/develop/plugin.py

@@ -1,15 +1,28 @@
-from logging import getLevelNamesMapping
+from contextlib import suppress
 from os import getenv
 from pathlib import Path
 
 import click
+from aiogram import Bot
 from aiogram.enums import ParseMode
-from click import Choice
+from click import Choice, style
 from watchfiles import run_process
 from watchfiles.main import FileChange
 
 from aiogram_cli.develop._polling import start_polling
 from aiogram_cli.develop._webhook import start_webhook
+from aiogram_cli.wraps import async_command
+
+LOGGING_LEVELS = [
+    "CRITICAL",
+    "FATAL",
+    "ERROR",
+    "WARN",
+    "WARNING",
+    "INFO",
+    "DEBUG",
+    "NOTSET",
+]
 
 
 @click.group("run")
@@ -53,7 +66,7 @@ def develop_runner():
 @click.option(
     "--log-level",
     "-l",
-    type=Choice(list(getLevelNamesMapping().keys())),
+    type=Choice(LOGGING_LEVELS, case_sensitive=False),
     help="Logging level",
 )
 @click.option(
@@ -103,15 +116,16 @@ def command_polling(
         "log_level": log_level,
         "log_format": log_format,
     }
-    if reload:
-        run_process(
-            *reload_path,
-            target=start_polling,
-            kwargs=kwargs,
-            callback=_changes_detected,
-        )
-    else:
-        return start_polling(**kwargs)
+    with suppress(KeyboardInterrupt):
+        if reload:
+            run_process(
+                *reload_path,
+                target=start_polling,
+                kwargs=kwargs,
+                callback=_changes_detected,
+            )
+        else:
+            return start_polling(**kwargs)
 
 
 @develop_runner.command("webhook")
@@ -183,7 +197,7 @@ def command_polling(
 @click.option(
     "--log-level",
     "-l",
-    type=Choice(list(getLevelNamesMapping().keys())),
+    type=Choice(LOGGING_LEVELS, case_sensitive=False),
     help="Logging level",
 )
 @click.option(
@@ -247,17 +261,64 @@ def command_webhook(
         "log_level": log_level,
         "log_format": log_format,
     }
-    if reload:
-        run_process(
-            *reload_path,
-            target=start_webhook,
-            kwargs=kwargs,
-            callback=_changes_detected,
-        )
-    else:
-        return start_webhook(**kwargs)
+    with suppress(KeyboardInterrupt):
+        if reload:
+            run_process(
+                *reload_path,
+                target=start_webhook,
+                kwargs=kwargs,
+                callback=_changes_detected,
+            )
+        else:
+            return start_webhook(**kwargs)
 
 
 def _changes_detected(changes: set[FileChange]):
     files = {path for change, path in changes}
     click.echo(f"Changes detected: {files}")
+
+
+@develop_runner.command("info")
+@click.option(
+    "--token",
+    "-t",
+    help="Bot token",
+    required=True,
+    default=getenv("TELEGRAM_TOKEN"),
+)
+@async_command
+async def command_info(token: str):
+    async with Bot(token=token).context() as bot:
+        me = await bot.get_me()
+        webhook_info = await bot.get_webhook_info()
+
+    click.echo(style("Bot info:", fg="green", bold=True))
+    click.echo("  " + style("ID: ", bold=True, fg="green") + str(me.id))
+    click.echo("  " + style("Username: ", bold=True, fg="green") + f"@{me.username} (https://t.me/{me.username})")
+    if me.is_premium:
+        click.echo("  " + style("Premium: ", bold=True, fg="green") + "Yes")
+    if webhook_info.url:
+        click.echo(
+            style("Webhook info:", fg="green", bold=True),
+        )
+        click.echo("  " + style("URL: ", bold=True, fg="green") + webhook_info.url)
+    if webhook_info.allowed_updates:
+        click.echo("  " + style("Allowed updates: ", bold=True, fg="green") + ", ".join(webhook_info.allowed_updates))
+    if webhook_info.ip_address:
+        click.echo("  " + style("IP address: ", bold=True, fg="green") + webhook_info.ip_address)
+    if webhook_info.has_custom_certificate:
+        click.echo("  " + style("Has custom certificate: ", bold=True, fg="green") + "Yes")
+    if webhook_info.pending_update_count:
+        click.echo("  " + style("Pending updates: ", bold=True, fg="green") + str(webhook_info.pending_update_count))
+    if webhook_info.last_error_date:
+        click.echo("  " + style("Last error date: ", bold=True, fg="green") + str(webhook_info.last_error_date))
+    if webhook_info.last_error_message:
+        click.echo("  " + style("Last error message: ", bold=True, fg="green") + webhook_info.last_error_message)
+    if webhook_info.max_connections:
+        click.echo("  " + style("Max connections: ", bold=True, fg="green") + str(webhook_info.max_connections))
+    if webhook_info.last_synchronization_error_date:
+        click.echo(
+            "  "
+            + style("Last synchronization error date: ", bold=True, fg="green")
+            + str(webhook_info.last_synchronization_error_date)
+        )

aiogram_cli/main.py

@@ -1,8 +1,12 @@
+from collections import defaultdict
+from importlib.metadata import EntryPoint
+
+import aiogram
 import click
 from click_plugins import with_plugins
-from pkg_resources import iter_entry_points
 
 from aiogram_cli import __version__
+from aiogram_cli.resolver import iter_entry_points
 
 
 @with_plugins(iter_entry_points("aiogram_cli.plugins"))
@@ -12,3 +16,44 @@
 @click.version_option(version=__version__, prog_name="aiogram-cli", message="%(version)s")
 def cli():
     pass
+
+
+@cli.command("version")
+def command_version():
+    """Get an application version"""
+    click.echo(f"aiogram-cli: v{__version__}")
+    click.echo(f"aiogram: v{aiogram.__version__}")
+
+
+@cli.command("plugins")
+def command_plugins():
+    """Get a list of available plugins"""
+    entrypoint: EntryPoint
+    broken_libs = defaultdict(set)
+    for index, entrypoint in enumerate(iter_entry_points("aiogram_cli.plugins"), start=1):
+        click.echo(f"{index:3}.", nl=False)
+
+        dist = entrypoint.dist
+        try:
+            plugin = entrypoint.load()
+        except Exception as e:
+            broken_libs[dist.name].add(entrypoint.name)
+            click.secho(f" {entrypoint.name}", fg="red", err=True, nl=False)
+            click.echo(f" (by {dist.name} v{dist.version})", err=True, nl=False)
+            click.echo(f" is unusable in due to {type(e).__name__}: {e}")
+            continue
+
+        description = plugin.__doc__.split("\n")[0] if plugin.__doc__ else ""
+
+        click.secho(f" {entrypoint.name}", fg="green", nl=False)
+        click.echo(f" (by {dist.name} v{dist.version})", nl=False)
+        if description:
+            click.echo(f": {description}")
+        else:
+            click.echo()
+
+    if broken_libs:
+        click.echo()
+        click.secho("Broken libraries:", fg="red", err=True)
+        for lib, plugins in broken_libs.items():
+            click.echo(f"  {lib} ({', '.join(plugins)})")