database.Database.update_mailing_attaches() - Code Metrics - Inspection of "Merge branch 'release/6.4.0'" - dadyarri/ralph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 1663f9...461fea )

by Daniil

created 2020-03-22 08:15 UTC

database.Database.update_mailing_attaches() A

↳ Parent: database

Complexity

Conditions

Size

Total Lines	6
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	4
dl	0
loc	6
rs	10
c	0
b	0
f	0
cc	1
nop	3

from typing import List
from typing import NoReturn
from typing import Tuple
from typing import Union

from database.base import Base

"""
Модуль, содержащий класс с методами для работы с БД, выполняющих конечную цель
"""


class Database(Base):
    """
        Класс для методов работы с БД, выполняющих конечную цель
    """

    def get_active_students_ids(self) -> list:
        ids = self.query(
            f"SELECT user_id FROM users_info WHERE academic_status > 0 ORDER BY user_id"
        )
        vk_ids = [str(self.get_vk_id(_id)) for (_id,) in ids]
        return vk_ids

    def get_last_names_letters(self) -> List[str]:
        """
        Получает из базы данных все уникальные первые буквы фамилий
        """
        letters = self.query(
            "SELECT DISTINCT substring(second_name from  '^.') FROM users_info "
            "ORDER BY substring(second_name from  '^.')"
        )
        return [letter for (letter,) in letters]

    def get_list_of_names(self, letter: str) -> List[Tuple]:
        """
        Получает из базы данных все фамилии, начинающиеся на букву
        """
        names = self.query(
            f"SELECT user_id, first_name, second_name FROM users_info "
            f"WHERE substring(second_name from '^.') = '{letter}' "
            f"AND academic_status > 0 ORDER BY user_id"
        )
        return names

    def get_vk_id(self, _id: Union[str, int]) -> int:
        """
        Получает из базы данных идентификатор ВКонтакте по идентификатору студента
        """
        vk_id = self.query(f"SELECT vk_id from users WHERE id={_id}")[0][0]
        return vk_id

    def get_user_id(self, vk_id: int) -> int:
        """
        Получает из базы данных идентификатор студента по идентификатору ВКонтакте
        """
        user_id = self.query(f"SELECT id from users WHERE vk_id={vk_id}")[0][0]
        return user_id

    def get_user_name(self, _id: int) -> str:
        """
        Получает из базы данных имя по идентификатор студента
        """
        name = self.query(f"SELECT first_name FROM users_info WHERE user_id={_id}")[0][
            0
        ]
        return name

    def get_mailings_list(self) -> List[Tuple]:
        """
        Получает из базы данных весь список доступных рассылок
        """
        mailings = self.query(
            "SELECT mailing_id, mailing_name, mailing_slug from mailings"
        )
        return mailings

    def get_subscription_status(self, slug: str, user_id: int) -> int:
        """
        Получает статус подписки пользователя на рассылку
        """
        status = self.query(
            f"SELECT {slug} FROM vk_subscriptions WHERE user_id={user_id}"
        )[0][0]
        return status

    def is_user_exist(self, user_id: int) -> bool:
        """
        Возвращает информацию о том, существует ли пользователь в базе данных
        """
        user = self.query(f"SELECT id FROM users WHERE vk_id={user_id}")
        return bool(user)

    def is_session_exist(self, user_id: int) -> bool:
        """
        Проверяет существование сессии текущего пользователя
        """
        user = self.query(f"SELECT id FROM sessions WHERE vk_id={user_id}")
        return bool(user)

    def create_user(self, user_id: int) -> NoReturn:
        """
        Добавляет нового пользователя в таблицы информации и рассылок
        """
        self.query(f"INSERT INTO users (vk_id) VALUES ({user_id})")
        self.query(f"INSERT INTO vk_subscriptions DEFAULT VALUES")

    def create_session(self, user_id: int) -> NoReturn:
        """
        Создает новую сессию для пользователя
        """
        _id = self.query(f"SELECT id FROM users WHERE vk_id={user_id}")[0][0]
        self.query(
            f"INSERT INTO sessions (id, vk_id, state) VALUES ({_id}, {user_id}, 'main')"
        )

    def get_session_state(self, user_id: int) -> str:
        """
        Получает текущий статус бота из сессии
        """
        st = self.query(f"SELECT state FROM sessions WHERE vk_id={user_id}")
        return st[0][0]

    def get_session_id(self, user_id: int) -> int:
        """
        Получает идентификатор сессии
        """
        s_id = self.query(f"SELECT id FROM sessions WHERE vk_id={user_id}")[0][0]
        return s_id

    def update_session_state(self, user_id: int, state: str) -> NoReturn:
        """
        Изменяет текущий статус бота из сессии
        """
        self.query(f"UPDATE sessions SET state='{state}' WHERE vk_id={user_id}")

    def call_session_exist(self, user_id: int) -> bool:
        """
        Проверяет существование сессии призыва
        """
        s_id = self.get_session_id(user_id)
        call_exist = self.query(f"SELECT session_id FROM calls WHERE session_id={s_id}")
        texts_exist = self.query(
            f"SELECT session_id FROM texts WHERE session_id" f"={s_id}"
        )
        return bool(call_exist and texts_exist)

    def create_call_session(self, user_id: int) -> NoReturn:
        """
        Создает новую сессию призыва
        """
        s_id = self.get_session_id(user_id)
        self.query(f"INSERT INTO calls (session_id) VALUES ({s_id})")
        self.query(f"INSERT INTO texts (session_id) VALUES ({s_id})")

    def get_call_message(self, user_id: int) -> str:
        """
        Получает текст призыва
        """
        s_id = self.get_session_id(user_id)
        return self.query(f"SELECT text FROM texts WHERE session_id={s_id}")[0][0]

    def update_call_message(self, user_id: int, message: str) -> NoReturn:
        """
        Обновляет текст призыва
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE texts SET text='{message}' WHERE session_id={s_id}")

    def get_call_ids(self, user_id: int) -> str:
        """
        Получить список идентификаторов для призыва
        """
        s_id = self.get_session_id(user_id)
        ids = self.query(f"SELECT ids FROM calls WHERE session_id={s_id}")[0][0]
        return ids

    def update_call_ids(self, user_id: int, ids: str) -> NoReturn:
        """
        Перезаписать список идентификаторов для призыва
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE calls SET ids='{ids}' WHERE session_id={s_id}")

    def append_to_call_ids(self, user_id: int, _id: int) -> NoReturn:
        """
        Добавить идентификатор к списку для призыва
        """
        ids = self.get_call_ids(user_id)
        if ids is None:
            ids = ""
        ids += f"{_id},"
        self.update_call_ids(user_id, ids)

    def empty_call_storage(self, user_id: int) -> NoReturn:
        """
        Очистить хранилище призыва (текст призыва и список идентификатора)
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE calls SET ids=NULL WHERE session_id={s_id}")
        self.query(f"UPDATE texts SET text=NULL WHERE session_id={s_id}")
        self.query(f"UPDATE texts SET attach=NULL WHERE session_id={s_id}")

    def empty_mailing_storage(self, user_id: int) -> NoReturn:
        """
        Очистить хранилище рассылок (выбранную рассылку и текст рассылки)
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE mailing_mgmt SET mailing=NULL WHERE session_id={s_id}")
        self.query(f"UPDATE mailing_mgmt SET m_text=NULL WHERE session_id={s_id}")
        self.query(f"UPDATE mailing_mgmt SET m_attach=NULL WHERE session_id={s_id}")

    def get_mailing_message(self, user_id: int) -> str:
        """
        Получить текст рассылки
        """
        s_id = self.get_session_id(user_id)
        return self.query(f"SELECT m_text FROM mailing_mgmt WHERE session_id={s_id}")[
            0
        ][0]

    def update_mailing_message(self, user_id: int, message: str) -> NoReturn:
        """
        Перезаписывает текст рассылки
        """
        s_id = self.get_session_id(user_id)
        self.query(
            f"UPDATE mailing_mgmt SET m_text='{message}' WHERE session_id={s_id}"
        )

    def get_conversation(self, user_id: int) -> int:
        """
        Получает активную беседу
        """
        s_id = self.get_session_id(user_id)
        return self.query(f"SELECT conversation FROM sessions WHERE id={s_id}")[0][0]

    def update_conversation(self, user_id: int, conv_id: int) -> NoReturn:
        """
        Изменяет активную беседу
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE sessions SET conversation={conv_id} WHERE id" f"={s_id}")

    def mailing_session_exist(self, user_id: int) -> bool:
        """
        Проверяет наличие сессии рассылки
        """
        s_id = self.get_session_id(user_id)
        return self.query(
            f"SELECT session_id FROM mailing_mgmt WHERE session_id={s_id}"
        )

    def create_mailing_session(self, user_id: int) -> NoReturn:
        """
        Создает сессию рассылки
        """
        s_id = self.get_session_id(user_id)
        self.query(f"INSERT INTO mailing_mgmt (session_id) VALUES ({s_id})")

    def update_mailing_session(self, user_id: int, m_slug: str) -> NoReturn:
        """
        Изменяет выбранную рассылку
        """
        s_id = self.get_session_id(user_id)
        self.query(
            f"UPDATE mailing_mgmt SET mailing='{m_slug}' WHERE session_id={s_id}"
        )

    def get_mailing_session(self, user_id: int) -> str:
        """
        Получает выбранную рассылку
        """
        s_id = self.get_session_id(user_id)
        return self.query(f"SELECT mailing FROM mailing_mgmt WHERE session_id={s_id}")[
            0
        ][0]

    def fetch_subcribers(self, slug: str) -> str:
        """
        Собирает подписчиков рассылки
        """
        user_ids = self.query(f"SELECT user_id FROM vk_subscriptions WHERE {slug}=1")
        user_ids = [_id for (_id,) in user_ids]
        ids = []
        for i, _id in enumerate(user_ids):
            ids.append(self.query(f"SELECT vk_id FROM users WHERE id={_id}")[0][0])
        ids = ",".join(map(str, ids))
        return ids

    def get_names_using_status(self, user_id: int) -> bool:
        """
        Получает статус использования имён
        """
        s_id = self.get_session_id(user_id)
        names_using = self.query(f"SELECT names_using FROM sessions WHERE id={s_id}")[
            0
        ][0]
        return bool(names_using)

    def update_names_using_status(self, user_id: int, value: int):
        """
        Изменяет статус использования имён
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE sessions SET names_using={value} WHERE id={s_id}")

    def get_users_names(self, ids: list) -> List[str]:
        """
        Получает список имён по списку идентификаторов ВКонтакте
        """
        ids = ", ".join(ids)
        query = self.query(
            f"SELECT first_name FROM users INNER JOIN users_info ON users.id = "
            f"users_info.user_id WHERE vk_id in ({ids}) ORDER BY "
            f"position(vk_id::text in '{ids}')"
        )
        names = [i for (i,) in query]
        return names

    def get_call_attaches(self, user_id: int):
        """Получает список вложений для сообщения с призывом
        """
        s_id = self.get_session_id(user_id)
        query = self.query(f"SELECT attach FROM texts WHERE session_id={s_id}")
        return query[0][0]

    def update_call_attaches(self, user_id: int, attach: str):
        """Обновляет список вложений для сообщения с призывом
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE texts SET attach='{attach}' WHERE session_id={s_id}")

    def get_mailing_attaches(self, user_id: int):
        """Получает список вложений для сообщения рассылки
        """
        s_id = self.get_session_id(user_id)
        query = self.query(f"SELECT m_attach FROM mailing_mgmt WHERE session_id={s_id}")
        return query[0][0]

    def update_mailing_attaches(self, user_id: int, attach: str):
        """Обновляет список вложений для сообщения рассылки
        """
        s_id = self.get_session_id(user_id)
        self.query(
            f"UPDATE mailing_mgmt SET m_attach='{attach}' WHERE session_id={s_id}"
        )


1			from typing import List
2			from typing import NoReturn
3			from typing import Tuple
4			from typing import Union
5
6			from database.base import Base
7
8			"""
9			Модуль, содержащий класс с методами для работы с БД, выполняющих конечную цель
10			"""
11
12
13			class Database(Base):
14			"""
15			Класс для методов работы с БД, выполняющих конечную цель
16			"""
17
18			def get_active_students_ids(self) -> list:
19			ids = self.query(
20			f"SELECT user_id FROM users_info WHERE academic_status > 0 ORDER BY user_id"
21			)
22			vk_ids = [str(self.get_vk_id(_id)) for (_id,) in ids]
23			return vk_ids
24
25			def get_last_names_letters(self) -> List[str]:
26			"""
27			Получает из базы данных все уникальные первые буквы фамилий
28			"""
29			letters = self.query(
30			"SELECT DISTINCT substring(second_name from '^.') FROM users_info "
31			"ORDER BY substring(second_name from '^.')"
32			)
33			return [letter for (letter,) in letters]
34
35			def get_list_of_names(self, letter: str) -> List[Tuple]:
36			"""
37			Получает из базы данных все фамилии, начинающиеся на букву
38			"""
39			names = self.query(
40			f"SELECT user_id, first_name, second_name FROM users_info "
41			f"WHERE substring(second_name from '^.') = '{letter}' "
42			f"AND academic_status > 0 ORDER BY user_id"
43			)
44			return names
45
46			def get_vk_id(self, _id: Union[str, int]) -> int:
47			"""
48			Получает из базы данных идентификатор ВКонтакте по идентификатору студента
49			"""
50			vk_id = self.query(f"SELECT vk_id from users WHERE id={_id}")[0][0]
51			return vk_id
52
53			def get_user_id(self, vk_id: int) -> int:
54			"""
55			Получает из базы данных идентификатор студента по идентификатору ВКонтакте
56			"""
57			user_id = self.query(f"SELECT id from users WHERE vk_id={vk_id}")[0][0]
58			return user_id
59
60			def get_user_name(self, _id: int) -> str:
61			"""
62			Получает из базы данных имя по идентификатор студента
63			"""
64			name = self.query(f"SELECT first_name FROM users_info WHERE user_id={_id}")[0][
65			0
66			]
67			return name
68
69			def get_mailings_list(self) -> List[Tuple]:
70			"""
71			Получает из базы данных весь список доступных рассылок
72			"""
73			mailings = self.query(
74			"SELECT mailing_id, mailing_name, mailing_slug from mailings"
75			)
76			return mailings
77
78			def get_subscription_status(self, slug: str, user_id: int) -> int:
79			"""
80			Получает статус подписки пользователя на рассылку
81			"""
82			status = self.query(
83			f"SELECT {slug} FROM vk_subscriptions WHERE user_id={user_id}"
84			)[0][0]
85			return status
86
87			def is_user_exist(self, user_id: int) -> bool:
88			"""
89			Возвращает информацию о том, существует ли пользователь в базе данных
90			"""
91			user = self.query(f"SELECT id FROM users WHERE vk_id={user_id}")
92			return bool(user)
93
94			def is_session_exist(self, user_id: int) -> bool:
95			"""
96			Проверяет существование сессии текущего пользователя
97			"""
98			user = self.query(f"SELECT id FROM sessions WHERE vk_id={user_id}")
99			return bool(user)
100
101			def create_user(self, user_id: int) -> NoReturn:
102			"""
103			Добавляет нового пользователя в таблицы информации и рассылок
104			"""
105			self.query(f"INSERT INTO users (vk_id) VALUES ({user_id})")
106			self.query(f"INSERT INTO vk_subscriptions DEFAULT VALUES")
107
108			def create_session(self, user_id: int) -> NoReturn:
109			"""
110			Создает новую сессию для пользователя
111			"""
112			_id = self.query(f"SELECT id FROM users WHERE vk_id={user_id}")[0][0]
113			self.query(
114			f"INSERT INTO sessions (id, vk_id, state) VALUES ({_id}, {user_id}, 'main')"
115			)
116
117			def get_session_state(self, user_id: int) -> str:
118			"""
119			Получает текущий статус бота из сессии
120			"""
121			st = self.query(f"SELECT state FROM sessions WHERE vk_id={user_id}")
122			return st[0][0]
123
124			def get_session_id(self, user_id: int) -> int:
125			"""
126			Получает идентификатор сессии
127			"""
128			s_id = self.query(f"SELECT id FROM sessions WHERE vk_id={user_id}")[0][0]
129			return s_id
130
131			def update_session_state(self, user_id: int, state: str) -> NoReturn:
132			"""
133			Изменяет текущий статус бота из сессии
134			"""
135			self.query(f"UPDATE sessions SET state='{state}' WHERE vk_id={user_id}")
136
137			def call_session_exist(self, user_id: int) -> bool:
138			"""
139			Проверяет существование сессии призыва
140			"""
141			s_id = self.get_session_id(user_id)
142			call_exist = self.query(f"SELECT session_id FROM calls WHERE session_id={s_id}")
143			texts_exist = self.query(
144			f"SELECT session_id FROM texts WHERE session_id" f"={s_id}"
145			)
146			return bool(call_exist and texts_exist)
147
148			def create_call_session(self, user_id: int) -> NoReturn:
149			"""
150			Создает новую сессию призыва
151			"""
152			s_id = self.get_session_id(user_id)
153			self.query(f"INSERT INTO calls (session_id) VALUES ({s_id})")
154			self.query(f"INSERT INTO texts (session_id) VALUES ({s_id})")
155
156			def get_call_message(self, user_id: int) -> str:
157			"""
158			Получает текст призыва
159			"""
160			s_id = self.get_session_id(user_id)
161			return self.query(f"SELECT text FROM texts WHERE session_id={s_id}")[0][0]
162
163			def update_call_message(self, user_id: int, message: str) -> NoReturn:
164			"""
165			Обновляет текст призыва
166			"""
167			s_id = self.get_session_id(user_id)
168			self.query(f"UPDATE texts SET text='{message}' WHERE session_id={s_id}")
169
170			def get_call_ids(self, user_id: int) -> str:
171			"""
172			Получить список идентификаторов для призыва
173			"""
174			s_id = self.get_session_id(user_id)
175			ids = self.query(f"SELECT ids FROM calls WHERE session_id={s_id}")[0][0]
176			return ids
177
178			def update_call_ids(self, user_id: int, ids: str) -> NoReturn:
179			"""
180			Перезаписать список идентификаторов для призыва
181			"""
182			s_id = self.get_session_id(user_id)
183			self.query(f"UPDATE calls SET ids='{ids}' WHERE session_id={s_id}")
184
185			def append_to_call_ids(self, user_id: int, _id: int) -> NoReturn:
186			"""
187			Добавить идентификатор к списку для призыва
188			"""
189			ids = self.get_call_ids(user_id)
190			if ids is None:
191			ids = ""
192			ids += f"{_id},"
193			self.update_call_ids(user_id, ids)
194
195			def empty_call_storage(self, user_id: int) -> NoReturn:
196			"""
197			Очистить хранилище призыва (текст призыва и список идентификатора)
198			"""
199			s_id = self.get_session_id(user_id)
200			self.query(f"UPDATE calls SET ids=NULL WHERE session_id={s_id}")
201			self.query(f"UPDATE texts SET text=NULL WHERE session_id={s_id}")
202			self.query(f"UPDATE texts SET attach=NULL WHERE session_id={s_id}")
203
204			def empty_mailing_storage(self, user_id: int) -> NoReturn:
205			"""
206			Очистить хранилище рассылок (выбранную рассылку и текст рассылки)
207			"""
208			s_id = self.get_session_id(user_id)
209			self.query(f"UPDATE mailing_mgmt SET mailing=NULL WHERE session_id={s_id}")
210			self.query(f"UPDATE mailing_mgmt SET m_text=NULL WHERE session_id={s_id}")
211			self.query(f"UPDATE mailing_mgmt SET m_attach=NULL WHERE session_id={s_id}")
212
213			def get_mailing_message(self, user_id: int) -> str:
214			"""
215			Получить текст рассылки
216			"""
217			s_id = self.get_session_id(user_id)
218			return self.query(f"SELECT m_text FROM mailing_mgmt WHERE session_id={s_id}")[
219			0
220			][0]
221
222			def update_mailing_message(self, user_id: int, message: str) -> NoReturn:
223			"""
224			Перезаписывает текст рассылки
225			"""
226			s_id = self.get_session_id(user_id)
227			self.query(
228			f"UPDATE mailing_mgmt SET m_text='{message}' WHERE session_id={s_id}"
229			)
230
231			def get_conversation(self, user_id: int) -> int:
232			"""
233			Получает активную беседу
234			"""
235			s_id = self.get_session_id(user_id)
236			return self.query(f"SELECT conversation FROM sessions WHERE id={s_id}")[0][0]
237
238			def update_conversation(self, user_id: int, conv_id: int) -> NoReturn:
239			"""
240			Изменяет активную беседу
241			"""
242			s_id = self.get_session_id(user_id)
243			self.query(f"UPDATE sessions SET conversation={conv_id} WHERE id" f"={s_id}")
244
245			def mailing_session_exist(self, user_id: int) -> bool:
246			"""
247			Проверяет наличие сессии рассылки
248			"""
249			s_id = self.get_session_id(user_id)
250			return self.query(
251			f"SELECT session_id FROM mailing_mgmt WHERE session_id={s_id}"
252			)
253
254			def create_mailing_session(self, user_id: int) -> NoReturn:
255			"""
256			Создает сессию рассылки
257			"""
258			s_id = self.get_session_id(user_id)
259			self.query(f"INSERT INTO mailing_mgmt (session_id) VALUES ({s_id})")
260
261			def update_mailing_session(self, user_id: int, m_slug: str) -> NoReturn:
262			"""
263			Изменяет выбранную рассылку
264			"""
265			s_id = self.get_session_id(user_id)
266			self.query(
267			f"UPDATE mailing_mgmt SET mailing='{m_slug}' WHERE session_id={s_id}"
268			)
269
270			def get_mailing_session(self, user_id: int) -> str:
271			"""
272			Получает выбранную рассылку
273			"""
274			s_id = self.get_session_id(user_id)
275			return self.query(f"SELECT mailing FROM mailing_mgmt WHERE session_id={s_id}")[
276			0
277			][0]
278
279			def fetch_subcribers(self, slug: str) -> str:
280			"""
281			Собирает подписчиков рассылки
282			"""
283			user_ids = self.query(f"SELECT user_id FROM vk_subscriptions WHERE {slug}=1")
284			user_ids = [_id for (_id,) in user_ids]
285			ids = []
286			for i, _id in enumerate(user_ids):
287			ids.append(self.query(f"SELECT vk_id FROM users WHERE id={_id}")[0][0])
288			ids = ",".join(map(str, ids))
289			return ids
290
291			def get_names_using_status(self, user_id: int) -> bool:
292			"""
293			Получает статус использования имён
294			"""
295			s_id = self.get_session_id(user_id)
296			names_using = self.query(f"SELECT names_using FROM sessions WHERE id={s_id}")[
297			0
298			][0]
299			return bool(names_using)
300
301			def update_names_using_status(self, user_id: int, value: int):
302			"""
303			Изменяет статус использования имён
304			"""
305			s_id = self.get_session_id(user_id)
306			self.query(f"UPDATE sessions SET names_using={value} WHERE id={s_id}")
307
308			def get_users_names(self, ids: list) -> List[str]:
309			"""
310			Получает список имён по списку идентификаторов ВКонтакте
311			"""
312			ids = ", ".join(ids)
313			query = self.query(
314			f"SELECT first_name FROM users INNER JOIN users_info ON users.id = "
315			f"users_info.user_id WHERE vk_id in ({ids}) ORDER BY "
316			f"position(vk_id::text in '{ids}')"
317			)
318			names = [i for (i,) in query]
319			return names
320
321			def get_call_attaches(self, user_id: int):
322			"""Получает список вложений для сообщения с призывом
323			"""
324			s_id = self.get_session_id(user_id)
325			query = self.query(f"SELECT attach FROM texts WHERE session_id={s_id}")
326			return query[0][0]
327
328			def update_call_attaches(self, user_id: int, attach: str):
329			"""Обновляет список вложений для сообщения с призывом
330			"""
331			s_id = self.get_session_id(user_id)
332			self.query(f"UPDATE texts SET attach='{attach}' WHERE session_id={s_id}")
333
334			def get_mailing_attaches(self, user_id: int):
335			"""Получает список вложений для сообщения рассылки
336			"""
337			s_id = self.get_session_id(user_id)
338			query = self.query(f"SELECT m_attach FROM mailing_mgmt WHERE session_id={s_id}")
339			return query[0][0]
340
341			def update_mailing_attaches(self, user_id: int, attach: str):
342			"""Обновляет список вложений для сообщения рассылки
343			"""
344			s_id = self.get_session_id(user_id)
345			self.query(
346			f"UPDATE mailing_mgmt SET m_attach='{attach}' WHERE session_id={s_id}"
347			)
348

dadyarri / ralph

Push — master ( 1663f9...461fea )

database.Database.update_mailing_attaches() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like