q1TstanP

1import sqlite3
2
3
4def ensure_connection(func):
5    """ Декоратор для подключения к СУБД: открывает соединение,
6        выполняет переданную функцию и закрывает за собой соединение.
7        Потокобезопасно!
8    """
9    def inner(*args, **kwargs):
10        with sqlite3.connect('anketa.db') as conn:
11            kwargs['conn'] = conn
12            res = func(*args, **kwargs)
13        return res
14
15    return inner
16
17
18@ensure_connection
19def init_db(conn, force: bool = False):
20    """ Проверить что нужные таблицы существуют, иначе создать их
21
22        Важно: миграции на такие таблицы вы должны производить самостоятельно!
23
24        :param conn: подключение к СУБД
25        :param force: явно пересоздать все таблицы
26    """
27    c = conn.cursor()
28
29    # Информация о пользователе
30    # TODO: создать при необходимости...
31
32    # Сообщения от пользователей
33    if force:
34        c.execute('DROP TABLE IF EXISTS user_message')
35
36    c.execute('''
37        CREATE TABLE IF NOT EXISTS user_message (
38            id          INTEGER PRIMARY KEY,
39            user_id     INTEGER NOT NULL,
40            text        TEXT NOT NULL
41        )
42    ''')
43
44    # Сохранить изменения
45    conn.commit()
46
47
48@ensure_connection
49def add_message(conn, user_id: int, text: str):
50    c = conn.cursor()
51    c.execute('INSERT INTO user_message (user_id, text) VALUES (?, ?)', (user_id, text))
52    conn.commit()
53
54
55@ensure_connection
56def count_messages(conn, user_id: int):
57    c = conn.cursor()
58    c.execute('SELECT COUNT(*) FROM user_message WHERE user_id = ? LIMIT 1', (user_id, ))
59    (res, ) = c.fetchone()
60    return res
61
62
63@ensure_connection
64def list_messages(conn, user_id: int, limit: int = 10):
65    c = conn.cursor()
66    c.execute('SELECT id, text FROM user_message WHERE user_id = ? ORDER BY id DESC LIMIT ?', (user_id, limit))
67    return c.fetchall()