upgrade db adapter

Author: thehighestmath

example/db_adapter.py

@@ -1,98 +1,160 @@
 from datetime import datetime
 
-from .handlers.custom_types import Question
+import sqlalchemy
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
 
+from handlers import custom_types
 
-def get_connection_to_data_base():
+USERNAME = 'umschooluser'
+PASSWORD = 'umschoolpswd'
+DATABASE_NAME = 'umschooldb'
+PORT = 6432
+
+
+class Base(DeclarativeBase):
     pass
 
 
-def decorator_add_connection(func):
-    def wrapper(*args, **kwargs):
-        connection = get_connection_to_data_base()
-        result = func(*args, **kwargs, connection=connection)
-        connection.close()
-        return result
+class Question(Base):
+    __tablename__ = 'question'
+    id: Mapped[int] = mapped_column(primary_key=True)
+    question_text: Mapped[str] = mapped_column(sqlalchemy.String(255))
+    publish_date: Mapped[datetime] = mapped_column(sqlalchemy.DateTime)
 
-    return wrapper
+    def __repr__(self) -> str:
+        return f'Question(id={self.id}, question_text={self.question_text}, publish_date={self.publish_date})'
 
 
-@decorator_add_connection
-def create_tables_from_db(connection=None):
-    """
-    создать таблицы в бд
-    """
-    pass
+class Choice(Base):
+    __tablename__ = 'choice'
+    id: Mapped[int] = mapped_column(primary_key=True)
+    choice_text: Mapped[str] = mapped_column(sqlalchemy.String(255))
+    votes: Mapped[int] = mapped_column(sqlalchemy.Integer, default=0)
+    question_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey('question.id'))
+
+    def __repr__(self) -> str:
+        return f'Choice(id={self.id}, choice_text={self.choice_text}, votes={self.votes}, question_id={self.question_id})'
+
+
+class UserStat(Base):
+    __tablename__ = 'user_stat'
+    id: Mapped[int] = mapped_column(primary_key=True)
+    tg_user_id: Mapped[int] = mapped_column(sqlalchemy.Integer)
+    question_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey('question.id'))
+    choice_id: Mapped[int] = mapped_column(sqlalchemy.ForeignKey('choice.id'))
+
+    def __repr__(self) -> str:
+        return f'UserStat(id={self.id}, tg_user_id={self.tg_user_id}, question_id={self.question_id}, choice_id={self.choice_id})'
+
+
+def get_engine():
+    return sqlalchemy.create_engine('postgresql://{username}:{password}@localhost:{port}/{database_name}'.format(
+        username=USERNAME,
+        password=PASSWORD,
+        port=PORT,
+        database_name=DATABASE_NAME,
+    ))
+
+
+def get_session() -> Session:
+    return Session(get_engine())
 
 
-@decorator_add_connection
-def delete_question_by_id_db(question_id: int, connection=None):
+def decorator_add_session(func):
+    def wrapper(*args, **kwargs):
+        session = get_session()
+        try:
+            return func(*args, session=session, **kwargs)
+        finally:
+            session.close()
+    return wrapper
+
+
+def create_tables_in_db():
     """
-    удалить вопрос, варианты ответа,
-    а также всю статистику связанную с ним из бд
+    создать таблицы в бд
     """
-    pass
+    engine = get_engine()
+    Base.metadata.create_all(engine)
 
 
-@decorator_add_connection
+@decorator_add_session
 def add_question_to_db(
-    question_text: str, publish_date: datetime, connection=None
+    question_text: str, publish_date: datetime, session: Session
 ) -> int:
     """
     добавить вопрос в бд
     """
-    pass
+    with session.begin():
+        question = Question(question_text=question_text, publish_date=publish_date)
+        session.add(question)
+        session.commit()
+    return question.id
 
 
-@decorator_add_connection
-def add_choice_to_db(choice_text: str, question_id: int, connection=None):
+@decorator_add_session
+def add_choice_to_db(choice_text: str, question_id: int, session: Session):
     """
     добавить ответ в бд
     """
+    with session.begin():
+        choice = Choice(choice_text=choice_text, question_id=question_id)
+        session.add(choice)
+        session.commit()
+
+
+@decorator_add_session
+def delete_question_by_id_db(question_id: int, session: Session):
+    """
+    удалить вопрос, варианты ответа,
+    а также всю статистику связанную с ним из бд
+    """
     pass
 
 
-@decorator_add_connection
-def get_all_stat(connection=None) -> list[list]:
+@decorator_add_session
+def get_all_stat(session: Session) -> list[list]:
     """
     получить статистику пользователей
     """
     return [[]]
 
 
-@decorator_add_connection
-def get_personal_stat(telegram_id: int, connection=None) -> list[list]:
+@decorator_add_session
+def get_personal_stat(telegram_id: int, session: Session) -> list[list]:
     """
     получить личную статистику пользователя
     """
     return [[]]
 
 
-@decorator_add_connection
-def get_random_question(telegram_id: int, connection=None) -> Question:
+@decorator_add_session
+def get_random_question(telegram_id: int, session: Session) -> custom_types.Question:
     """
     получить случайный, неотвеченный вопрос из бд
 
     return: Question
     """
-    return Question(id=0, text="")
+    return custom_types.Question(id=0, text="")
 
 
-@decorator_add_connection
-def get_choices_by_question_id(question_id: int, connection=None) -> list[str]:
+@decorator_add_session
+def get_choices_by_question_id(question_id: int, session: Session) -> list[str]:
     """
     получить ответы по заданному question_id
     """
     return []
 
 
-@decorator_add_connection
-def add_user_vote_db(choice_id: int, telegram_id: int, connection=None):
+@decorator_add_session
+def add_user_vote_db(choice_id: int, telegram_id: int, session: Session):
     """
     добавить голос пользователя выбранному варианту ответа
     """
     pass
 
 
 if __name__ == "__main__":
-    create_tables_from_db()
+    create_tables_in_db()
+    add_question_to_db("test-question", datetime.now())
+    add_choice_to_db("test-choice", 20)

example/docker-compose.yml

@@ -0,0 +1,40 @@
+version: "3.9"
+services:
+  postgres:
+    container_name: postgres_container
+    image: postgres:16.1-alpine3.18
+    environment:
+      POSTGRES_DB: "umschooldb"
+      POSTGRES_USER: "umschooluser"
+      POSTGRES_PASSWORD: "umschoolpswd"
+      PGDATA: "/var/lib/postgresql/data/pgdata"
+    volumes:
+      - umschooldb-data:/var/lib/postgresql/data
+    ports:
+      - "6432:5432"
+    restart: unless-stopped
+    networks:
+      - postgres
+
+  pgadmin:
+    container_name: pgadmin_container
+    image: dpage/pgadmin4:8.13
+    environment:
+      PGADMIN_DEFAULT_EMAIL: "[email protected]"
+      PGADMIN_DEFAULT_PASSWORD: "pgadminpswd"
+      PGADMIN_CONFIG_SERVER_MODE: "False" # To allow running in desktop mode
+    volumes:
+      - pgadmin-data:/var/lib/pgadmin
+    ports:
+      - "5050:80"
+    restart: unless-stopped
+    networks:
+      - postgres
+
+volumes:
+  umschooldb-data:
+  pgadmin-data:
+
+networks:
+  postgres:
+    driver: bridge

example/main.py

@@ -3,15 +3,15 @@
 
 import telebot
 
-from db_adapter import create_tables_from_db
+from db_adapter import create_tables_in_db
 from handlers import register_handlers
 from init_bot import create_bot
 
 
 telebot.logger.setLevel(logging.DEBUG)  # Outputs messages to console.
 
 if __name__ == "__main__":
-    create_tables_from_db()
+    create_tables_in_db()
     bot = create_bot(os.getenv('TG_TOKEN'))
     register_handlers(bot)
     bot.add_custom_filter(telebot.custom_filters.StateFilter(bot))

example/requirements.txt

@@ -0,0 +1,3 @@
+SQLAlchemy==2.0
+pyTelegramBotAPI==4.24.0
+psycopg2-binary==2.9.10