model.py 11 KB

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