vtt/src/ttfrog/schema.py
2025-10-08 00:46:09 -07:00

189 lines
6.0 KiB
Python

from __future__ import annotations
from datetime import datetime
from typing import List
from enum import StrEnum
from grung.types import BackReference, Collection, DateTime, Dict, Field, Password, Pointer, Record, Timestamp
from tinydb import where
class Permissions(StrEnum):
READ = "r"
WRITE = "w"
DELETE = "d"
class Page(Record):
"""
A page in the wiki. Just about everything in the databse is either a Page or a subclass of a Page.
"""
@classmethod
def fields(cls):
# fmt: off
return [
*super().fields(),
Field("uri", unique=True), # The URI for the page, relative to the app's VIEW_URI
Field("name"), # The portion of the URI after the last /
Field("title"), # The page title
Field("body"), # The main content blob of the page
Collection("members", Page), # The pages that exist below this page's URI
BackReference("parent", value_type=Page), # The page that exists above this page's URI
Pointer("author", value_type=User), # The last user to touch the page.
DateTime("created"), # When the page was created
Timestamp("last_modified"), # The last time the page was modified.
Dict("acl"),
]
# fmt: on
def before_insert(self, db):
"""
Make the following adjustments before saving this record:
* Derive the name from the title, or the title from the name
* Derive the URI from the hierarchy of the parent.
"""
super().before_insert(db)
now = datetime.utcnow()
if not self.doc_id and self.created < now:
self.created = now
if not self.name and not self.title:
raise Exception("Must provide either a name or a title!")
if not self.name:
self.name = self.title.title().replace(" ", "")
if not self.title:
self.title = self.name
self.uri = (self.parent.uri + "/" if self.parent and self.parent.uri != "/" else "") + self.name
def after_insert(self, db):
"""
After saving this record, ensure that any page in the members collection is updated with the
correct URI. This ensures that if a page is moved from one collection to another, the URI is updated.
"""
super().after_insert(db)
if not hasattr(self, "members"):
return
for child in self.members:
obj = BackReference.dereference(child, db)
obj.uri = f"{self.uri}/{obj.name}"
child = db.save(obj)
def add_member(self, child: Record):
from ttfrog import app
app.check_state()
self.members = list(set(self.members + [app.db.save(child)]))
app.db.save(self)
return self.get_child(child)
def get_child(self, obj: Record):
for page in self.members:
if page[page._metadata.primary_key] == obj[obj._metadata.primary_key]:
return page
return None
def set_permissions(self, entity: Entity, permissions: List) -> str:
from ttfrog import app
app.check_state()
perms = "".join(permissions)
self.acl[entity.reference] = perms
app.db.save(self)
return perms
def get_acl_for_entity(self, entity) -> (str, str | None):
"""
Search upward through the page hierarchy looking for one with an ACL that either
has a grant for the entity we care about, or at least one group in which the entity is a member.
"""
from ttfrog import app
app.check_state()
def find_acl(obj):
if hasattr(obj, "acl"):
if entity.reference in obj.acl:
return {entity.reference: obj.acl[entity.reference]}
group_grants = {}
for ref, grant in obj.acl.items():
(table_name, pkey, pval) = ref.split("::")
if table_name == "Group":
group = app.db.Group.get(where(pkey) == pval, recurse=False)
if entity.reference in group.members:
group_grants[ref] = grant
if group_grants:
return group_grants
if hasattr(obj, "parent"):
return find_acl(obj.parent)
return {"": ""}
return find_acl(self)
class Entity(Page):
@classmethod
def fields(cls):
inherited = [
field
for field in super().fields()
if field.name not in ("members", "uid")
]
return inherited + [
Field("name", primary_key=True),
]
def has_permission(self, record: Record, requested: str) -> bool | None:
for entity, grants in record.get_acl_for_entity(self).items():
if requested in grants:
return True
return False
def can_read(self, record: Record):
return self.has_permission(record, Permissions.READ)
def can_write(self, record: Record):
return self.has_permission(record, Permissions.WRITE)
def can_delete(self, record: Record):
return self.has_permission(record, Permissions.DELETE)
class User(Entity):
"""
A website user, editable as a wiki page.
"""
@classmethod
def fields(cls):
return super().fields() + [
Field("email", unique=True),
Password("password"),
]
def check_credentials(self, username: str, password: str) -> bool:
return username == self.name and self._metadata.fields["password"].compare(password, self.password)
class Group(Entity):
"""
A set of users, editable as a wiki page.
"""
@classmethod
def fields(cls):
return super().fields() + [
Collection("members", Entity)
]
class NPC(Page):
"""
An NPC, editable as a wiki page.
"""