primera clase telegram

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "test"
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

poetry.lock

@@ -184,6 +184,20 @@ pytest = ">=7.0.0,<9"
 docs = ["sphinx (>=5.3)", "sphinx-rtd-theme (>=1.0)"]
 testing = ["coverage (>=6.2)", "hypothesis (>=5.7.1)"]
 
+[[package]]
+name = "python-dotenv"
+version = "1.0.1"
+description = "Read key-value pairs from a .env file and set them as environment variables"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "python-dotenv-1.0.1.tar.gz", hash = "sha256:e324ee90a023d808f1959c46bcbc04446a10ced277783dc6ee09987c37ec10ca"},
+    {file = "python_dotenv-1.0.1-py3-none-any.whl", hash = "sha256:f7b63ef50f1b690dddf550d03497b66d609393b40b564ed0d674909a68ebf16a"},
+]
+
+[package.extras]
+cli = ["click (>=5.0)"]
+
 [[package]]
 name = "sniffio"
 version = "1.3.0"
@@ -198,4 +212,4 @@ files = [
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.11"
-content-hash = "51da538602c65298d4821ae81cf77d4c8a96bfd352fcd6d90efe2e9de2e0cd2e"
+content-hash = "3abee5cfd82d3233384bc46dea8c2e067bcccf7ca8478215c95bc5c0907d1811"

pyproject.toml

@@ -9,7 +9,7 @@ readme = "README.md"
 [tool.poetry.dependencies]
 python = "^3.11"
 httpx = "^0.26.0"
-
+python-dotenv = "^1.0.1"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.0.0"
@@ -18,3 +18,7 @@ pytest-asyncio = "^0.23.5"
 [build-system]
 requires = ["poetry-core"]
 build-backend = "poetry.core.masonry.api"
+
+[tool.pyright]
+include = ["src"]
+defineConstant = { DEBUG = true }

src/telegram.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2024 Lorenzo Carbonell <a.k.a. atareao>
+
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+"""A module for telegram
+"""
+
+import logging
+from httpx import AsyncClient
+
+logger = logging.getLogger(__name__)
+
+
+class TelegramException(Exception):
+    """Telegram exception
+
+    Args:
+        Exception (_type_): exception
+    """
+
+
+class Telegram:
+    """A class to work with Telegram
+    """
+
+    def __init__(self, url: str, token: str):
+        """Init Telegram class
+
+        Args:
+            url (str): base url for telegram
+            token (str): token for bot
+        """
+        logger.info("__info__")
+        self._url = f"https://{url}/bot{token}"
+        self._headers = {
+            "Accept": "application/json",
+            "Content-Type": "application/json"
+        }
+
+    async def send_message(self, chat_id: int, thread_id: int, message: str):
+        """Send message to Telegram
+
+        Args:
+            chat_id (int): id for chat
+            thread_id (int): id for thread
+            message (str): message to send
+
+        Returns:
+            _type_: _description_
+        """
+        logger.info("send_message")
+        url = f"{self._url}/sendMessage"
+        payload = {
+            "chat_id": chat_id,
+            "message_thread_id": thread_id,
+            "parse_mode": "markdown",
+            "text": message
+        }
+        logger.debug(payload)
+        async with AsyncClient() as client:
+            response = await client.post(
+                url, headers=self._headers, json=payload
+            )
+            logger.debug(response)
+            if response.status_code == 200:
+                return response.json()

test/test_telegram.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+import os
+import pytest
+from dotenv import load_dotenv
+from src.telegram import Telegram
+
+pytest_plugins = ("pytest_asyncio",)
+
+
+@pytest.mark.asyncio
+async def test_telegram():
+    """Test telegram class
+    """
+    load_dotenv()
+    url = os.getenv("TELEGRAM_URL")
+    token = os.getenv("TELEGRAM_TOKEN")
+    chat_id = os.getenv("TELEGRAM_CHAT_ID")
+    thread_id = os.getenv("TELEGRAM_THREAD_ID")
+    if url and token and chat_id and thread_id:
+        telegram = Telegram(url, token)
+        chat_id = int(chat_id)
+        thread_id = int(thread_id)
+        message = "Esto es una prueba"
+        response = await telegram.send_message(chat_id, thread_id, message)
+    else:
+        response = None
+    assert response is not None
+    assert response != ""