model.py 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200
  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. class Model(peewee.Model):
  11. class Meta:
  12. database = c.db
  13. def to_dict(self) -> dict:
  14. return playhouse.shortcuts.model_to_dict(self)
  15. @dataclass
  16. class Pagination:
  17. current: int
  18. last: int
  19. def previous(self) -> Optional[dict]:
  20. if self.current > 1:
  21. return {"page": self.current - 1}
  22. return None
  23. def next(self) -> Optional[dict]:
  24. if self.current < self.last:
  25. return {"page": self.current + 1}
  26. return None
  27. @classmethod
  28. def from_total(cls, current, total) -> "Pagination":
  29. return cls(current=current, last=((total - 1) // c.per_page) + 1,)
  30. # TODO: figure out authorization for users (oauth? passwd?)
  31. class User(Model):
  32. """
  33. A user! you know tf this is about
  34. """
  35. name = peewee.TextField(unique=True)
  36. passhash = peewee.TextField()
  37. is_admin = peewee.BooleanField(default=False)
  38. @staticmethod
  39. def from_request(user: r.User) -> "User":
  40. passhash = pwd.hash(user.password)
  41. try:
  42. return User.create(name=user.name, passhash=passhash,)
  43. except peewee.IntegrityError:
  44. raise e.UserExists(name=user.name)
  45. def authenticate(self, password: str) -> bool:
  46. return pwd.verify(password, self.passhash)
  47. def set_as_admin(self):
  48. self.is_admin = True
  49. self.save()
  50. @staticmethod
  51. def login(user: r.User) -> Tuple["User", str]:
  52. u = User.by_slug(user.name)
  53. if not u.authenticate(user.password):
  54. raise e.BadPassword(name=user.name)
  55. return u, user.to_token()
  56. @staticmethod
  57. def by_slug(slug: str) -> "User":
  58. u = User.get_or_none(name=slug)
  59. if u is None:
  60. raise e.NoSuchUser(name=slug)
  61. return u
  62. def base_url(self) -> str:
  63. return f"/u/{self.name}"
  64. def get_links(self, page: int) -> Tuple[List["Link"], Pagination]:
  65. links = (
  66. Link.select()
  67. .where(Link.user == self)
  68. .order_by(-Link.created)
  69. .paginate(page, c.per_page)
  70. )
  71. pagination = Pagination.from_total(page, Link.select().count())
  72. return links, pagination
  73. def get_link(self, link_id: int) -> "Link":
  74. return Link.get((Link.user == self) & (Link.id == link_id))
  75. def get_tag(self, tag_name: str) -> "Tag":
  76. return Tag.get((Tag.user == self) & (Tag.name == tag_name))
  77. def to_dict(self) -> dict:
  78. return {"id": self.id, "name": self.name}
  79. class Link(Model):
  80. """
  81. A link as stored in the database
  82. """
  83. url = peewee.TextField()
  84. name = peewee.TextField()
  85. description = peewee.TextField()
  86. # TODO: do we need to track modified time?
  87. created = peewee.DateTimeField()
  88. # is the field entirely private?
  89. private = peewee.BooleanField()
  90. # owned by
  91. user = peewee.ForeignKeyField(User, backref="links")
  92. def link_url(self) -> str:
  93. return f"/u/{self.user.name}/l/{self.id}"
  94. @staticmethod
  95. def from_request(user: User, link: r.Link) -> "Link":
  96. l = Link.create(
  97. url=link.url,
  98. name=link.name,
  99. description=link.description,
  100. private=link.private,
  101. created=link.created or datetime.datetime.now(),
  102. user=user,
  103. )
  104. for tag_name in link.tags:
  105. t = Tag.get_or_create_tag(user, tag_name)
  106. HasTag.create(
  107. link=l, tag=t,
  108. )
  109. return l
  110. class Tag(Model):
  111. """
  112. A tag. This just indicates that a user has used this tag at some point.
  113. """
  114. name = peewee.TextField()
  115. parent = peewee.ForeignKeyField("self", null=True, backref="children")
  116. user = peewee.ForeignKeyField(User, backref="tags")
  117. def url(self) -> str:
  118. return f"/u/{self.user.name}/t/{self.name}"
  119. def get_links(self, page: int) -> Tuple[List[Link], Pagination]:
  120. links = [
  121. ht.link
  122. for ht in HasTag.select()
  123. .join(Link)
  124. .where((HasTag.tag == self))
  125. .order_by(-Link.created)
  126. .paginate(page, c.per_page)
  127. ]
  128. pagination = Pagination.from_total(
  129. page, HasTag.select().where((HasTag.tag == self)).count(),
  130. )
  131. return links, pagination
  132. @staticmethod
  133. def get_or_create_tag(user: User, tag_name: str):
  134. if (t := Tag.get_or_none(name=tag_name, user=user)) :
  135. return t
  136. parent = None
  137. if "/" in tag_name:
  138. parent_name = tag_name[: tag_name.rindex("/")]
  139. parent = Tag.get_or_create_tag(user, parent_name)
  140. return Tag.create(name=tag_name, parent=parent, user=user)
  141. class HasTag(Model):
  142. """
  143. Establishes that a link is tagged with a given tag.
  144. """
  145. link = peewee.ForeignKeyField(Link, backref="tags")
  146. tag = peewee.ForeignKeyField(Tag, backref="models")
  147. class UserInvite(Model):
  148. token: str
  149. MODELS = [
  150. User,
  151. Link,
  152. Tag,
  153. HasTag,
  154. ]
  155. def create_tables():
  156. c.db.create_tables(MODELS, safe=True)