Ну давай попробуем. Опустим бизнес-логику получения данных о пользователе, это всё таки зависит от реализации твоего бота. Пойдём с того момента что у тебя уже есть эти данные и тебе их нужно сохранить.
Смотри первое что тебе нужно сделать - это проверить существует ли нужная тебе таблица в базе.
Если нет то создать её.
Создаёшь модуль допустим
db
В нём создаёшь файл допустим
utils.py c таким содержимым:
utils.pyfrom sqlite3 import Connection as SqliteConnection
from sqlite3 import Cursor as SqliteCursor
sql = """
pragma foreign_keys = on;
create table if not exists users
(
id integer primary key autoincrement,
internal_id text not null,
first_name text not null,
last_name text not null,
username text not null unique,
created date not null default current_date
);
create unique index if not exists users_id_uindex
on users (id);
create unique index if not exists users_internal_id_uindex
on users (internal_id);
create index if not exists users_first_name_index
on users (created);
create index if not exists users_last_name_index
on users (created);
create unique index if not exists users_username_uindex
on users (username);
create index if not exists users_created_index
on users (created);
"""
def check_db(db_connect: SqliteConnection,
db_cursor: SqliteCursor) -> None:
db_cursor.executescript(sql)
db_connect.commit()
Создаешь ещё один файл в модуле
db c именем
queries.py
В нём для удобства описываешь запросы к базе.
Тебе нужно 2 запроса:
1) На сохранение данным
2 На получение данных
Делаем:
queries.pyfrom sqlite3 import Connection as SqliteConnection
from sqlite3 import Cursor as SqliteCursor
from typing import List
def insert_user(db_cursor: SqliteCursor,
db_connect: SqliteConnection,
internal_id: str, username: str,
first_name: str, last_name: str) -> None:
db_cursor.execute("""
insert into users(internal_id, username, first_name, last_name)
select :internal_id, :username, :first_name, :last_name
where not exists(select 1 from users where internal_id = :internal_id and username = :username)
""", {
'internal_id': internal_id,
'username': username,
'first_name': first_name,
'last_name': last_name,
})
db_connect.commit()
def select_user(db_connect: SqliteConnection,
username: str) -> List[tuple] or None:
result = db_connect.execute(
f"""select *
from users
where username = :username""",
{'username': username},
).fetchone()
return result
Дальше в том месте, где тебе нужно сохранить данные делаешь что-то такое:
somewhere _in_your_codefrom contextlib import closing
import sqlite3
from db.utils import check_db
from db.queries import insert_user
with closing(sqlite3.connect('db_name')) as db_connect:
with closing(db_connect.cursor()) as db_cursor:
check_db(db_connect, db_cursor) # проверяем есть ли таблица в базе, если нет то создаём её
insert_user(db_cursor, db_connect, 'internal_id', 'username', 'first_name', 'last_name') # если в базе ещё нет такого юзера с internal_id и username то он создастся
А в том месте, где тебе нужно получить данные что-то такое:
somewhere _in_your_codefrom contextlib import closing
import sqlite3
from db.utils import check_db
from db.queries import select_user
with closing(sqlite3.connect('db_name')) as db_connect:
with closing(db_connect.cursor()) as db_cursor:
check_db(db_connect, db_cursor) # проверяем есть ли таблица в базе, если нет то создаём её
user = select_user(db_connect, 'username')
Вот вроде и всё.
Пы.Сы. Код писался в текстовом редакторе, возможно что-то проебал.
Да и пользуйся
докой, не стесняйся
