added OwnerMiddleware

config.py

@@ -15,10 +15,10 @@ class Config(metaclass=Singleton):
     * `teletoken`: telegram api token.
     * `bot_owner`: telegram id of bot owner.
     """
-    base_url: str
-    port: int
-    teletoken: str
-    bot_owner: int
+    base_url: str = ''
+    port: int = 3001
+    teletoken: str = ''
+    bot_owner: int = 0
 
 
 class ConfigError(Exception):
@@ -28,7 +28,8 @@ class ConfigError(Exception):
 
 def _get_teletoken(cmd_arguments: Namespace) -> str:
     """
-        Get Telegram bot token.
+        Get Telegram bot token from command line arguments or
+        **LTLNOTIFYER_TELETOKEN** environment variable.
         Can raise **ConfigError** if Telegram token not set or incorrect.
 
         :param cmd_arguments: command line arguments.
@@ -46,13 +47,14 @@ def _get_teletoken(cmd_arguments: Namespace) -> str:
 
 def _get_bot_owner_id(cmd_arguments: Namespace) -> int:
     """
-        Get bot owner telegram id.
+        Get bot owner telegram id from command line arguments or
+        **LTLNOTIFYER_BOT_OWNER** environment variable.
         Can raise **ConfigError** if bot owner id not set or incorrect.
 
-        :param cmd_arguments: command line arguments.
+        :param cmd_arguments: command line arguments object.
         :return: bot owner Telegram id.
     """
-    bot_owner = getenv('LTLNOTIFYER_BOT_OWNER')
+    bot_owner = cmd_arguments.owner or getenv('LTLNOTIFYER_BOT_OWNER')
     if not bot_owner:
         raise ConfigError('Bot owner telegram id not set.')
     try:
@@ -66,32 +68,42 @@ def _get_bot_owner_id(cmd_arguments: Namespace) -> int:
 
 def _get_base_url(cmd_arguments: Namespace) -> str:
     """
-        Get server base url.
+        Get server base url from command line arguments or
+        **LTLNOTIFYER_BASE_URL** environment variable.
         Can raise **Config error** if base url not set.
 
-        :param cmd_arguments: command line arguments.
+        :param cmd_arguments: command line arguments object.
         :return: server base url.
         """
     base_url = cmd_arguments.base_url or getenv('LTLNOTIFYER_BASE_URL')
     if not base_url:
         raise ConfigError('Server base url not set.')
+    return base_url
 
 
 def _get_port(cmd_arguments: Namespace) -> int:
     """
-        Get server port.
+    Get server port from command line arguments or
+    **LTLNOTIFYER_PORT** environment variable.
 
-        :param cmd_arguments: command line arguments.
-        :return:
-        """
+    :param cmd_arguments: command line arguments object.
+    :return: port number or 3001 if port not set or incorrect.
+    """
     port = cmd_arguments.port or getenv('LTLNOTIFIER_PORT', default='')
     port = int(port) if port.isdigit() else 3001
+    return port
 
 
-def _init_config(arguments) -> None:
+def _init_config(arguments: Namespace) -> None:
+    """
+    Initialisation Config singleton.
+
+    :param arguments: command line arguments object.
+    """
     teletoken = _get_teletoken(arguments)
     bot_owner = _get_bot_owner_id(arguments)
-    base_url, port = _get_base_url(arguments)
+    base_url = _get_base_url(arguments)
+    port = _get_port(arguments)
     Config(teletoken=teletoken,
            port=port,
            bot_owner=bot_owner,
@@ -114,7 +126,7 @@ _parser.add_argument('--base_url', default='',
 _parser.add_argument('--port', default='',
                      help='Server port (default = 3001. \n'
                           'Can be set in LTLNOTIFYER_PORT environment '
-                          'variable.')
+                          'variable.)')
 
 _init_config(_parser.parse_args())
 

db/__init__.py

@@ -1,4 +1,4 @@
-import aiosqlite
+from os import path
 
 from .create import create_tables_if_not_exists
 
@@ -6,5 +6,5 @@ from .create import create_tables_if_not_exists
 DB_PATH = 'res/db/'
 DB_FILENAME = 'database.db'
 
-
-create_tables_if_not_exists(f'{DB_PATH}{DB_FILENAME}')
+if not path.exists(f'{DB_PATH}{DB_FILENAME}'):
+    create_tables_if_not_exists(f'{DB_PATH}{DB_FILENAME}')

db/create.py

@@ -39,6 +39,7 @@ Create database file and tables if not exists.
                                *(text, not null)*.
 """
 import sqlite3
+import logging
 
 
 CREATE_SERVICES_TABLE = '''CREATE TABLE IF NOT EXISTS services (
@@ -75,3 +76,4 @@ def create_tables_if_not_exists(dbfile: str) -> None:
     connection.commit()
     cursor.close()
     connection.close()
+    logging.info('Database created!')

start.py

@@ -2,11 +2,11 @@ import logging
 
 from aiohttp import web
 
-from config import config
+from config import Config
 from telegram import telegram_bot, telegram_dispatcher
-import db
 from web import create_app
 
+config = Config()
 app = create_app(telegram_bot, telegram_dispatcher, config)
 logging.basicConfig(level=logging.INFO)
 

telegram/__init__.py

@@ -1,18 +1,18 @@
 from aiogram import Bot, Dispatcher
-from aiogram.dispatcher.webhook import SendMessage
 from aiogram.contrib.middlewares.logging import LoggingMiddleware
 
-from config import config
+from config import Config
+
+from .commands import register_handlers
+from .middlewares import OwnerMiddleware
 
 
+config = Config()
 telegram_bot = Bot(token=config.teletoken)
 telegram_dispatcher = Dispatcher(telegram_bot)
 telegram_dispatcher.middleware.setup(LoggingMiddleware())
+telegram_dispatcher.middleware.setup(OwnerMiddleware())
+register_handlers(telegram_dispatcher)
 
 
-@telegram_dispatcher.message_handler(commands=['test'])
-async def test_telegram(message):
-    return SendMessage(message.chat.id, 'passed')
-
-
-__all__ = ('telegram_bot', telegram_dispatcher)
+__all__ = ('telegram_bot', 'telegram_dispatcher')

telegram/commands/__init__.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass
+import typing
+
+from aiogram import Dispatcher
+
+from .test import telegram_test
+
+
+@dataclass
+class Handler:
+    """
+    Dataclass of handlers with help string.
+    """
+    function: typing.Callable
+    help_string: str
+
+
+handlers: dict[str, Handler] = {
+    'test': Handler(telegram_test, 'Отвечает "passed"')
+}
+
+
+def register_handlers(dispatcher: Dispatcher):
+    for command, handler in handlers.items():
+        dispatcher.register_message_handler(handler.function,
+                                            commands=[command])

telegram/commands/test.py

@@ -0,0 +1,6 @@
+from aiogram.types import Message
+from aiogram.dispatcher.webhook import SendMessage
+
+
+async def telegram_test(message: Message) -> SendMessage:
+    return SendMessage(message.chat.id, 'passed!')

telegram/middlewares.py

@@ -0,0 +1,19 @@
+import logging
+
+from aiogram.types import Message
+from aiogram.dispatcher.middlewares import BaseMiddleware
+
+from config import Config
+
+
+class OwnerMiddleware(BaseMiddleware):
+    """
+    Ignore commands from non-owner.
+    """
+    def __init__(self):
+        self.bot_owner = Config().bot_owner
+        super(OwnerMiddleware, self).__init__()
+
+    async def on_pre_process_message(self, message: Message, _):
+        if message.chat.id != self.bot_owner:
+            message.text = ''

web/routes.py

@@ -1,9 +1,10 @@
 from aiohttp import web
 
-from config import config
+from config import Config
 from .views import test_route
 
 
+config = Config()
 TELEGRAM_WEBHOOK_ROUTE = '/' + config.teletoken
 
 routes = [