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]