sm/flask_driver/sql_worker.py - master - MaxIs/savemoney

sm/flask_driver/sql_worker.py

55 строк2 KB

zalma2006

добавлен flask для работы с selenium в отдельном сервисе, добавлены точки API, добавлена загрузка отчётов из источника, подробнее в HISTORY.md

15 мар 2026, 19:36

15 мар 2026, 19:36ae3ea09

import sqlite3
from logging import getLogger
import pandas as pd


class SQLWorker:
    """Класс для работы с БД, только чтение"""
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(SQLWorker, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        if not hasattr(SQLWorker, "inicialized"):
            self.con_string = None
            self.logger = getLogger(self.__class__.__name__)
            self.inicialized = True
            self.metadata_name = 'downloads/metaData'
            self.dest_file_name = 'downloads/destinationFileURI'

    def set_con_string(self, con_string: str):
        if not isinstance(con_string, str):
            raise TypeError("con_string must be a string")
        if len(con_string) == 0:
            raise ValueError(f"con_string {con_string} is empty!")

        self.con_string = con_string

    def exec_query(self, query: str):
        if self.con_string is None:
            raise TypeError("first use self.set_con_string")
        if not isinstance(query, str):
            raise TypeError("query must be a string")
        query = query.strip()
        if len(query) == 0:
            raise ValueError("query must not be empty")

        if query.lower().startswith("select"):
            with sqlite3.connect(self.con_string) as conn:
                res = conn.cursor().execute(query)
                rows = res.fetchall()
                columns = [d[0] for d in res.description]
            return rows, columns
        else:
            return None, None

    def exec_query_from_table(self, query: str):
        """Выполнить запрос select и получить данные в виде таблицы"""
        rows, columns = self.exec_query(query=query)
        data = pd.DataFrame(data=rows, columns=columns)
        return data

sql_worker = SQLWorker()