Source code for aiogbclient.utils.mixins

import json

from aiogbclient.client_config import ENCODING
from aiogbclient.database.controller import ClientMessages
from aiogbclient.database.models import CBase


[docs]class ConvertMixin: def _dict_to_bytes(self, msg_dict): """ Преобразование словаря в байты :param msg_dict: словарь :return: bytes """ # Проверям, что пришел словарь if isinstance(msg_dict, dict): jmessage = json.dumps(msg_dict) # Преобразуем словарь в json bmessage = jmessage.encode(ENCODING) # Переводим json в байты return bmessage else: raise TypeError def _bytes_to_dict(self, msg_bytes): """ Получение словаря из байтов :param msg_bytes: сообщение в виде байтов :return: словарь сообщения """ # Если переданы байты if isinstance(msg_bytes, bytes): jmessage = msg_bytes.decode(ENCODING) # Декодируем message = json.loads(jmessage) # Из json делаем словарь # Если там был словарь if isinstance(message, dict): return message # Возвращаем сообщение else: raise TypeError # Нам прислали неверный тип else:
raise TypeError # Передан неверный тип
[docs]class DbInterfaceMixin: def __init__(self, db_path): self._cm = ClientMessages(db_path, CBase, echo=False) # init DB
[docs] def add_client(self, username, info=None):
return self._cm.add_client(username, info)
[docs] def get_client_by_username(self, username):
return self._cm.get_client_by_username(username)
[docs] def add_contact(self, client_username, contact_username):
return self._cm.add_contact(client_username, contact_username)
[docs] def del_contact(self, client_username, contact_username):
return self._cm.del_contact(client_username, contact_username)
[docs] def get_contacts(self, client_username):
return self._cm.get_contacts(client_username)
[docs] def get_all_clients(self):
return self._cm.get_all_clients()
[docs] def add_client_history(self, client_username, ip_addr='8.8.8.8'):
return self._cm.add_client_history(client_username, ip_addr)
[docs] def get_client_history(self, client_username):
return self._cm.get_client_history(client_username)
[docs] def add_client_message(self, client_username, contact_username, text_msg):
return self._cm.add_client_message(client_username, contact_username, text_msg)
[docs] def get_client_messages(self, client_username):
return self._cm.get_client_messages(client_username)
[docs] def set_user_offline(self, client_username):
return self._cm.set_user_offline(client_username)
[docs] def set_user_online(self, client_username):
return self._cm.set_user_online(client_username)
[docs] def get_user_status(self, client_username):
return self._cm.get_user_status(client_username)