49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
from os import read
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Tuple
|
|
|
|
|
|
@dataclass
|
|
class Link:
|
|
link: str
|
|
user_id: str
|
|
id: int = None
|
|
read_at: datetime = None
|
|
added_at: datetime = None
|
|
|
|
def create(self, db: sqlite3.Connection):
|
|
self.added_at = datetime.now()
|
|
cur = db.cursor()
|
|
rows = (cur.execute(
|
|
"INSERT INTO links(link, user_id, added_at) VALUES(?, ?, ?) RETURNING id;",
|
|
(self.link, self.user_id, self.added_at)
|
|
))
|
|
r = list(rows)
|
|
assert len(r) == 1
|
|
assert len(r[0]) == 1
|
|
assert int(r[0][0])
|
|
self.id = r[0][0]
|
|
|
|
@classmethod
|
|
def _from_tuple(cls, tuple) -> 'Link':
|
|
id, link, user_id, read_at, added_at = tuple
|
|
return cls(
|
|
id=id,
|
|
link=link,
|
|
user_id=user_id,
|
|
read_at=read_at,
|
|
added_at=added_at
|
|
)
|
|
|
|
@classmethod
|
|
def _get(cls, db: sqlite3.Connection, where: str, values: Tuple = ()) -> 'Link':
|
|
cur = db.cursor()
|
|
rows = cur.execute(f"SELECT * FROM links WHERE {where}", values)
|
|
return map(cls._from_tuple, rows)
|
|
|
|
@classmethod
|
|
def get_unread(cls, db: sqlite3.Connection, user_id: str):
|
|
return cls._get(db, "user_id = ? AND read_at IS NULL", (user_id,))
|