model.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425
  1. from contextlib import contextmanager
  2. from dataclasses import dataclass
  3. import datetime
  4. import json
  5. from passlib.apps import custom_app_context as pwd
  6. import peewee
  7. import playhouse.shortcuts
  8. from typing import Iterator, List, Optional, Tuple
  9. import lc.config as c
  10. import lc.error as e
  11. import lc.request as r
  12. import lc.view as v
  13. class Model(peewee.Model):
  14. class Meta:
  15. database = c.app.db
  16. def to_dict(self) -> dict:
  17. return playhouse.shortcuts.model_to_dict(self)
  18. @contextmanager
  19. def atomic(self):
  20. with c.app.db.atomic():
  21. yield
  22. class User(Model):
  23. """
  24. A user! you know tf this is about
  25. """
  26. name = peewee.TextField(unique=True)
  27. passhash = peewee.TextField()
  28. is_admin = peewee.BooleanField(default=False)
  29. @staticmethod
  30. def from_request(user: r.User) -> "User":
  31. passhash = pwd.hash(user.password)
  32. try:
  33. return User.create(name=user.name, passhash=passhash,)
  34. except peewee.IntegrityError:
  35. raise e.UserExists(name=user.name)
  36. def change_password(self, req: r.PasswordChange):
  37. if not pwd.verify(req.old, self.passhash):
  38. raise e.BadPassword(name=self.name)
  39. self.passhash = pwd.hash(req.n1)
  40. self.save()
  41. @staticmethod
  42. def from_invite(user: r.User, token: str) -> "User":
  43. invite = UserInvite.by_code(token)
  44. if invite.claimed_by is not None or invite.claimed_at is not None:
  45. raise e.AlreadyUsedInvite(invite=token)
  46. u = User.from_request(user)
  47. invite.claimed_at = datetime.datetime.now()
  48. invite.claimed_by = u
  49. invite.save()
  50. return u
  51. def authenticate(self, password: str) -> bool:
  52. return pwd.verify(password, self.passhash)
  53. def set_as_admin(self):
  54. self.is_admin = True
  55. self.save()
  56. @staticmethod
  57. def login(user: r.User) -> Tuple["User", str]:
  58. u = User.by_slug(user.name)
  59. if not u.authenticate(user.password):
  60. raise e.BadPassword(name=user.name)
  61. return u, user.to_token()
  62. @staticmethod
  63. def by_slug(slug: str) -> "User":
  64. u = User.get_or_none(name=slug)
  65. if u is None:
  66. raise e.NoSuchUser(name=slug)
  67. return u
  68. def base_url(self) -> str:
  69. return f"/u/{self.name}"
  70. def config_url(self) -> str:
  71. return f"/u/{self.name}/config"
  72. def get_links(
  73. self, as_user: Optional["User"], page: int
  74. ) -> Tuple[List[v.Link], v.Pagination]:
  75. query = Link.select().where(
  76. (Link.user == self) & ((self == as_user) | (Link.private == False))
  77. )
  78. links = query.order_by(-Link.created).paginate(page, c.app.per_page)
  79. link_views = [l.to_view(as_user) for l in links]
  80. pagination = v.Pagination.from_total(page, query.count())
  81. return link_views, pagination
  82. def get_link(self, link_id: int) -> "Link":
  83. try:
  84. return Link.get((Link.user == self) & (Link.id == link_id))
  85. except Link.DoesNotExist:
  86. raise e.NoSuchLink(link_id)
  87. def get_tag(self, tag_name: str) -> "Tag":
  88. return Tag.get((Tag.user == self) & (Tag.name == tag_name))
  89. def to_dict(self) -> dict:
  90. return {"id": self.id, "name": self.name}
  91. def get_config(self, status_msg: Optional[int]) -> v.Config:
  92. admin_pane = None
  93. if self.is_admin:
  94. user_invites = [
  95. v.UserInvite(
  96. claimed=ui.claimed_by is not None,
  97. claimant=ui.claimed_by and ui.claimed_by.name,
  98. token=ui.token,
  99. )
  100. for ui in UserInvite.select().where(UserInvite.created_by == self)
  101. ]
  102. admin_pane = v.AdminPane(invites=user_invites)
  103. return v.Config(username=self.name, admin_pane=admin_pane, msg=status_msg)
  104. def import_pinboard_data(self, stream):
  105. try:
  106. links = json.load(stream)
  107. except json.decoder.JSONDecodeError as exn:
  108. raise e.BadFileUpload("could not parse file as JSON")
  109. if not isinstance(links, list):
  110. raise e.BadFileUpload(f"expected a list")
  111. # create and (for this session) cache the tags
  112. tags = {}
  113. for l in links:
  114. if "tags" not in l:
  115. raise e.BadFileUpload("missing key {exn.args[0]}")
  116. for t in l["tags"].split():
  117. if t in tags:
  118. continue
  119. tags[t] = Tag.get_or_create_tag(self, t)
  120. with self.atomic():
  121. for l in links:
  122. try:
  123. time = datetime.datetime.strptime(l["time"], "%Y-%m-%dT%H:%M:%SZ")
  124. ln = Link.create(
  125. url=l["href"],
  126. name=l["description"],
  127. description=l["extended"],
  128. private=l["shared"] == "no",
  129. created=time,
  130. user=self,
  131. )
  132. except KeyError as exn:
  133. raise e.BadFileUpload(f"missing key {exn.args[0]}")
  134. for t in l["tags"].split():
  135. HasTag.get_or_create(link=ln, tag=tags[t])
  136. def get_tags(self) -> List[v.Tag]:
  137. return sorted(
  138. (t.to_view() for t in self.tags), # type: ignore
  139. key=lambda t: t.name,
  140. )
  141. def get_related_tags(self, tag: "Tag") -> List[v.Tag]:
  142. # SELECT * from has_tag t1, has_tag t2, link l WHERE t1.link_id == l.id AND t2.link_id == l.id AND t1.id != t2.id AND t1 = self
  143. SelfTag = HasTag.alias()
  144. query = (
  145. HasTag.select(HasTag.tag)
  146. .join(Link, on=(HasTag.link == Link.id))
  147. .join(SelfTag, on=(SelfTag.link == Link.id))
  148. .where((SelfTag.tag == tag) & (SelfTag.id != HasTag.id))
  149. .group_by(HasTag.tag)
  150. )
  151. return sorted((t.tag.to_view() for t in query), key=lambda t: t.name,)
  152. def get_string_search(
  153. self, needle: str, as_user: Optional["User"], page: int
  154. ) -> Tuple[List[v.Link], v.Pagination]:
  155. """
  156. Find all links that contain the string `needle` somewhere in their URL or title
  157. """
  158. query = Link.select().where(
  159. (Link.user == self)
  160. & ((self == as_user) | (Link.private == False))
  161. & (Link.name.contains(needle) | Link.description.contains(needle))
  162. )
  163. links = query.order_by(-Link.created).paginate(page, c.app.per_page)
  164. link_views = [l.to_view(as_user) for l in links]
  165. pagination = v.Pagination.from_total(page, query.count())
  166. return link_views, pagination
  167. class Link(Model):
  168. """
  169. A link as stored in the database
  170. """
  171. url = peewee.TextField()
  172. name = peewee.TextField()
  173. description = peewee.TextField()
  174. # TODO: do we need to track modified time?
  175. created = peewee.DateTimeField()
  176. # is the field entirely private?
  177. private = peewee.BooleanField()
  178. # owned by
  179. user = peewee.ForeignKeyField(User, backref="links")
  180. def link_url(self) -> str:
  181. return f"/u/{self.user.name}/l/{self.id}"
  182. @staticmethod
  183. def by_id(id: int) -> Optional["Link"]:
  184. return Link.get_or_none(id=id)
  185. @staticmethod
  186. def get_all(
  187. as_user: Optional[User], page: int
  188. ) -> Tuple[List["Link"], v.Pagination]:
  189. links = (
  190. Link.select()
  191. .where((Link.user == as_user) | (Link.private == False))
  192. .order_by(-Link.created)
  193. .paginate(page, c.app.per_page)
  194. )
  195. link_views = [l.to_view(as_user) for l in links]
  196. pagination = v.Pagination.from_total(page, Link.select().count())
  197. return link_views, pagination
  198. @staticmethod
  199. def from_request(user: User, link: r.Link) -> "Link":
  200. l = Link.create(
  201. url=link.url,
  202. name=link.name,
  203. description=link.description,
  204. private=link.private,
  205. created=link.created or datetime.datetime.now(),
  206. user=user,
  207. )
  208. for tag_name in link.tags:
  209. tag = Tag.get_or_create_tag(user, tag_name)
  210. HasTag.get_or_create(link=l, tag=tag)
  211. return l
  212. def update_from_request(self, user: User, link: r.Link):
  213. with self.atomic():
  214. req_tags = set(link.tags)
  215. for hastag in self.tags: # type: ignore
  216. name = hastag.tag.name
  217. if name not in req_tags:
  218. hastag.delete_instance()
  219. else:
  220. req_tags.remove(name)
  221. for tag_name in req_tags:
  222. t = Tag.get_or_create_tag(user, tag_name)
  223. HasTag.get_or_create(link=self, tag=t)
  224. Tag.clean()
  225. self.url = link.url
  226. self.name = link.name
  227. self.description = link.description
  228. self.private = link.private
  229. self.save()
  230. def to_view(self, as_user: Optional[User]) -> v.Link:
  231. return v.Link(
  232. id=self.id,
  233. url=self.url,
  234. name=self.name,
  235. description=self.description,
  236. private=self.private,
  237. tags=self.get_tags_view(),
  238. created=self.created,
  239. is_mine=self.user.id == as_user.id if as_user else False,
  240. link_url=self.link_url(),
  241. user=self.user.name,
  242. )
  243. def get_tags_view(self) -> List[v.Tag]:
  244. return [
  245. v.Tag(url=f"/u/{self.user.name}/t/{ht.tag.name}", name=ht.tag.name)
  246. for ht in HasTag().select(Tag.name).join(Tag).where(HasTag.link == self)
  247. ]
  248. class Tag(Model):
  249. """
  250. A tag. This just indicates that a user has used this tag at some point.
  251. """
  252. name = peewee.TextField()
  253. parent = peewee.ForeignKeyField("self", null=True, backref="children")
  254. user = peewee.ForeignKeyField(User, backref="tags")
  255. def url(self) -> str:
  256. return f"/u/{self.user.name}/t/{self.name}"
  257. def get_links(
  258. self, as_user: Optional[User], page: int
  259. ) -> Tuple[List[Link], v.Pagination]:
  260. query = (
  261. HasTag.select()
  262. .join(Link)
  263. .where(
  264. (HasTag.tag == self)
  265. & ((HasTag.link.user == as_user) | (HasTag.link.private == False))
  266. )
  267. )
  268. links = [
  269. ht.link.to_view(as_user)
  270. for ht in query.order_by(-Link.created).paginate(page, c.app.per_page)
  271. ]
  272. pagination = v.Pagination.from_total(page, query.count())
  273. return links, pagination
  274. def get_family(self) -> Iterator["Tag"]:
  275. yield self
  276. p = self
  277. while (p := p.parent) :
  278. yield p
  279. BAD_TAG_CHARS = set("{}[]\\()#?")
  280. @staticmethod
  281. def is_valid_tag_name(tag_name: str) -> bool:
  282. return all((c not in Tag.BAD_TAG_CHARS for c in tag_name))
  283. @staticmethod
  284. def get_or_create_tag(user: User, tag_name: str) -> "Tag":
  285. if (t := Tag.get_or_none(name=tag_name, user=user)) :
  286. return t
  287. if not Tag.is_valid_tag_name(tag_name):
  288. raise e.BadTagName(tag_name)
  289. parent = None
  290. if "/" in tag_name:
  291. parent_name = tag_name[: tag_name.rindex("/")]
  292. parent = Tag.get_or_create_tag(user, parent_name)
  293. return Tag.create(name=tag_name, parent=parent, user=user)
  294. def to_view(self) -> v.Tag:
  295. return v.Tag(url=self.url(), name=self.name)
  296. @staticmethod
  297. def clean():
  298. unused = (
  299. Tag.select(Tag.id) # type: ignore
  300. .join(HasTag, peewee.JOIN.LEFT_OUTER)
  301. .group_by(Tag.name)
  302. .having(peewee.fn.COUNT(HasTag.id) == 0)
  303. )
  304. Tag.delete().where(Tag.id.in_(unused)).execute() # type: ignore
  305. class HasTag(Model):
  306. """
  307. Establishes that a link is tagged with a given tag.
  308. """
  309. link = peewee.ForeignKeyField(Link, backref="tags")
  310. tag = peewee.ForeignKeyField(Tag, backref="models")
  311. @staticmethod
  312. def get_or_create(link: Link, tag: Tag):
  313. res = HasTag.get_or_none(link=link, tag=tag)
  314. if res is None:
  315. res = HasTag.create(link=link, tag=tag)
  316. if tag.parent:
  317. HasTag.get_or_create(link, tag.parent)
  318. return res
  319. class UserInvite(Model):
  320. token = peewee.TextField(unique=True)
  321. created_by = peewee.ForeignKeyField(User, backref="invites")
  322. created_at = peewee.DateTimeField()
  323. claimed_by = peewee.ForeignKeyField(User, null=True)
  324. claimed_at = peewee.DateTimeField(null=True)
  325. @staticmethod
  326. def by_code(token: str) -> "UserInvite":
  327. if (u := UserInvite.get_or_none(token=token)) :
  328. return u
  329. raise e.NoSuchInvite(invite=token)
  330. @staticmethod
  331. def manufacture(creator: User) -> "UserInvite":
  332. now = datetime.datetime.now()
  333. token = c.app.serialize_token(
  334. {"created_at": now.timestamp(), "created_by": creator.name,}
  335. )
  336. return UserInvite.create(
  337. token=token,
  338. created_by=creator,
  339. created_at=now,
  340. claimed_by=None,
  341. claimed_at=None,
  342. )
  343. MODELS = [
  344. User,
  345. Link,
  346. Tag,
  347. HasTag,
  348. UserInvite,
  349. ]
  350. def create_tables():
  351. c.app.db.create_tables(MODELS, safe=True)