model.py 8.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288
  1. from dataclasses import dataclass
  2. import datetime
  3. from passlib.apps import custom_app_context as pwd
  4. import peewee
  5. import playhouse.shortcuts
  6. from typing import List, Optional, Tuple
  7. import lc.config as c
  8. import lc.error as e
  9. import lc.request as r
  10. import lc.view as v
  11. class Model(peewee.Model):
  12. class Meta:
  13. database = c.db
  14. def to_dict(self) -> dict:
  15. return playhouse.shortcuts.model_to_dict(self)
  16. class User(Model):
  17. """
  18. A user! you know tf this is about
  19. """
  20. name = peewee.TextField(unique=True)
  21. passhash = peewee.TextField()
  22. is_admin = peewee.BooleanField(default=False)
  23. @staticmethod
  24. def from_request(user: r.User) -> "User":
  25. passhash = pwd.hash(user.password)
  26. try:
  27. return User.create(name=user.name, passhash=passhash,)
  28. except peewee.IntegrityError:
  29. raise e.UserExists(name=user.name)
  30. @staticmethod
  31. def from_invite(user: r.User, token: str) -> "User":
  32. invite = UserInvite.by_code(token)
  33. if invite.claimed_by is not None or invite.claimed_at is not None:
  34. raise e.AlreadyUsedInvite(invite=token)
  35. u = User.from_request(user)
  36. invite.claimed_at = datetime.datetime.now()
  37. invite.claimed_by = u
  38. invite.save()
  39. return u
  40. def authenticate(self, password: str) -> bool:
  41. return pwd.verify(password, self.passhash)
  42. def set_as_admin(self):
  43. self.is_admin = True
  44. self.save()
  45. @staticmethod
  46. def login(user: r.User) -> Tuple["User", str]:
  47. u = User.by_slug(user.name)
  48. if not u.authenticate(user.password):
  49. raise e.BadPassword(name=user.name)
  50. return u, user.to_token()
  51. @staticmethod
  52. def by_slug(slug: str) -> "User":
  53. u = User.get_or_none(name=slug)
  54. if u is None:
  55. raise e.NoSuchUser(name=slug)
  56. return u
  57. def base_url(self) -> str:
  58. return f"/u/{self.name}"
  59. def get_links(self, as_user: r.User, page: int) -> Tuple[v.Link, v.Pagination]:
  60. links = (
  61. Link.select()
  62. .where((Link.user == self) & ((self == as_user) | (Link.private == False)))
  63. .order_by(-Link.created)
  64. .paginate(page, c.per_page)
  65. )
  66. link_views = [l.to_view(as_user) for l in links]
  67. pagination = v.Pagination.from_total(page, Link.select().count())
  68. return link_views, pagination
  69. def get_link(self, link_id: int) -> "Link":
  70. return Link.get((Link.user == self) & (Link.id == link_id))
  71. def get_tag(self, tag_name: str) -> "Tag":
  72. return Tag.get((Tag.user == self) & (Tag.name == tag_name))
  73. def to_dict(self) -> dict:
  74. return {"id": self.id, "name": self.name}
  75. def get_config(self) -> v.Config:
  76. admin_pane = None
  77. if self.is_admin:
  78. user_invites = [
  79. v.UserInvite(
  80. claimed=ui.claimed_by is not None,
  81. claimant=ui.claimed_by and ui.claimed_by.name,
  82. token=ui.token,
  83. )
  84. for ui in UserInvite.select().where(UserInvite.created_by == self)
  85. ]
  86. admin_pane = v.AdminPane(invites=user_invites)
  87. return v.Config(username=self.name, admin_pane=admin_pane,)
  88. class Link(Model):
  89. """
  90. A link as stored in the database
  91. """
  92. url = peewee.TextField()
  93. name = peewee.TextField()
  94. description = peewee.TextField()
  95. # TODO: do we need to track modified time?
  96. created = peewee.DateTimeField()
  97. # is the field entirely private?
  98. private = peewee.BooleanField()
  99. # owned by
  100. user = peewee.ForeignKeyField(User, backref="links")
  101. def link_url(self) -> str:
  102. return f"/u/{self.user.name}/l/{self.id}"
  103. @staticmethod
  104. def by_id(id: int) -> Optional["Link"]:
  105. return Link.get_or_none(id=id)
  106. @staticmethod
  107. def from_request(user: User, link: r.Link) -> "Link":
  108. l = Link.create(
  109. url=link.url,
  110. name=link.name,
  111. description=link.description,
  112. private=link.private,
  113. created=link.created or datetime.datetime.now(),
  114. user=user,
  115. )
  116. for tag_name in link.tags:
  117. t = Tag.get_or_create_tag(user, tag_name)
  118. HasTag.create(
  119. link=l, tag=t,
  120. )
  121. return l
  122. def update_from_request(self, user: User, link: r.Link):
  123. req_tags = set(link.tags)
  124. for hastag in self.tags:
  125. name = hastag.tag.name
  126. if name not in req_tags:
  127. hastag.delete_instance()
  128. else:
  129. req_tags.remove(name)
  130. for tag_name in req_tags:
  131. t = Tag.get_or_create_tag(user, tag_name)
  132. HasTag.get_or_create(link=self, tag=t)
  133. self.url = link.url
  134. self.name = link.name
  135. self.description = link.description
  136. self.private = link.private
  137. self.save()
  138. def to_view(self, as_user: User) -> v.Link:
  139. return v.Link(
  140. id=self.id,
  141. url=self.url,
  142. name=self.name,
  143. description=self.description,
  144. private=self.private,
  145. tags=[t.tag.to_view() for t in self.tags],
  146. created=self.created,
  147. is_mine=self.user.id == as_user.id,
  148. link_url=self.link_url(),
  149. )
  150. class Tag(Model):
  151. """
  152. A tag. This just indicates that a user has used this tag at some point.
  153. """
  154. name = peewee.TextField()
  155. parent = peewee.ForeignKeyField("self", null=True, backref="children")
  156. user = peewee.ForeignKeyField(User, backref="tags")
  157. def url(self) -> str:
  158. return f"/u/{self.user.name}/t/{self.name}"
  159. def get_links(self, as_user: r.User, page: int) -> Tuple[List[Link], v.Pagination]:
  160. links = [
  161. ht.link.to_view(as_user)
  162. for ht in HasTag.select()
  163. .join(Link)
  164. .where(
  165. (HasTag.tag == self)
  166. & ((HasTag.link.user == as_user) | (HasTag.link.private == False))
  167. )
  168. .order_by(-Link.created)
  169. .paginate(page, c.per_page)
  170. ]
  171. pagination = v.Pagination.from_total(
  172. page, HasTag.select().where((HasTag.tag == self)).count(),
  173. )
  174. return links, pagination
  175. @staticmethod
  176. def get_or_create_tag(user: User, tag_name: str):
  177. if (t := Tag.get_or_none(name=tag_name, user=user)) :
  178. return t
  179. parent = None
  180. if "/" in tag_name:
  181. parent_name = tag_name[: tag_name.rindex("/")]
  182. parent = Tag.get_or_create_tag(user, parent_name)
  183. return Tag.create(name=tag_name, parent=parent, user=user)
  184. def to_view(self) -> v.Tag:
  185. return v.Tag(url=self.url(), name=self.name)
  186. class HasTag(Model):
  187. """
  188. Establishes that a link is tagged with a given tag.
  189. """
  190. link = peewee.ForeignKeyField(Link, backref="tags")
  191. tag = peewee.ForeignKeyField(Tag, backref="models")
  192. @staticmethod
  193. def get_or_create(link: Link, tag: Tag):
  194. res = HasTag.get_or_none(link=link, tag=tag)
  195. if res is None:
  196. res = HasTag.create(link=link, tag=tag)
  197. return res
  198. class UserInvite(Model):
  199. token = peewee.TextField(unique=True)
  200. created_by = peewee.ForeignKeyField(User, backref="invites")
  201. created_at = peewee.DateTimeField()
  202. claimed_by = peewee.ForeignKeyField(User, null=True)
  203. claimed_at = peewee.DateTimeField(null=True)
  204. @staticmethod
  205. def by_code(token: str) -> "UserInvite":
  206. if (u := UserInvite.get_or_none(token=token)) :
  207. return u
  208. raise e.NoSuchInvite(invite=token)
  209. @staticmethod
  210. def manufacture(creator: User) -> "UserInvite":
  211. now = datetime.datetime.now()
  212. token = c.serializer.dumps(
  213. {"created_at": now.timestamp(), "created_by": creator.name,}
  214. )
  215. return UserInvite.create(
  216. token=token,
  217. created_by=creator,
  218. created_at=now,
  219. claimed_by=None,
  220. claimed_at=None,
  221. )
  222. MODELS = [
  223. User,
  224. Link,
  225. Tag,
  226. HasTag,
  227. UserInvite,
  228. ]
  229. def create_tables():
  230. c.db.create_tables(MODELS, safe=True)