feat(tg_message): Save sended messages

Save sended messages in sqlite database to delete later

.gitignore

@@ -1,2 +1,3 @@
 __pycache__
-.venv
+.venv
+sql_app.db

src/crud.py

@@ -0,0 +1,12 @@
+from sqlalchemy.orm import Session
+from . import models, schemas
+
+def get_TelegramBotMessage(db: Session, id: int):
+    return db.query(models.TelegramBotMessages).filter(models.TelegramBotMessages.id == id).first()
+
+def create_TelegramBotMessage(db: Session, telegram_bot_message: schemas.TelegramBotMessageCreate):
+    db_TelegramBotMessages = models.TelegramBotMessages(message_id=telegram_bot_message.message_id, chat_id=telegram_bot_message.chat_id, datetime=telegram_bot_message.datetime)
+    db.add(db_TelegramBotMessages)
+    db.commit()
+    db.refresh(db_TelegramBotMessages)
+    return db_TelegramBotMessages

src/database.py

@@ -0,0 +1,13 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
+)
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

src/docs.py

@@ -0,0 +1,11 @@
+sendMessageExamples = {
+    "normal": {
+        "summary": "A normal example",
+        "description": "A **normal** body that works correctly",
+        "value": {
+            "chat_id": "fdhgzuew",
+            "message": "Hewwo",
+            "bot_token": "dqwtftzwfz"
+        }
+    }
+}

src/main.py

@@ -1,36 +1,47 @@
 import uvicorn
-import json
+import requests
+from typing import Annotated
 from pydantic import BaseModel
-from fastapi import FastAPI, status
-from fastapi.responses import JSONResponse
-from os.path import expanduser
-from pathlib import Path
+from datetime import datetime
+from sqlalchemy.orm import Session
+from fastapi import FastAPI, Body, Depends, HTTPException
+
+from . import crud, models, schemas
+from .database import SessionLocal, engine
+from .docs import sendMessageExamples
+
+models.Base.metadata.create_all(bind=engine)
 
 app = FastAPI()
-Path(f"{expanduser("~")}/.local/share/telegram-autoclear").mkdir(parents=True, exist_ok=True)
 
-class SendMessageBody(BaseModel):
-    name: str
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
 
 @app.get("/")
 def read_root():
     return {"Made by": "Nikurasu", "Version": "0.0.0"}
 
-@app.post("/send_message")
-def send_message(body: SendMessageBody):
-    with open(f"{expanduser("~")}/.local/share/telegram-autoclear/messages.json", "r") as messages_file:
-        try:
-            messages_data = json.load(messages_file)
-        except Exception as e:
-            return JSONResponse(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": f"{e}"})
-        if "messages" not in messages_data:
-            write_data = {"messages": [{"name": body.name}]}
-        else:
-            write_data = {"messages": messages_data["messages"]}
-            write_data["messages"].append({"name": body.name})
-    with open(f"{expanduser("~")}/.local/share/telegram-autoclear/messages.json", "w") as messages_file:
-        json.dump(write_data, messages_file)
-    return {"name": body.name}    
+
+
+@app.post("/send_message", response_model=schemas.TelegramBotMessage)
+def send_message(body: Annotated[schemas.SendMessagePostBody, Body(openapi_examples=sendMessageExamples)], db: Session = Depends(get_db)):
+    tg_url = f"https://api.telegram.org/bot{body.bot_token}/sendMessage"
+    tg_body = {"chat_id": body.chat_id, "text": body.message}
+    tg_result = requests.post(url=tg_url, json=tg_body)
+    if tg_result.status_code != 200:
+        raise HTTPException(status_code=500, detail=f"internal server error")
+    res_json = tg_result.json()
+    chat_id_from_result = res_json["result"]["chat"]["id"]
+    message_id_from_result = res_json["result"]["message_id"]
+    return crud.create_TelegramBotMessage(db=db, telegram_bot_message=schemas.TelegramBotMessageCreate(
+        chat_id=str(chat_id_from_result),
+        message_id=str(message_id_from_result),
+        datetime=datetime.now()
+    ))
 
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=8080)

src/models.py

@@ -0,0 +1,11 @@
+from sqlalchemy import Column, Integer, String, DateTime
+from sqlalchemy.orm import relationship
+
+from .database import Base
+
+class TelegramBotMessages(Base):
+    __tablename__ = "telegram_bot_messages"
+    id = Column(Integer, primary_key=True)
+    message_id = Column(String, unique=True, nullable=False)
+    chat_id = Column(String, nullable=False)
+    datetime = Column(DateTime, nullable=False)

src/schemas.py

@@ -0,0 +1,20 @@
+from pydantic import BaseModel
+from datetime import datetime
+
+class TelegramBotMessageBase(BaseModel):
+    chat_id: str
+    message_id: str
+    datetime: datetime
+
+class TelegramBotMessage(TelegramBotMessageBase):
+    id: int
+    class Config:
+        from_attributes = True
+
+class TelegramBotMessageCreate(TelegramBotMessageBase):
+    pass
+    
+class SendMessagePostBody(BaseModel):
+    chat_id: str
+    bot_token: str
+    message: str