Инспекция кода: запросы к базе данных

Делаю бота оповещения расписания, пришлось затронуть классы, но я ни разу с ними не сталкивался:

import sqlite3

class RequestsInBD:
    def __init__(self):
        self.connection = sqlite3.Connection('datebase_of_users.db')
        self.cursor = self.connection.cursor()

    def _commit_and_close_bd(self, connection):
        connection.commit()
        connection.close()
    
    def check(self, user_id):
        all_id = self.cursor.execute('SELECT id FROM grade_of_users').fetchall()
        self._commit_and_close_bd(self.connection)
        for item in all_id:
            if user_id in item:
                return False
        return True
    
    def add_user(self, user_id, grade):
        self.cursor.execute(f'INSERT INTO grade_of_users (id, grade, notifications) VALUES (?, ?, ?)', (user_id, grade, 1))
        self._commit_and_close_bd(self.connection)

    def return_user_grade(self, user_id):
        self.cursor.execute(f'SELECT id, grade FROM grade_of_users WHERE id = {user_id}')
        grade = self.cursor.fetchone()[1]
        self._commit_and_close_bd(self.connection)
        return grade if grade else False
    
    def get_users(self):
        self.cursor.execute('SELECT id, grade, notifications FROM grade_of_users')
        res = self.cursor.fetchall()
        self._commit_and_close_bd(self.connection)
        return res
    
    def new_grade(self, grade, user_id):
        self.cursor.execute(f'UPDATE grade_of_users SET grade = ? WHERE id = ?', (grade, user_id))
        self._commit_and_close_bd(self.connection)

вроде код рабочий, но правильно ли он написан. Chatgpt говорит, что в принципе все нормально, но я как то не уверен в его словах. (этот код представлен в отдельном от основной программы файле, откуда я импортирую этот класс, и при каждом обращении я создаю объект этого класса (экземпляр) и я не уверен что это тоже правильно)


Ответы (2 шт):

Автор решения: Швеев Алексей

SQL инъекция

Самая опасная ваша ошибка - вы используете прямую подстановку строк. Злоумышленник может легко расширить свой запрос и получить практически любые данные из вашей БД.

Для безопасной передачи параметров в запрос вы можете использовать инструментарий библиотеки sqlite3 таким образом:

self.cursor.execute(f'INSERT INTO grade_of_users (id, grade, notifications) VALUES (?, ?, 1)', (user_id, grade))

Документация

Закрытие соединения каждый запрос

подметил Stanislav Volodarskiy

Объекты позволяют быть созданными однажды, и использованными по всему коду. Поэтому постоянное создание и удаление подключение к базе данных не имеет смысла.

_commit_and_close_bd (а точнее _commit_and_close_db) лучше разбить на 2 метода:

commit и close соответственно.

Теперь в конце каждого запроса можно сохранять данные изменения в базе данных используя commit, но продолжать использовать то же соединение для новых запросов:

db = RequestsInBD()

db.add_user(5325, "other")
db.add_user(1241, "something")
...
# Завершение работы приложения:
db.close()

По хорошему в вашем случае создавать соединение надо только в самом начале программы и закрывать только в самом её конце.

При этом данные методы так же не должны принимать параметры, т. к. соединение и так доступно через self

Нейминг

Во-первых не bd, а db, от слов DataBase.

Во-вторых я бы переименовал RequestsInBD в UsersDatabase.

Ну и функцию return_user_grade лучше переименовать в get_user_grade, а функцию add_grade в set_grade

Универсальность

Ваша база данных сейчас импортируется только из файла с названием datebase_of_users.db

Но что если вы хотите поменять его, вывести в конфиг или вам вообще необходимо одновременно подключиться одновременно к нескольким инстансам?

Лучше передавать параметры при инициализации:

class UsersDatabase:
    def __init__(self, file):
        self.connection = sqlite3.Connection(file)
        self.cursor = self.connection.cursor()

...

users_db = UsersDatabase("database_of_users.db")

Модульность

Не уверен, что у вас всё в одном файле, однако на всякий случай...

Класс с базой данных желательно вынести в свой файл, а после импортировать через import file_name:

Файл database/users_database.py:

import sqlite3

class UsersDatabase:
    ...

Файл main.py:

from database.users_database import UsersDatabase

users_db = UsersDatabase("database.db")

# Основное приложение
...

# Конец приложения:
db.close()

Оптимизация

В методе check вместо получения всех id-шников из базы данных и поиска своего (чьё количество может достигать тысячи), лучше сразу проверить, имеется ли id в базе данных инструментами SQL:

def check(self, user_id):
    result = self.cursor.execute('SELECT 1 FROM grade_of_users WHERE id = ?', (user_id,)).fetchone()
    
    self.commit() # А нужен ли вообще?
    
    return result is None

Итого

Получается следующий код:

import sqlite3

class UsersDatabase:
    def __init__(self, path):
        self.connection = sqlite3.Connection(path)
        self.cursor = self.connection.cursor()

    def commit(self):
        self.connection.commit()
        
    def close(self):
        self.connection.close()
    
    def check(self, user_id):
        result = self.cursor.execute('SELECT 1 FROM grade_of_users WHERE id = ?', (user_id,)).fetchone()
    
        self.commit() # А нужен ли вообще?
    
        return result is None
    
    def add_user(self, user_id, grade):
        self.cursor.execute('INSERT INTO grade_of_users (id, grade, notifications) VALUES (?, ?, 1)', (user_id, grade))
        self.commit()

    def get_user_grade(self, user_id):
        self.cursor.execute('SELECT id, grade FROM grade_of_users WHERE id = ?', (user_id,))
        grade = self.cursor.fetchone()[1]
        self.commit()
        return grade if grade else False
    
    def get_users(self):
        self.cursor.execute('SELECT id, grade, notifications FROM grade_of_users')
        res = self.cursor.fetchall()
        self.commit()
        return res
    
    def set_grade(self, grade, user_id):
        self.cursor.execute(f'UPDATE grade_of_users SET grade = ? WHERE id = ?', (grade, user_id))
        self.commit()

Который надо использовать так:

from users_database import UsersDatabase

users_db = UsersDatabase("datebase_of_users.db")

users_db.add_user(1324, "abc")
users_db.set_grade(1324, "other")
...

# в конце:
users_db.close()
→ Ссылка
Автор решения: Noskill
def get_user_grade(self, user_id):
        self.cursor.execute('SELECT id, grade FROM grade_of_users WHERE id = ?', (user_id,))
        grade = self.cursor.fetchone()[1]
        self.commit()
        return grade if grade else False

а если ничего нет? grade = self.cursor.fetchone()[1] что вернет, если тут None? что делать если самой таблицы grade_of_users нет? почему не засунуть соединение с базой(открытие и закрытие) в контекстный менеджер?

→ Ссылка