59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
from dto import Device, SimCard, SmsMessage, QueuedSmsMessage
|
|
|
|
|
|
def get_all_devices(cur) -> list[Device]:
|
|
res = cur.execute("SELECT access_key, type, name FROM devices")
|
|
devices_from_db = res.fetchall()
|
|
|
|
devices = []
|
|
for item in devices_from_db:
|
|
devices.append(Device.convert(item))
|
|
return devices
|
|
|
|
|
|
def get_sim_cards_by_device(cur, access_key: str) -> list[SimCard]:
|
|
if access_key is None or not access_key:
|
|
return []
|
|
|
|
sim_cards_from_db = cur.execute("SELECT phone_number, device_access_key FROM sim_cards WHERE device_access_key = ?", (access_key,)).fetchall()
|
|
sim_cards = []
|
|
for item in sim_cards_from_db:
|
|
sim_cards.append(SimCard.convert(item))
|
|
return sim_cards
|
|
|
|
|
|
def get_sms_messages_by_local_phone_number(cur, local_phone_number: str) -> list[SmsMessage]:
|
|
if local_phone_number is None or not local_phone_number:
|
|
return []
|
|
|
|
msgs_from_db = cur.execute("SELECT * FROM messages WHERE local_phone_number = ?", (local_phone_number,)).fetchall()
|
|
msgs = []
|
|
for item in msgs_from_db:
|
|
msgs.append(SmsMessage.convert(item))
|
|
return msgs
|
|
|
|
|
|
def send_sms_message(cur, content: str, sender_access_key: str, local_phone_number: str, remote_phone_number: str) -> bool:
|
|
if content is None or not content \
|
|
or sender_access_key is None or not sender_access_key \
|
|
or local_phone_number is None or not local_phone_number \
|
|
or remote_phone_number is None or not remote_phone_number:
|
|
return False
|
|
|
|
cur.execute("INSERT INTO message_queue VALUES (?, ?, ?, ?)", \
|
|
(content, sender_access_key, local_phone_number, remote_phone_number))
|
|
cur.commit()
|
|
return True
|
|
|
|
|
|
def get_queued_sms_messages(cur, local_phone_number) -> list[QueuedSmsMessage]:
|
|
if local_phone_number is None or not local_phone_number:
|
|
return []
|
|
|
|
msgs_from_db = cur.execute("DELETE FROM message_queue WHERE local_phone_number = ? RETURNING *", (local_phone_number,)).fetchall()
|
|
cur.commit()
|
|
msgs = []
|
|
for item in msgs_from_db:
|
|
msgs.append(QueuedSmsMessage.convert(item))
|
|
return msgs
|