model.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147
  1. import datetime
  2. from passlib.apps import custom_app_context as pwd
  3. import peewee
  4. import typing
  5. import lc.config as c
  6. import lc.error as e
  7. import lc.request as r
  8. class Model(peewee.Model):
  9. class Meta:
  10. database = c.DB
  11. # TODO: figure out authorization for users (oauth? passwd?)
  12. class User(Model):
  13. """
  14. A user! you know tf this is about
  15. """
  16. name = peewee.TextField(unique=True)
  17. passhash = peewee.TextField()
  18. @staticmethod
  19. def from_request(user: r.User) -> "User":
  20. passhash = pwd.hash(user.password)
  21. try:
  22. return User.create(name=user.name, passhash=passhash,)
  23. except peewee.IntegrityError:
  24. raise e.UserExists(name=user.name)
  25. def authenticate(self, password: str) -> bool:
  26. return pwd.verify(password, self.passhash)
  27. @staticmethod
  28. def login(user: r.User) -> "User":
  29. u = User.by_slug(user.name)
  30. if not u.authenticate(user.password):
  31. raise e.BadPassword(name=user.name)
  32. return u
  33. @staticmethod
  34. def by_slug(slug: str) -> "User":
  35. u = User.get_or_none(name=slug)
  36. if u is None:
  37. raise e.NoSuchUser(name=slug)
  38. return u
  39. def base_url(self) -> str:
  40. return f"/u/{self.name}"
  41. def get_links(self, page: int) -> typing.List["Link"]:
  42. return Link.select().where(Link.user == self).paginate(page, c.PER_PAGE)
  43. def get_tag(self, tag_name: str) -> "Tag":
  44. return Tag.get((Tag.user == self) & (Tag.name == tag_name))
  45. class Link(Model):
  46. """
  47. A link as stored in the database
  48. """
  49. url = peewee.TextField()
  50. name = peewee.TextField()
  51. description = peewee.TextField()
  52. # TODO: do we need to track modified time?
  53. created = peewee.DateTimeField()
  54. # is the field entirely private?
  55. private = peewee.BooleanField()
  56. # owned by
  57. user = peewee.ForeignKeyField(User, backref="links")
  58. def link_url(self) -> str:
  59. return f"/u/{self.user.name}/l/{self.id}"
  60. @staticmethod
  61. def from_request(user: User, link: r.Link) -> "Link":
  62. l = Link.create(
  63. url=link.url,
  64. name=link.name,
  65. description=link.description,
  66. private=link.private,
  67. created=datetime.datetime.now(),
  68. user=user,
  69. )
  70. for tag_name in link.tags:
  71. t = Tag.get_or_create_tag(user, tag_name)
  72. HasTag.create(
  73. link=l, tag=t,
  74. )
  75. return l
  76. class Tag(Model):
  77. """
  78. A tag. This just indicates that a user has used this tag at some point.
  79. """
  80. name = peewee.TextField()
  81. parent = peewee.ForeignKeyField("self", null=True, backref="children")
  82. user = peewee.ForeignKeyField(User, backref="tags")
  83. def url(self) -> str:
  84. return f"/u/{self.user.name}/t/{self.name}"
  85. def get_links(self, page: int) -> typing.List[Link]:
  86. return [
  87. ht.link
  88. for ht in HasTag.select()
  89. .where((HasTag.tag == self))
  90. .paginate(page, c.PER_PAGE)
  91. ]
  92. @staticmethod
  93. def get_or_create_tag(user: User, tag_name: str):
  94. if (t := Tag.get_or_none(name=tag_name, user=user)) :
  95. return t
  96. parent = None
  97. if "/" in tag_name:
  98. parent_name = tag_name[: tag_name.rindex("/")]
  99. parent = Tag.get_or_create_tag(user, parent_name)
  100. return Tag.create(name=tag_name, parent=parent, user=user)
  101. class HasTag(Model):
  102. """
  103. Establishes that a link is tagged with a given tag.
  104. """
  105. link = peewee.ForeignKeyField(Link, backref="tags")
  106. tag = peewee.ForeignKeyField(Tag, backref="models")
  107. MODELS = [
  108. User,
  109. Link,
  110. Tag,
  111. HasTag,
  112. ]
  113. def create_tables():
  114. c.DB.create_tables(MODELS, safe=True)