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

Passed

Push — master ( 0e504e...758c62 )

by Daniil

created 2020-03-31 09:52 UTC

database.Database.get_summ_of_donate() A

↳ Parent: database

Complexity

Conditions

Size

Total Lines	11
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	3
dl	0
loc	11
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 update_subscribe_state(self, slug: str, u_id: int, state: int):
        """Обновляет статус подписки на рассылку
        """
        self.query(
            f"UPDATE vk_subscriptions SET '{slug}'={state} WHERE " f"user_id={u_id}"
        )

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

    def get_list_of_finances_categories(self) -> List[Tuple]:
        """Получает список доступных категорий расходов
        """
        query = self.query("SELECT name, slug FROM finances_categories")
        return query

    def add_expences_category(self, name: str, slug: str, s: int):
        """Созадет новую категорию расходов
        """
        query = self.query(
            f"INSERT INTO finances_categories (name, slug, sum) VALUES ('{name}', "
            f"'{slug}', {s}) RETURNING id"
        )

        return query[0][0]

    def get_active_expenses_category(self, user_id: int):
        """Получает текущую выбранную пользователем статью расходов
        """
        s_id = self.get_session_id(user_id)
        query = self.query(f"SELECT fin_cat FROM sessions WHERE id={s_id}")
        return query[0][0]

    def update_active_expenses_category(self, user_id: int, cat: str):
        """Обновляет текущую выбранную пользователем статью расходов
        """
        s_id = self.get_session_id(user_id)
        self.query(f"UPDATE sessions SET fin_cat='{cat}' WHERE id={s_id}")

    def add_expense(self, slug: str, summ: int):
        """Создает новый расход
        """
        self.query(
            f"INSERT INTO finances_expenses (category, sum) VALUES ('{slug}', {summ})"
        )

    def get_expense_category_by_slug(self, slug: str) -> int:
        """Получет название категории расхода по слагу
        """
        query = self.query(f"SELECT name FROM finances_categories WHERE slug='{slug}'")
        return query[0][0]

    def update_expense_summ(self, slug: str, summ: int):
        """Обновляет сумму сборов категории расхода
        """
        self.query(f"UPDATE finances_categories SET sum={summ} WHERE slug='{slug}'")

    def update_expense_name(self, slug: str, name: str):
        """Обновляет имя категории расхода
        """
        self.query(f"UPDATE finances_categories SET name='{name}' WHERE slug='{slug}'")

    def delete_expense_catgory(self, slug: str):
        """Удаляет категорию расхода
        """
        self.query(f"DELETE FROM finances_categories WHERE slug='{slug}'")

    def get_all_donates_in_category(self, slug: str):
        """Получает сумму всех сборов по категории
        """
        query = self.query(f"SELECT sum FROM finances_donates WHERE category='{slug}'")
        return [i for (i,) in query]

    def get_all_donates(self):
        """Получает сумму всех сборов
        """
        query = self.query(f"SELECT sum FROM finances_donates")
        return [i for (i,) in query]

    def get_all_expenses_in_category(self, slug: str):
        """Получает все расходы в категории
        """
        query = self.query(f"SELECT sum FROM finances_expenses WHERE category='{slug}'")
        return [i for (i,) in query]

    def get_all_expenses(self):
        """Получает все расходы
        """
        query = self.query(f"SELECT sum FROM finances_expenses")
        return [i for (i,) in query]

    def get_expense_summ(self, slug: str):
        """Получает сумму сбора категории
        """
        query = self.query(f"SELECT sum FROM finances_categories WHERE slug='{slug}'")
        return query[0][0]

    def create_donate(self, s_id: int, slug: str):
        """Создает новый взнос
        """
        query = self.query(
            f"INSERT INTO finances_donates (student_id, category) VALUES ({s_id}, "
            f"'{slug}') RETURNING id"
        )
        return query[0][0]

    def delete_donate(self, d_id: int):
        """Удаляет взнос
        """
        self.query(f"DELETE FROM finances_donates WHERE id={d_id}")

    def get_summ_of_donate(self, d_id: int):
        """Получает сумму взноса

        Args:
            d_id: Идентификатор взноса

        Returns:
            int: Сумма взноса
        """
        query = self.query(f"SELECT sum FROM finances_donates WHERE id={d_id}")
        return query[0][0]

    def append_summ_to_donate(self, d_id: int, summ: int):
        """Добавляет сумму к сбору
        """
        source = self.get_summ_of_donate(d_id)
        new = summ + source
        self.query(f"UPDATE finances_donates SET sum={new} WHERE id={d_id}")

    def update_donate_id(self, u_id: int, d_id: int):
        """Обновляет иденифиткатор создаваемого взноса в хранилище
        """
        self.query(f"UPDATE sessions SET donate_id={d_id} WHERE vk_id={u_id}")

    def get_donate_id(self, u_id: int):
        """Получет иденифиткатор создаваемого взноса в хранилище
        """
        query = self.query(f"SELECT donate_id FROM sessions WHERE vk_id={u_id}")
        return query[0][0]

    def get_list_of_donaters_by_slug(self, slug: str, summ: int = 0):
        """Получает список идентификаторов всех внесших деньги на определенную категорию
        """
        if summ:
            query = self.query(
                f"SELECT student_id FROM finances_donates WHERE category='{slug}' AND sum >= {summ}"
            )
        else:
            query = self.query(
                f"SELECT student_id FROM finances_donates WHERE category='{slug}'"
            )
        return [i for (i,) in query]

    def get_id_of_donate_record(self, _id, slug) -> int:
        """Получает идентификатор записи взноса
        Args:
            _id: Идентификатор студента, внесшего деньги
            slug: Слаг категории расхода

        Returns:
            int: Идентификатор записи
        """
        query = self.query(
            f"SELECT id FROM finances_donates WHERE student_id={_id} AND category='{slug}'"
        )
        return query[0][0]


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			def get_active_students_ids(self) -> list:
18			"""Получает список активных студентов (с любым статусом, кроме "отчислен")
19			"""
20			ids = self.query(
21			f"SELECT user_id FROM users_info WHERE academic_status > 0 ORDER BY user_id"
22			)
23			vk_ids = [str(self.get_vk_id(_id)) for (_id,) in ids]
24			return vk_ids
25
26			def get_last_names_letters(self) -> List[str]:
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			names = self.query(
39			f"SELECT user_id, first_name, second_name FROM users_info "
40			f"WHERE substring(second_name from '^.') = '{letter}' "
41			f"AND academic_status > 0 ORDER BY user_id"
42			)
43			return names
44
45			def get_vk_id(self, _id: Union[str, int]) -> int:
46			"""Получает из базы данных идентификатор ВКонтакте по идентификатору студента
47			"""
48			vk_id = self.query(f"SELECT vk_id from users WHERE id={_id}")[0][0]
49			return vk_id
50
51			def get_user_id(self, vk_id: int) -> int:
52			"""Получает из базы данных идентификатор студента по идентификатору ВКонтакте
53			"""
54			user_id = self.query(f"SELECT id from users WHERE vk_id={vk_id}")[0][0]
55			return user_id
56
57			def get_user_name(self, _id: int) -> str:
58			"""Получает из базы данных имя по идентификатор студента
59			"""
60			name = self.query(f"SELECT first_name FROM users_info WHERE user_id={_id}")[0][
61			0
62			]
63			return name
64
65			def get_mailings_list(self) -> List[Tuple]:
66			"""Получает из базы данных весь список доступных рассылок
67			"""
68			mailings = self.query(
69			"SELECT mailing_id, mailing_name, mailing_slug from mailings"
70			)
71			return mailings
72
73			def get_subscription_status(self, slug: str, user_id: int) -> int:
74			"""Получает статус подписки пользователя на рассылку
75			"""
76			status = self.query(
77			f"SELECT {slug} FROM vk_subscriptions WHERE user_id={user_id}"
78			)[0][0]
79			return status
80
81			def is_user_exist(self, user_id: int) -> bool:
82			"""Возвращает информацию о том, существует ли пользователь в базе данных
83			"""
84			user = self.query(f"SELECT id FROM users WHERE vk_id={user_id}")
85			return bool(user)
86
87			def is_session_exist(self, user_id: int) -> bool:
88			"""Проверяет существование сессии текущего пользователя
89			"""
90			user = self.query(f"SELECT id FROM sessions WHERE vk_id={user_id}")
91			return bool(user)
92
93			def create_user(self, user_id: int) -> NoReturn:
94			"""Добавляет нового пользователя в таблицы информации и рассылок
95			"""
96			self.query(f"INSERT INTO users (vk_id) VALUES ({user_id})")
97			self.query(f"INSERT INTO vk_subscriptions DEFAULT VALUES")
98
99			def create_session(self, user_id: int) -> NoReturn:
100			"""Создает новую сессию для пользователя
101			"""
102			_id = self.query(f"SELECT id FROM users WHERE vk_id={user_id}")[0][0]
103			self.query(
104			f"INSERT INTO sessions (id, vk_id, state) VALUES ({_id}, {user_id}, 'main')"
105			)
106
107			def get_session_state(self, user_id: int) -> str:
108			"""Получает текущий статус бота из сессии
109			"""
110			st = self.query(f"SELECT state FROM sessions WHERE vk_id={user_id}")
111			return st[0][0]
112
113			def get_session_id(self, user_id: int) -> int:
114			"""Получает идентификатор сессии
115			"""
116			s_id = self.query(f"SELECT id FROM sessions WHERE vk_id={user_id}")[0][0]
117			return s_id
118
119			def update_session_state(self, user_id: int, state: str) -> NoReturn:
120			"""Изменяет текущий статус бота из сессии
121			"""
122			self.query(f"UPDATE sessions SET state='{state}' WHERE vk_id={user_id}")
123
124			def call_session_exist(self, user_id: int) -> bool:
125			"""Проверяет существование сессии призыва
126			"""
127			s_id = self.get_session_id(user_id)
128			call_exist = self.query(f"SELECT session_id FROM calls WHERE session_id={s_id}")
129			texts_exist = self.query(
130			f"SELECT session_id FROM texts WHERE session_id" f"={s_id}"
131			)
132			return bool(call_exist and texts_exist)
133
134			def create_call_session(self, user_id: int) -> NoReturn:
135			"""Создает новую сессию призыва
136			"""
137			s_id = self.get_session_id(user_id)
138			self.query(f"INSERT INTO calls (session_id) VALUES ({s_id})")
139			self.query(f"INSERT INTO texts (session_id) VALUES ({s_id})")
140
141			def get_call_message(self, user_id: int) -> str:
142			"""Получает текст призыва
143			"""
144			s_id = self.get_session_id(user_id)
145			return self.query(f"SELECT text FROM texts WHERE session_id={s_id}")[0][0]
146
147			def update_call_message(self, user_id: int, message: str) -> NoReturn:
148			"""Обновляет текст призыва
149			"""
150			s_id = self.get_session_id(user_id)
151			self.query(f"UPDATE texts SET text='{message}' WHERE session_id={s_id}")
152
153			def get_call_ids(self, user_id: int) -> str:
154			"""Получить список идентификаторов для призыва
155			"""
156			s_id = self.get_session_id(user_id)
157			ids = self.query(f"SELECT ids FROM calls WHERE session_id={s_id}")[0][0]
158			return ids
159
160			def update_call_ids(self, user_id: int, ids: str) -> NoReturn:
161			"""Перезаписать список идентификаторов для призыва
162			"""
163			s_id = self.get_session_id(user_id)
164			self.query(f"UPDATE calls SET ids='{ids}' WHERE session_id={s_id}")
165
166			def append_to_call_ids(self, user_id: int, _id: int) -> NoReturn:
167			"""Добавить идентификатор к списку для призыва
168			"""
169			ids = self.get_call_ids(user_id)
170			if ids is None:
171			ids = ""
172			ids += f"{_id},"
173			self.update_call_ids(user_id, ids)
174
175			def update_subscribe_state(self, slug: str, u_id: int, state: int):
176			"""Обновляет статус подписки на рассылку
177			"""
178			self.query(
179			f"UPDATE vk_subscriptions SET '{slug}'={state} WHERE " f"user_id={u_id}"
180			)
181
182			def empty_call_storage(self, user_id: int) -> NoReturn:
183			"""Очистить хранилище призыва (текст призыва и список идентификатора)
184			"""
185			s_id = self.get_session_id(user_id)
186			self.query(f"UPDATE calls SET ids=NULL WHERE session_id={s_id}")
187			self.query(f"UPDATE texts SET text=NULL WHERE session_id={s_id}")
188			self.query(f"UPDATE texts SET attach=NULL WHERE session_id={s_id}")
189
190			def empty_mailing_storage(self, user_id: int) -> NoReturn:
191			"""Очистить хранилище рассылок (выбранную рассылку и текст рассылки)
192			"""
193			s_id = self.get_session_id(user_id)
194			self.query(f"UPDATE mailing_mgmt SET mailing=NULL WHERE session_id={s_id}")
195			self.query(f"UPDATE mailing_mgmt SET m_text=NULL WHERE session_id={s_id}")
196			self.query(f"UPDATE mailing_mgmt SET m_attach=NULL WHERE session_id={s_id}")
197
198			def get_mailing_message(self, user_id: int) -> str:
199			"""Получить текст рассылки
200			"""
201			s_id = self.get_session_id(user_id)
202			return self.query(f"SELECT m_text FROM mailing_mgmt WHERE session_id={s_id}")[
203			0
204			][0]
205
206			def update_mailing_message(self, user_id: int, message: str) -> NoReturn:
207			"""Перезаписывает текст рассылки
208			"""
209			s_id = self.get_session_id(user_id)
210			self.query(
211			f"UPDATE mailing_mgmt SET m_text='{message}' WHERE session_id={s_id}"
212			)
213
214			def get_conversation(self, user_id: int) -> int:
215			"""Получает активную беседу
216			"""
217			s_id = self.get_session_id(user_id)
218			return self.query(f"SELECT conversation FROM sessions WHERE id={s_id}")[0][0]
219
220			def update_conversation(self, user_id: int, conv_id: int) -> NoReturn:
221			"""Изменяет активную беседу
222			"""
223			s_id = self.get_session_id(user_id)
224			self.query(f"UPDATE sessions SET conversation={conv_id} WHERE id" f"={s_id}")
225
226			def mailing_session_exist(self, user_id: int) -> bool:
227			"""Проверяет наличие сессии рассылки
228			"""
229			s_id = self.get_session_id(user_id)
230			return self.query(
231			f"SELECT session_id FROM mailing_mgmt WHERE session_id={s_id}"
232			)
233
234			def create_mailing_session(self, user_id: int) -> NoReturn:
235			"""Создает сессию рассылки
236			"""
237			s_id = self.get_session_id(user_id)
238			self.query(f"INSERT INTO mailing_mgmt (session_id) VALUES ({s_id})")
239
240			def update_mailing_session(self, user_id: int, m_slug: str) -> NoReturn:
241			"""Изменяет выбранную рассылку
242			"""
243			s_id = self.get_session_id(user_id)
244			self.query(
245			f"UPDATE mailing_mgmt SET mailing='{m_slug}' WHERE session_id={s_id}"
246			)
247
248			def get_mailing_session(self, user_id: int) -> str:
249			"""Получает выбранную рассылку
250			"""
251			s_id = self.get_session_id(user_id)
252			return self.query(f"SELECT mailing FROM mailing_mgmt WHERE session_id={s_id}")[
253			0
254			][0]
255
256			def fetch_subcribers(self, slug: str) -> str:
257			"""Собирает подписчиков рассылки
258			"""
259			user_ids = self.query(f"SELECT user_id FROM vk_subscriptions WHERE {slug}=1")
260			user_ids = [_id for (_id,) in user_ids]
261			ids = []
262			for i, _id in enumerate(user_ids):
263			ids.append(self.query(f"SELECT vk_id FROM users WHERE id={_id}")[0][0])
264			ids = ",".join(map(str, ids))
265			return ids
266
267			def get_names_using_status(self, user_id: int) -> bool:
268			"""Получает статус использования имён
269			"""
270			s_id = self.get_session_id(user_id)
271			names_using = self.query(f"SELECT names_using FROM sessions WHERE id={s_id}")[
272			0
273			][0]
274			return bool(names_using)
275
276			def update_names_using_status(self, user_id: int, value: int):
277			"""Изменяет статус использования имён
278			"""
279			s_id = self.get_session_id(user_id)
280			self.query(f"UPDATE sessions SET names_using={value} WHERE id={s_id}")
281
282			def get_users_names(self, ids: list) -> List[str]:
283			"""Получает список имён по списку идентификаторов ВКонтакте
284			"""
285			ids = ", ".join(ids)
286			query = self.query(
287			f"SELECT first_name FROM users INNER JOIN users_info ON users.id = "
288			f"users_info.user_id WHERE vk_id in ({ids}) ORDER BY "
289			f"position(vk_id::text in '{ids}')"
290			)
291			names = [i for (i,) in query]
292			return names
293
294			def get_call_attaches(self, user_id: int):
295			"""Получает список вложений для сообщения с призывом
296			"""
297			s_id = self.get_session_id(user_id)
298			query = self.query(f"SELECT attach FROM texts WHERE session_id={s_id}")
299			return query[0][0]
300
301			def update_call_attaches(self, user_id: int, attach: str):
302			"""Обновляет список вложений для сообщения с призывом
303			"""
304			s_id = self.get_session_id(user_id)
305			self.query(f"UPDATE texts SET attach='{attach}' WHERE session_id={s_id}")
306
307			def get_mailing_attaches(self, user_id: int):
308			"""Получает список вложений для сообщения рассылки
309			"""
310			s_id = self.get_session_id(user_id)
311			query = self.query(f"SELECT m_attach FROM mailing_mgmt WHERE session_id={s_id}")
312			return query[0][0]
313
314			def update_mailing_attaches(self, user_id: int, attach: str):
315			"""Обновляет список вложений для сообщения рассылки
316			"""
317			s_id = self.get_session_id(user_id)
318			self.query(
319			f"UPDATE mailing_mgmt SET m_attach='{attach}' WHERE session_id={s_id}"
320			)
321
322			def get_list_of_finances_categories(self) -> List[Tuple]:
323			"""Получает список доступных категорий расходов
324			"""
325			query = self.query("SELECT name, slug FROM finances_categories")
326			return query
327
328			def add_expences_category(self, name: str, slug: str, s: int):
329			"""Созадет новую категорию расходов
330			"""
331			query = self.query(
332			f"INSERT INTO finances_categories (name, slug, sum) VALUES ('{name}', "
333			f"'{slug}', {s}) RETURNING id"
334			)
335
336			return query[0][0]
337
338			def get_active_expenses_category(self, user_id: int):
339			"""Получает текущую выбранную пользователем статью расходов
340			"""
341			s_id = self.get_session_id(user_id)
342			query = self.query(f"SELECT fin_cat FROM sessions WHERE id={s_id}")
343			return query[0][0]
344
345			def update_active_expenses_category(self, user_id: int, cat: str):
346			"""Обновляет текущую выбранную пользователем статью расходов
347			"""
348			s_id = self.get_session_id(user_id)
349			self.query(f"UPDATE sessions SET fin_cat='{cat}' WHERE id={s_id}")
350
351			def add_expense(self, slug: str, summ: int):
352			"""Создает новый расход
353			"""
354			self.query(
355			f"INSERT INTO finances_expenses (category, sum) VALUES ('{slug}', {summ})"
356			)
357
358			def get_expense_category_by_slug(self, slug: str) -> int:
359			"""Получет название категории расхода по слагу
360			"""
361			query = self.query(f"SELECT name FROM finances_categories WHERE slug='{slug}'")
362			return query[0][0]
363
364			def update_expense_summ(self, slug: str, summ: int):
365			"""Обновляет сумму сборов категории расхода
366			"""
367			self.query(f"UPDATE finances_categories SET sum={summ} WHERE slug='{slug}'")
368
369			def update_expense_name(self, slug: str, name: str):
370			"""Обновляет имя категории расхода
371			"""
372			self.query(f"UPDATE finances_categories SET name='{name}' WHERE slug='{slug}'")
373
374			def delete_expense_catgory(self, slug: str):
375			"""Удаляет категорию расхода
376			"""
377			self.query(f"DELETE FROM finances_categories WHERE slug='{slug}'")
378
379			def get_all_donates_in_category(self, slug: str):
380			"""Получает сумму всех сборов по категории
381			"""
382			query = self.query(f"SELECT sum FROM finances_donates WHERE category='{slug}'")
383			return [i for (i,) in query]
384
385			def get_all_donates(self):
386			"""Получает сумму всех сборов
387			"""
388			query = self.query(f"SELECT sum FROM finances_donates")
389			return [i for (i,) in query]
390
391			def get_all_expenses_in_category(self, slug: str):
392			"""Получает все расходы в категории
393			"""
394			query = self.query(f"SELECT sum FROM finances_expenses WHERE category='{slug}'")
395			return [i for (i,) in query]
396
397			def get_all_expenses(self):
398			"""Получает все расходы
399			"""
400			query = self.query(f"SELECT sum FROM finances_expenses")
401			return [i for (i,) in query]
402
403			def get_expense_summ(self, slug: str):
404			"""Получает сумму сбора категории
405			"""
406			query = self.query(f"SELECT sum FROM finances_categories WHERE slug='{slug}'")
407			return query[0][0]
408
409			def create_donate(self, s_id: int, slug: str):
410			"""Создает новый взнос
411			"""
412			query = self.query(
413			f"INSERT INTO finances_donates (student_id, category) VALUES ({s_id}, "
414			f"'{slug}') RETURNING id"
415			)
416			return query[0][0]
417
418			def delete_donate(self, d_id: int):
419			"""Удаляет взнос
420			"""
421			self.query(f"DELETE FROM finances_donates WHERE id={d_id}")
422
423			def get_summ_of_donate(self, d_id: int):
424			"""Получает сумму взноса
425
426			Args:
427			d_id: Идентификатор взноса
428
429			Returns:
430			int: Сумма взноса
431			"""
432			query = self.query(f"SELECT sum FROM finances_donates WHERE id={d_id}")
433			return query[0][0]
434
435			def append_summ_to_donate(self, d_id: int, summ: int):
436			"""Добавляет сумму к сбору
437			"""
438			source = self.get_summ_of_donate(d_id)
439			new = summ + source
440			self.query(f"UPDATE finances_donates SET sum={new} WHERE id={d_id}")
441
442			def update_donate_id(self, u_id: int, d_id: int):
443			"""Обновляет иденифиткатор создаваемого взноса в хранилище
444			"""
445			self.query(f"UPDATE sessions SET donate_id={d_id} WHERE vk_id={u_id}")
446
447			def get_donate_id(self, u_id: int):
448			"""Получет иденифиткатор создаваемого взноса в хранилище
449			"""
450			query = self.query(f"SELECT donate_id FROM sessions WHERE vk_id={u_id}")
451			return query[0][0]
452
453			def get_list_of_donaters_by_slug(self, slug: str, summ: int = 0):
454			"""Получает список идентификаторов всех внесших деньги на определенную категорию
455			"""
456			if summ:
457			query = self.query(
458			f"SELECT student_id FROM finances_donates WHERE category='{slug}' AND sum >= {summ}"
459			)
460			else:
461			query = self.query(
462			f"SELECT student_id FROM finances_donates WHERE category='{slug}'"
463			)
464			return [i for (i,) in query]
465
466			def get_id_of_donate_record(self, _id, slug) -> int:
467			"""Получает идентификатор записи взноса
468			Args:
469			_id: Идентификатор студента, внесшего деньги
470			slug: Слаг категории расхода
471
472			Returns:
473			int: Идентификатор записи
474			"""
475			query = self.query(
476			f"SELECT id FROM finances_donates WHERE student_id={_id} AND category='{slug}'"
477			)
478			return query[0][0]
479

dadyarri / ralph

Push — master ( 0e504e...758c62 )

database.Database.get_summ_of_donate() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like