Base bot.

.gitignore

@@ -86,3 +86,7 @@ ENV/
 
 ## File-based project format:
 *.iws
+
+
+# Current project
+experiment.py

aiogram/__init__.py

@@ -1 +1,8 @@
+import logging
+
 __version__ = '0.1b'
+
+log = logging.getLogger('aiogram')
+
+API_URL = "https://api.telegram.org/bot{token}/{method}"
+FILE_URL = "https://api.telegram.org/file/bot{token}/{file_id}"

aiogram/api.py

@@ -0,0 +1,53 @@
+from aiogram import API_URL, log
+from aiogram.exceptions import ValidationError, TelegramAPIError
+
+
+def check_token(token):
+    if any(x.isspace() for x in token):
+        raise ValidationError('Token is invalid!')
+
+    left, sep, right = token.partition(':')
+    if (not sep) or (not left.isdigit()) or (len(left) < 3):
+        raise ValidationError('Token is invalid!')
+
+    return True
+
+
+async def _check_result(method_name, response):
+    """
+    Checks whether `result` is a valid API response.
+    A result is considered invalid if:
+        - The server returned an HTTP response code other than 200
+        - The content of the result is invalid JSON.
+        - The method call was unsuccessful (The JSON 'ok' field equals False)
+
+    :raises ApiException: if one of the above listed cases is applicable
+    :param method_name: The name of the method called
+    :param response: The returned response of the method request
+    :return: The result parsed to a JSON dictionary.
+    """
+    if response.status != 200:
+        body = await response.text()
+        raise TelegramAPIError(f"The server returned HTTP {response.status}. Response body:\n[{body}]",
+                               method_name, response.status, body)
+
+    result_json = await response.json()
+
+    if not result_json.get('ok'):
+        body = await response.text()
+        code = result_json.get('error_code')
+        description = result_json.get('description')
+        raise TelegramAPIError(f"Error code: {code} Description {description}",
+                               method_name, response.status, body)
+    return result_json.get('result')
+
+
+async def request(session, token, method, data=None):
+    log.debug(f"Make request: '{method}' with data: {data or {}}")
+    url = API_URL.format(token=token, method=method)
+    async with session.post(url, json=data) as response:
+        return await _check_result(method, response)
+
+
+class ApiMethods:
+    GET_ME = 'getMe'

aiogram/bot.py

@@ -0,0 +1,46 @@
+import asyncio
+import signal
+
+import aiohttp
+
+from . import api
+from .api import ApiMethods
+from .types.user import User
+
+
+class AIOGramBot:
+    def __init__(self, token, loop=None, connections_limit=10):
+        """
+        :param token: 
+        :param loop: 
+        :param connections_limit: 
+        """
+        api.check_token(token)
+        self.__token = token
+
+        if loop is None:
+            loop = asyncio.get_event_loop()
+
+        self.loop = loop
+        self.session = aiohttp.ClientSession(
+            connector=aiohttp.TCPConnector(limit=connections_limit),
+            loop=self.loop)
+
+        self.loop.add_signal_handler(signal.SIGINT, self._on_exit)
+
+    def _on_exit(self):
+        self.session.close()
+
+    def _prepare_object(self, obj):
+        obj.bot = self
+        return obj
+
+    @property
+    async def me(self) -> User:
+        if not hasattr(self, '_me'):
+            setattr(self, '_me', await self.get_me())
+        return getattr(self, '_me')
+
+    async def get_me(self) -> User:
+        raw = await api.request(self.session, self.__token, ApiMethods.GET_ME)
+        return self._prepare_object(User.de_json(raw))

aiogram/exceptions.py

@@ -0,0 +1,11 @@
+class ValidationError(Exception):
+    pass
+
+
+class TelegramAPIError(Exception):
+    def __init__(self, message, method, status, body):
+        super(TelegramAPIError, self).__init__(
+            f"A request to the Telegram API was unsuccessful (Status code: {status}). {message}")
+        self.method = method
+        self.status = status
+        self.body = body

aiogram/types/__init__.py

@@ -0,0 +1,61 @@
+import json
+
+
+class Serializable:
+    def to_json(self):
+        """
+        Returns a JSON string representation of this class.
+
+        This function must be overridden by subclasses.
+        :return: a JSON formatted string.
+        """
+        raise NotImplementedError
+
+
+class Deserializable:
+    """
+    Subclasses of this class are guaranteed to be able to be created from a json-style dict or json formatted string.
+    All subclasses of this class must override de_json.
+    """
+
+    @property
+    def bot(self):
+        if not hasattr(self, '_bot'):
+            raise AttributeError('object is not configured for bot.')
+        return getattr(self, '_bot')
+
+    @bot.setter
+    def bot(self, bot):
+        setattr(self, '_bot', bot)
+
+    def to_json(self):
+        return getattr(self, 'data', {})
+
+    @classmethod
+    def de_json(cls, data):
+        """
+        Returns an instance of this class from the given json dict or string.
+
+        This function must be overridden by subclasses.
+        :return: an instance of this class created from the given json dict or string.
+        """
+        raise NotImplementedError
+
+    @staticmethod
+    def check_json(data):
+        """
+        Checks whether json_type is a dict or a string. If it is already a dict, it is returned as-is.
+        If it is not, it is converted to a dict by means of json.loads(json_type)
+        :param data:
+        :return:
+        """
+
+        if isinstance(data, dict):
+            return data
+        elif isinstance(data, str):
+            return json.loads(data)
+        else:
+            raise ValueError("data should be a json dict or string.")
+
+    def __str__(self):
+        return json.dumps(self.to_json())

aiogram/types/user.py

@@ -0,0 +1,48 @@
+from aiogram.types import Deserializable
+
+
+class User(Deserializable):
+    __slots__ = ('data', 'id', 'first_name', 'last_name', 'username', 'language_code')
+
+    def __init__(self, data, id, first_name, last_name, username, language_code):
+        self.data: dict = data
+
+        self.id: int = id
+        self.first_name: str = first_name
+        self.last_name: str = last_name
+        self.username: str = username
+        self.language_code: str = language_code
+
+    @classmethod
+    def de_json(cls, data: str or dict) -> 'User':
+        """
+        id	Integer	Unique identifier for this user or bot
+        first_name	String	User‘s or bot’s first name
+        last_name	String	Optional. User‘s or bot’s last name
+        username	String	Optional. User‘s or bot’s username
+        language_code	String	Optional. IETF language tag of the user's language
+        :param data: 
+        :return: 
+        """
+        data = cls.check_json(data)
+
+        id = data.get('id')
+        first_name = data.get('first_name')
+        last_name = data.get('last_name')
+        username = data.get('username')
+        language_code = data.get('language_code')
+
+        return User(data, id, first_name, last_name, username, language_code)
+
+    @property
+    def full_name(self):
+        full_name = self.first_name
+        if self.last_name:
+            full_name += ' ' + self.last_name
+        return full_name
+
+    @property
+    def mention(self):
+        if self.username:
+            return '@' + self.username
+        return self.full_name