eNG97jux

1import os
2import sqlite3
3
4
5class DBSerializerInterface:
6    def __init__(self):
7        self.directory = None
8        self.db_name = None
9        self.__db_connection = None
10        self.__cursor = None
11
12    def run_connection(self, db_dir='res/database/', db_name='replays.db'):
13        self.directory = db_dir
14        self.db_name = db_name
15        self.__db_connection = sqlite3.connect(database=self.directory + self.db_name)
16        self.__cursor = self.__db_connection.cursor()
17
18    # def .........
19    def calculate_rows(self, table_name: str, column_name: str) -> int:
20        script = f'SELECT COUNT({column_name}) from {table_name}'
21        result = self.execute(script)
22        if not result:
23            return 0
24        else:
25            return result[0][0]
26
27    def execute(self, script: str):
28        try:
29            self.__cursor.executescript(script)
30            result = self.__cursor.fetchall()
31            self.__db_connection.commit()
32            return result
33        except Exception as err:
34            print(f'{err} in {__class__.__name__} -> execute')
35
36    def close_connection(self):
37        self.__db_connection.commit()
38        self.__db_connection.close()
39
40
41class DBSerializer(DBSerializerInterface):
42    def obtain_from_db(self, table_name: str, *columns, order_by_column: str = None):
43        try:
44            script = 'SELECT '
45            if columns is None:
46                script += f'* FROM {table_name}'
47            else:
48                # для того, чтобы запятая не добавилась в конце всех столбцов
49                # SELECT col1, col2, col3 FROM table_name
50                for column in columns[:len(columns) - 1]:
51                    script += f'{column}, '
52                script += f'{columns[-1]} FROM {table_name}'
53
54            if order_by_column is not None:
55                script += f' ORDER BY {order_by_column}'
56
57            return {
58                'result': self.execute(script),
59                'columns_count': len(columns),
60                'rows': self.calculate_rows(table_name, columns[0])
61            }
62        except Exception as e:
63            print(f'{e} in DBSerializer -> obtain')
64
65    # def ...
66
67
68# Все константы, необходимые для парметров запроса
69# к серверу и вывода общей инфы реплея
70class ParametersConsts:
71    def __init__(self):
72        self.DIR = os.path.join(os.path.dirname(__file__), '../')
73        self.NAME = 'params.db'
74        self.CREATE_TABLE_SCRIPT = '''
75            CREATE TABLE IF NOT EXISTS spawn_point(
76                respawn INTEGER
77            );
78            insert into spawn_point(respawn) values (1), (2);
79            '''
80        self.ROW_COUNTER_VALUE = {
81            'tanks': 'id',
82            'levels': 'level',
83            'nations': 'nation',
84            'maps': 'map',
85            'medals': 'medals',
86            'battle_type': 'battle_type',
87            'spawn_point': 'respawn',
88            'ace': 'ace'
89        }
90
91
92class Parameters(DBSerializer):
93    def __init__(self):
94        super().__init__()
95        self.consts = ParametersConsts()
96
97    def run_connection(self, db_dir=None, db_name=None):
98        self.directory = self.consts.DIR
99        self.db_name = self.consts.NAME
100        self.__db_connection = sqlite3.connect(database=self.directory + self.db_name)
101        self.__cursor = self.__db_connection.cursor()
102
103    def execute(self, script: str):
104        try:
105            self.__cursor.executescript(script)
106            result = self.__cursor.fetchall()
107            self.__db_connection.commit()
108            return result
109        except Exception as err:
110            print(f'{err}. Error in {__class__.__name__} -> {self.execute.__name__}')
111
112    def close_connection(self):
113        self.__db_connection.commit()
114        self.__db_connection.close()
115
116    def create_parameters_db(self):
117        self.execute(self.consts.CREATE_TABLE_SCRIPT)
118
119    # .......
120
121    def row_count(self, table_name: str) -> int:
122        script = f'SELECT COUNT({self.consts.ROW_COUNTER_VALUE[table_name]}) from {table_name}'
123        res = self.execute(script)
124        if not res:
125            return 0
126        else:
127            return res[0][0]
128
129    def obtain_from_db(self, table_name: str, *columns, order_by_column: str = None):
130        try:
131            script = 'SELECT '
132            if not columns:
133                script += f'* FROM {table_name}'
134            else:
135                # для того, чтобы запятая не добавилась в конце всех столбцов
136                # SELECT col1, col2, col3 FROM table_name
137                for column in columns[:len(columns) - 1]:
138                    script += f'{column}, '
139                script += f'{columns[-1]} FROM {table_name}'
140
141            if order_by_column is not None:
142                script += f' ORDER BY {order_by_column}'
143
144            return {
145                'result': self.execute(script),
146                'columns_count': len(columns),
147                'rows': self.row_count(table_name),
148                'script': script
149            }
150        except Exception as e:
151            print(f'{e} in {__class__.__name__} -> obtain')
152
153
154param_s = Parameters()
155param_s.run_connection()  # устанавливает соединение с базой данных
156param_s.create_parameters_db()  # создает таблицы в бд
157param_s.execute('INSERT INTO spawn_point VALUES (1), (2)')  # Вставка в таблицу значений 1 и2 2
158print(param_s.obtain_from_db('spawn_point'))  # скрипт который должен вернуть, что надо 
159print(param_s.execute('SELECT respawn FROM spawn_point'))  #  но не работает походу только это
160param_s.close_connection()  # все, конец соединению