94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
import sqlite3
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from os import link
|
|
from typing import Dict, List, Tuple
|
|
|
|
|
|
@dataclass
|
|
class Link:
|
|
link: str
|
|
user_id: str
|
|
id: int = None
|
|
read_at: datetime = None
|
|
added_at: datetime = None
|
|
|
|
def as_dict(self):
|
|
def timestamp(t): return t.timestamp() if t is not None else None
|
|
return {
|
|
"id": self.id,
|
|
"user_id": self.user_id,
|
|
"read_at": timestamp(self.read_at),
|
|
"added_at": timestamp(self.added_at),
|
|
"link": self.link,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, dict: Dict) -> 'Link':
|
|
def convert_date(t): return datetime.fromtimestamp(
|
|
t) if t is not None else None
|
|
return Link(
|
|
id=dict.get('id', None),
|
|
user_id=dict["user_id"],
|
|
read_at=convert_date(dict.get("read_at", None)),
|
|
added_at=convert_date(dict.get("added_at", None)),
|
|
link=dict["link"]
|
|
)
|
|
|
|
@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
|
|
)
|
|
|
|
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]
|
|
|
|
def mark_as_read(self, db: sqlite3.Connection):
|
|
assert self.id is not None
|
|
self.read_at = datetime.now()
|
|
cur = db.cursor()
|
|
cur.execute(
|
|
"UPDATE links SET read_at=? WHERE user_id=? AND id=?", (self.read_at, self.user_id, self.id))
|
|
|
|
def delete(self, db: sqlite3.Connection):
|
|
assert self.id is not None
|
|
cur = db.cursor()
|
|
cur.execute(
|
|
"DELETE FROM links WHERE user_id=? AND id=?", (self.user_id, self.id))
|
|
self.id = None
|
|
|
|
|
|
@classmethod
|
|
def _get(cls, db: sqlite3.Connection, where: str, values: Tuple = ()) -> List['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) -> List['Link']:
|
|
return cls._get(db, "user_id = ? AND read_at IS NULL", (user_id,))
|
|
|
|
@classmethod
|
|
def get(cls, db: sqlite3.Connection, user_id: str, link_id: int) -> 'Link':
|
|
rows = list(cls._get(db, "user_id = ? AND id = ?", (user_id, link_id)))
|
|
if len(rows) < 1:
|
|
return None
|
|
else:
|
|
return rows[0]
|