2023-01-10 00:48:55 +00:00
import sqlite3
from typing import Optional
2023-01-14 19:03:56 +00:00
from collections import namedtuple
2023-01-10 00:48:55 +00:00
2023-01-14 19:03:56 +00:00
from mirrortea . repository . abstract_db_repository import AbstractDatabaseRepository
2023-01-10 00:48:55 +00:00
2023-01-14 19:03:56 +00:00
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 ,
2023-01-10 00:48:55 +00:00
MATRIX_ROOMS_SQL ,
)
TABLES_LIST = [
2023-01-14 19:03:56 +00:00
# PLATFORM_USER_MATRIX_CHATS_SQL,
PLATFORMS_USERS_SQL ,
2023-01-10 00:48:55 +00:00
MATRIX_ROOMS_SQL ,
]
2023-01-14 19:03:56 +00:00
PATH_TO_DB = " database.db "
2023-01-10 00:48:55 +00:00
class SQLiteDatabaseRepository ( AbstractDatabaseRepository ) :
2023-01-14 19:03:56 +00:00
def __init__ ( self , path = PATH_TO_DB ) :
2023-01-10 00:48:55 +00:00
self . path = path
2023-01-14 19:03:56 +00:00
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 )
2023-01-10 00:48:55 +00:00
def create_tables ( self ) - > None :
""" Init tables in database """
for table in TABLES_LIST :
self . conn . execute ( table )
self . conn . commit ( )
2023-01-14 19:03:56 +00:00
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 ,
) ,
2023-01-14 23:47:09 +00:00
)
2023-01-14 19:03:56 +00:00
except sqlite3 . IntegrityError :
return UserAlreadyExistError ( " User already exist! " )
self . conn . commit ( )
return self . get_user_by_id ( user_id = user . user_id )
2023-01-14 23:47:09 +00:00
def update_user ( self ) - > None :
""" """
2023-01-14 19:03:56 +00:00
def delete_user ( self ) - > None :
""" """
def create_matrix_room ( self ) - > MatrixRoom :
""" """
def get_matrix_room ( self ) - > MatrixRoom :
""" """
2023-01-14 23:47:09 +00:00
def update_matrix_room ( self ) - > MatrixRoom :
""" """
2023-01-14 19:03:56 +00:00
def delete_matrix_room ( self ) - > None :
""" """