56 lines
2 KiB
Python
56 lines
2 KiB
Python
from core.models.Model import Model
|
|
from dataclasses import dataclass
|
|
|
|
_table_name: str = 'operators'
|
|
|
|
_table_definition: str = """
|
|
'id' int UNIQUE,
|
|
'name' varchar,
|
|
'public_key' varchar,
|
|
'nostr_public_key' varchar,
|
|
'nostr_profile_reference' varchar,
|
|
'nostr_attestation_event_reference' varchar
|
|
"""
|
|
|
|
|
|
@dataclass
|
|
class Operator(Model):
|
|
id: int
|
|
name: str
|
|
public_key: str
|
|
nostr_public_key: str
|
|
nostr_profile_reference: str
|
|
nostr_attestation_event_reference: str
|
|
|
|
@staticmethod
|
|
def find_by_id(id: int):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_one('SELECT * FROM operators WHERE id = ? LIMIT 1', Operator.factory, [id])
|
|
|
|
@staticmethod
|
|
def exists(id: int):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_exists('SELECT * FROM operators WHERE id = ?', [id])
|
|
|
|
@staticmethod
|
|
def all():
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_all('SELECT * FROM operators', Operator.factory)
|
|
|
|
@staticmethod
|
|
def truncate():
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition, drop_existing=True)
|
|
|
|
@staticmethod
|
|
def save_many(operators):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
Model._insert_many('INSERT INTO operators VALUES(?, ?, ?, ?, ?, ?)', Operator.tuple_factory, operators)
|
|
|
|
@staticmethod
|
|
def factory(cursor, row):
|
|
local_fields = [column[0] for column in cursor.description]
|
|
return Operator(**{key: value for key, value in zip(local_fields, row)})
|
|
|
|
@staticmethod
|
|
def tuple_factory(operator):
|
|
return operator.id, operator.name, operator.public_key, operator.nostr_public_key, operator.nostr_profile_reference, operator.nostr_attestation_event_reference
|