104 lines
2.8 KiB
Python
104 lines
2.8 KiB
Python
import sqlite3
|
|
from typing import Optional
|
|
from collections import namedtuple
|
|
|
|
from mirrortea.repository.abstract_db_repository import AbstractDatabaseRepository
|
|
|
|
from mirrortea.models.matrix_room import MatrixRoom
|
|
from mirrortea.models.user import User
|
|
|
|
from mirrortea.repository.exceptions import (
|
|
UserNotFoundError,
|
|
MatrixRoomAlreadyExistError,
|
|
MatrixRoomNotFoundError,
|
|
UserAlreadyExistError,
|
|
)
|
|
|
|
from mirrortea.repository.sqlite_init_tables import (
|
|
# PLATFORM_USER_MATRIX_CHATS_SQL,
|
|
PLATFORMS_USERS_SQL,
|
|
MATRIX_ROOMS_SQL,
|
|
)
|
|
|
|
TABLES_LIST = [
|
|
# PLATFORM_USER_MATRIX_CHATS_SQL,
|
|
PLATFORMS_USERS_SQL,
|
|
MATRIX_ROOMS_SQL,
|
|
]
|
|
|
|
PATH_TO_DB = "database.db"
|
|
|
|
|
|
class SQLiteDatabaseRepository(AbstractDatabaseRepository):
|
|
def __init__(self, path=PATH_TO_DB):
|
|
self.path = path
|
|
self.conn = sqlite3.connect(self.path)
|
|
self.cursor = self.conn.cursor()
|
|
|
|
def _namedtuple_factory(cursor, row):
|
|
fields = [column[0] for column in cursor.description]
|
|
cls = namedtuple("Row", fields)
|
|
return cls._make(row)
|
|
|
|
def create_tables(self) -> None:
|
|
"""Init tables in database"""
|
|
|
|
for table in TABLES_LIST:
|
|
self.conn.execute(table)
|
|
self.conn.commit()
|
|
|
|
def get_user_by_id(self, user_id: str) -> User:
|
|
"""Get user by user id"""
|
|
|
|
self.conn.row_factory = self._namedtuple_factory
|
|
self.cur_userdata = self.conn.execute(
|
|
"SELECT user_id AS user_id, first_name AS first_name, last_name AS last_name, username AS username, avatar_hash AS avatar_hash FROM platforms_users WHERE user_id = ?",
|
|
(user_id,),
|
|
)
|
|
|
|
self.userdata = self.cur_userdata.fetchone()
|
|
|
|
if self.userdata == ():
|
|
return UserNotFoundError("User not found!")
|
|
|
|
return User(
|
|
id=user_id,
|
|
first_name=self.userdata.user_id,
|
|
last_name=self.userdata.first_name,
|
|
username=self.userdata.username,
|
|
avatar_hash=self.userdata.avatar_hash,
|
|
)
|
|
|
|
def create_user(self, user: User) -> User:
|
|
"""Create user"""
|
|
try:
|
|
self.cursor.execute(
|
|
"INSERT INTO platforms_users VALUES (?, ?, ?, ?, ?)",
|
|
(
|
|
user.user_id,
|
|
user.first_name,
|
|
user.last_name,
|
|
user.username,
|
|
user.avatar_hash,
|
|
),
|
|
).fetchall()
|
|
|
|
except sqlite3.IntegrityError:
|
|
return UserAlreadyExistError("User already exist!")
|
|
|
|
self.conn.commit()
|
|
|
|
return self.get_user_by_id(user_id=user.user_id)
|
|
|
|
def delete_user(self) -> None:
|
|
""""""
|
|
|
|
def create_matrix_room(self) -> MatrixRoom:
|
|
""""""
|
|
|
|
def get_matrix_room(self) -> MatrixRoom:
|
|
""""""
|
|
|
|
def delete_matrix_room(self) -> None:
|
|
""""""
|