database.Database.get_users_names() - Code Metrics - Inspection of "Remove Amazon scripts." - dadyarri/ralph - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( aaab20...67ce97 )

by Daniil

created 2020-03-10 16:47 UTC

database.Database.get_users_names() A

↳ Parent: database

Complexity

Conditions

Size

Total Lines	12
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	6
dl	0
loc	12
rs	10
c	0
b	0
f	0
cc	1
nop	2

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}")

    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}")

    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


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
203			def empty_mailing_storage(self, user_id: int) -> NoReturn:
204			"""
205			Очистить хранилище рассылок (выбранную рассылку и текст рассылки)
206			"""
207			s_id = self.get_session_id(user_id)
208			self.query(f"UPDATE mailing_mgmt SET mailing=NULL WHERE session_id={s_id}")
209			self.query(f"UPDATE mailing_mgmt SET m_text=NULL WHERE session_id={s_id}")
210
211			def get_mailing_message(self, user_id: int) -> str:
212			"""
213			Получить текст рассылки
214			"""
215			s_id = self.get_session_id(user_id)
216			return self.query(f"SELECT m_text FROM mailing_mgmt WHERE session_id={s_id}")[
217			0
218			][0]
219
220			def update_mailing_message(self, user_id: int, message: str) -> NoReturn:
221			"""
222			Перезаписывает текст рассылки
223			"""
224			s_id = self.get_session_id(user_id)
225			self.query(
226			f"UPDATE mailing_mgmt SET m_text='{message}' WHERE session_id={s_id}"
227			)
228
229			def get_conversation(self, user_id: int) -> int:
230			"""
231			Получает активную беседу
232			"""
233			s_id = self.get_session_id(user_id)
234			return self.query(f"SELECT conversation FROM sessions WHERE id={s_id}")[0][0]
235
236			def update_conversation(self, user_id: int, conv_id: int) -> NoReturn:
237			"""
238			Изменяет активную беседу
239			"""
240			s_id = self.get_session_id(user_id)
241			self.query(f"UPDATE sessions SET conversation={conv_id} WHERE id" f"={s_id}")
242
243			def mailing_session_exist(self, user_id: int) -> bool:
244			"""
245			Проверяет наличие сессии рассылки
246			"""
247			s_id = self.get_session_id(user_id)
248			return self.query(
249			f"SELECT session_id FROM mailing_mgmt WHERE session_id={s_id}"
250			)
251
252			def create_mailing_session(self, user_id: int) -> NoReturn:
253			"""
254			Создает сессию рассылки
255			"""
256			s_id = self.get_session_id(user_id)
257			self.query(f"INSERT INTO mailing_mgmt (session_id) VALUES ({s_id})")
258
259			def update_mailing_session(self, user_id: int, m_slug: str) -> NoReturn:
260			"""
261			Изменяет выбранную рассылку
262			"""
263			s_id = self.get_session_id(user_id)
264			self.query(
265			f"UPDATE mailing_mgmt SET mailing='{m_slug}' WHERE session_id={s_id}"
266			)
267
268			def get_mailing_session(self, user_id: int) -> str:
269			"""
270			Получает выбранную рассылку
271			"""
272			s_id = self.get_session_id(user_id)
273			return self.query(f"SELECT mailing FROM mailing_mgmt WHERE session_id={s_id}")[
274			0
275			][0]
276
277			def fetch_subcribers(self, slug: str) -> str:
278			"""
279			Собирает подписчиков рассылки
280			"""
281			user_ids = self.query(f"SELECT user_id FROM vk_subscriptions WHERE {slug}=1")
282			user_ids = [_id for (_id,) in user_ids]
283			ids = []
284			for i, _id in enumerate(user_ids):
285			ids.append(self.query(f"SELECT vk_id FROM users WHERE id={_id}")[0][0])
286			ids = ",".join(map(str, ids))
287			return ids
288
289			def get_names_using_status(self, user_id: int) -> bool:
290			"""
291			Получает статус использования имён
292			"""
293			s_id = self.get_session_id(user_id)
294			names_using = self.query(f"SELECT names_using FROM sessions WHERE id={s_id}")[
295			0
296			][0]
297			return bool(names_using)
298
299			def update_names_using_status(self, user_id: int, value: int):
300			"""
301			Изменяет статус использования имён
302			"""
303			s_id = self.get_session_id(user_id)
304			self.query(f"UPDATE sessions SET names_using={value} WHERE id={s_id}")
305
306			def get_users_names(self, ids: list) -> List[str]:
307			"""
308			Получает список имён по списку идентификаторов ВКонтакте
309			"""
310			ids = ", ".join(ids)
311			query = self.query(
312			f"SELECT first_name FROM users INNER JOIN users_info ON users.id = "
313			f"users_info.user_id WHERE vk_id in ({ids}) ORDER BY "
314			f"position(vk_id::text in '{ids}')"
315			)
316			names = [i for (i,) in query]
317			return names
318

dadyarri / ralph

Push — master ( aaab20...67ce97 )

database.Database.get_users_names() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like