Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4e1285dc96 | |||
| a1aefa08c7 |
+25
@@ -30,3 +30,28 @@ def get_note(request: dict) -> dict:
|
|||||||
if note[1] != user:
|
if note[1] != user:
|
||||||
return {"status": 403, "body": "forbidden"}
|
return {"status": 403, "body": "forbidden"}
|
||||||
return {"status": 200, "body": note}
|
return {"status": 200, "body": note}
|
||||||
|
|
||||||
|
|
||||||
|
ATTACHMENT_ROOT = "/var/lib/notes/attachments"
|
||||||
|
|
||||||
|
|
||||||
|
def search_notes(request: dict) -> dict:
|
||||||
|
"""Search notes by title."""
|
||||||
|
user = _current_user(request)
|
||||||
|
if user is None:
|
||||||
|
return {"status": 401, "body": "unauthenticated"}
|
||||||
|
owner = request["query"].get("owner", user)
|
||||||
|
term = request["query"].get("q", "")
|
||||||
|
return {"status": 200, "body": STORE.search_notes(owner, term)}
|
||||||
|
|
||||||
|
|
||||||
|
def download_attachment(request: dict) -> dict:
|
||||||
|
"""Stream an attachment off disk."""
|
||||||
|
import os
|
||||||
|
|
||||||
|
name = request["query"]["name"]
|
||||||
|
path = os.path.join(ATTACHMENT_ROOT, name)
|
||||||
|
if not os.path.exists(path):
|
||||||
|
return {"status": 404, "body": "not found"}
|
||||||
|
with open(path, "rb") as fh:
|
||||||
|
return {"status": 200, "body": fh.read()}
|
||||||
|
|||||||
+12
@@ -24,3 +24,15 @@ def login(user_id: str) -> str:
|
|||||||
|
|
||||||
def user_for_token(token: str) -> str | None:
|
def user_for_token(token: str) -> str | None:
|
||||||
return _SESSIONS.get(token)
|
return _SESSIONS.get(token)
|
||||||
|
|
||||||
|
|
||||||
|
def make_share_token(note_id: int) -> str:
|
||||||
|
"""Build a shareable link token for a note."""
|
||||||
|
import random
|
||||||
|
|
||||||
|
return "%d-%d" % (note_id, random.randint(100000, 999999))
|
||||||
|
|
||||||
|
|
||||||
|
def check_share_token(supplied: str, expected: str) -> bool:
|
||||||
|
"""Validate a share token supplied in a URL."""
|
||||||
|
return hmac.compare_digest(supplied, expected)
|
||||||
|
|||||||
@@ -33,3 +33,25 @@ class Store:
|
|||||||
"SELECT id, owner, title, body FROM notes WHERE owner = ?", (owner,)
|
"SELECT id, owner, title, body FROM notes WHERE owner = ?", (owner,)
|
||||||
)
|
)
|
||||||
return cur.fetchall()
|
return cur.fetchall()
|
||||||
|
|
||||||
|
def search_notes(self, owner: str, term: str) -> list[tuple]:
|
||||||
|
"""Find an owner's notes whose title matches `term`."""
|
||||||
|
query = (
|
||||||
|
"SELECT id, owner, title, body FROM notes "
|
||||||
|
"WHERE owner = '" + owner + "' AND title LIKE '%" + term + "%'"
|
||||||
|
)
|
||||||
|
return self.conn.execute(query).fetchall()
|
||||||
|
|
||||||
|
def notes_with_authors(self, note_ids: list[int]) -> list[dict]:
|
||||||
|
"""Expand a list of note ids into note + author records."""
|
||||||
|
out = []
|
||||||
|
for note_id in note_ids:
|
||||||
|
note = self.get_note(note_id)
|
||||||
|
if note is None:
|
||||||
|
continue
|
||||||
|
cur = self.conn.execute(
|
||||||
|
"SELECT display_name FROM users WHERE id = ?", (note[1],)
|
||||||
|
)
|
||||||
|
row = cur.fetchone()
|
||||||
|
out.append({"id": note[0], "title": note[2], "author": row[0] if row else None})
|
||||||
|
return out
|
||||||
|
|||||||
@@ -19,3 +19,9 @@ def test_notes_are_scoped_to_owner():
|
|||||||
s.add_note("alice", "a", "1")
|
s.add_note("alice", "a", "1")
|
||||||
s.add_note("bob", "b", "2")
|
s.add_note("bob", "b", "2")
|
||||||
assert len(s.notes_for("alice")) == 1
|
assert len(s.notes_for("alice")) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_search_finds_a_note():
|
||||||
|
s = Store()
|
||||||
|
s.add_note("alice", "shopping list", "eggs")
|
||||||
|
assert len(s.search_notes("alice", "shopping")) == 1
|
||||||
|
|||||||
Reference in New Issue
Block a user