DocsUnit 3: users

Build a notes app with Nexy, part 11: give each user their own notes

Part 11 ties every note to the user who wrote it: the notes pages ask for a sign-in, each user sees and changes only their own notes, and the API answers 401 to a visitor.

Before you start: the app of part 10, with an account.

Tie a note to its user

A note gets the id of its user. foreign_key makes the database refuse an id that is not a user:

src/models/note_model.py

from sqlmodel import Field, SQLModel  class Note(SQLModel, table=True):    id: int | None = Field(default=None, primary_key=True)    text: str    user_id: int | None = Field(default=None, foreign_key="user.id", index=True)

src/apps/notes/note_model.py

from sqlmodel import Field, SQLModel  class Note(SQLModel, table=True):    id: int | None = Field(default=None, primary_key=True)    text: str    user_id: int | None = Field(default=None, foreign_key="user.id", index=True)

user_id may be empty, because the notes written before this part have no user. They stay in the table, and no one sees them any more.

nx migrate -m "notes belong to users"nx migrate --upgrade

Ask the database for the notes of one user

Every function now takes the id of the user, and filters on it. A note of someone else is answered as if it did not exist, so its id tells a curious visitor nothing:

src/notes.py

from sqlmodel import Session, select from src.configs.db import enginefrom src.models.note_model import Note PAGE_SIZE = 5  def notes_page(user_id: int, number: int) -> tuple[list[Note], bool]:    """One page of the notes of *user_id*, oldest first, and whether a page comes after it."""    query = select(Note).where(Note.user_id == user_id).order_by(Note.id)    with Session(engine) as session:        found = list(session.exec(query.offset((number - 1) * PAGE_SIZE).limit(PAGE_SIZE + 1)))    return found[:PAGE_SIZE], len(found) > PAGE_SIZE  def all_notes(user_id: int) -> list[Note]:    with Session(engine) as session:        return list(session.exec(select(Note).where(Note.user_id == user_id).order_by(Note.id)))  def find(user_id: int, note_id: int) -> Note | None:    """The note, when it belongs to *user_id*: someone else's note is not found."""    with Session(engine) as session:        note = session.get(Note, note_id)    return note if note is not None and note.user_id == user_id else None  def add_note(user_id: int, text: str) -> Note:    note = Note(text=text, user_id=user_id)    with Session(engine) as session:        session.add(note)        session.commit()        session.refresh(note)    return note  def change_note(user_id: int, note_id: int, text: str) -> bool:    with Session(engine) as session:        note = session.get(Note, note_id)        if note is None or note.user_id != user_id:            return False        note.text = text        session.commit()    return True  def remove_note(user_id: int, note_id: int) -> bool:    with Session(engine) as session:        note = session.get(Note, note_id)        if note is None or note.user_id != user_id:            return False        session.delete(note)        session.commit()    return True

src/notes.py

from sqlmodel import Session, select from src.apps.notes.note_model import Notefrom src.configs.db import engine PAGE_SIZE = 5  def notes_page(user_id: int, number: int) -> tuple[list[Note], bool]:    """One page of the notes of *user_id*, oldest first, and whether a page comes after it."""    query = select(Note).where(Note.user_id == user_id).order_by(Note.id)    with Session(engine) as session:        found = list(session.exec(query.offset((number - 1) * PAGE_SIZE).limit(PAGE_SIZE + 1)))    return found[:PAGE_SIZE], len(found) > PAGE_SIZE  def all_notes(user_id: int) -> list[Note]:    with Session(engine) as session:        return list(session.exec(select(Note).where(Note.user_id == user_id).order_by(Note.id)))  def find(user_id: int, note_id: int) -> Note | None:    """The note, when it belongs to *user_id*: someone else's note is not found."""    with Session(engine) as session:        note = session.get(Note, note_id)    return note if note is not None and note.user_id == user_id else None  def add_note(user_id: int, text: str) -> Note:    note = Note(text=text, user_id=user_id)    with Session(engine) as session:        session.add(note)        session.commit()        session.refresh(note)    return note  def change_note(user_id: int, note_id: int, text: str) -> bool:    with Session(engine) as session:        note = session.get(Note, note_id)        if note is None or note.user_id != user_id:            return False        note.text = text        session.commit()    return True  def remove_note(user_id: int, note_id: int) -> bool:    with Session(engine) as session:        note = session.get(Note, note_id)        if note is None or note.user_id != user_id:            return False        session.delete(note)        session.commit()    return True

One function reads the session, and sends anyone without a user to /signin:

src/guard.py

from nexy import redirect, useSession  def signed_in_user() -> int:    """The id of the signed-in user; a visitor is sent to the sign-in page."""    user_id = useSession().get("user_id")    if user_id is None:        redirect("/signin")    return user_id

redirect stops the page or the Action right there: nothing after it runs.

Guard the Actions

Each Action asks for the user first, so a form posted without a session does nothing:

src/actions/notes.py

from urllib.parse import urlencode from nexy import notFoundfrom nexy.decorators import Actionfrom nexy.http import RedirectResponsefrom src.guard import signed_in_userfrom src.notes import add_note, change_note, remove_note MAX_LENGTH = 200  def back(path: str = "/notes", error: str = "") -> RedirectResponse:    query = f"?{urlencode({'error': error})}" if error else ""    return RedirectResponse(f"{path}{query}", status_code=303)  def problem(text: str) -> str:    if not text:        return "A note cannot be blank."    if len(text) > MAX_LENGTH:        return f"A note is at most {MAX_LENGTH} characters."    return ""  @Actionasync def create_note(text: str) -> RedirectResponse:    user_id = signed_in_user()    text = text.strip()    if error := problem(text):        return back(error=error)    add_note(user_id, text)    return back()  @Actionasync def update_note(note_id: int, text: str) -> RedirectResponse:    user_id = signed_in_user()    text = text.strip()    if error := problem(text):        return back(f"/notes/{note_id}", error)    if not change_note(user_id, note_id, text):        notFound()    return back(f"/notes/{note_id}")  @Actionasync def delete_note(note_id: int) -> RedirectResponse:    if not remove_note(signed_in_user(), note_id):        notFound()    return back()

Guard the pages and the API

A page cannot send a visitor to a form, but an API can answer why it refuses: the API answers 401, the pages redirect.

src/routes/notes/index.nexy

---from "@/components/NoteCard.nexy" import NoteCardfrom nexy import useSearchParamsfrom src.guard import signed_in_userfrom src.notes import notes_page user_id = signed_in_user()params = useSearchParams()error = params.get("error", "")asked = params.get("page", "1")number = max(int(asked), 1) if asked.isdigit() else 1found = notes_page(user_id, number)notes = found[0]more = found[1]---<section class="app-page">  <h1>Notes</h1>   <form method="post" action="{{ __actions_path['sign_out'] }}">    <button type="submit">Sign out</button>  </form>   {% if error %}  <p class="note-error" role="alert">{{ error | e }}</p>  {% endif %}   <form method="post" action="{{ __actions_path['create_note'] }}">    <input name="text" placeholder="A new note" maxlength="200" required>    <button type="submit">Add</button>  </form>   {% for note in notes %}  <NoteCard id="{{ note.id }}" text="{{ note.text }}" />  {% else %}  <p>No notes yet: write the first one above.</p>  {% endfor %}   <nav class="note-pages" aria-label="Pages of notes">    {% if number > 1 %}<a href="/notes?page={{ number - 1 }}" rel="prev">Newer</a>{% endif %}    {% if more %}<a href="/notes?page={{ number + 1 }}" rel="next">Older</a>{% endif %}  </nav></section>

src/routes/notes/[id].nexy

---from nexy import notFound, useSearchParamsfrom src.guard import signed_in_userfrom src.notes import find id: prop[int]note = find(signed_in_user(), id)if note is None:    notFound()error = useSearchParams().get("error", "")---<section class="app-page">  <h1>{{ note.text | e }}</h1>   {% if error %}  <p class="note-error" role="alert">{{ error | e }}</p>  {% endif %}   <form method="post" action="{{ __actions_path['update_note'] }}">    <input type="hidden" name="note_id" value="{{ note.id }}">    <input name="text" value="{{ note.text | e }}" maxlength="200" required>    <button type="submit">Save</button>  </form>   <form method="post" action="{{ __actions_path['delete_note'] }}">    <input type="hidden" name="note_id" value="{{ note.id }}">    <button type="submit">Delete</button>  </form>   <a href="/notes">All notes</a></section>

src/routes/api/notes.py

from nexy import useSessionfrom nexy.http import HTTPExceptionfrom src.notes import all_notes  async def GET():    user_id = useSession().get("user_id")    if user_id is None:        raise HTTPException(status_code=401, detail="Sign in first")    return [note.model_dump() for note in all_notes(user_id)]

src/apps/notes/notes_view.nexy

---from "@/components/NoteCard.nexy" import NoteCardfrom nexy import useSearchParamsfrom src.guard import signed_in_userfrom src.notes import notes_page user_id = signed_in_user()params = useSearchParams()error = params.get("error", "")asked = params.get("page", "1")number = max(int(asked), 1) if asked.isdigit() else 1found = notes_page(user_id, number)notes = found[0]more = found[1]---<section class="app-page">  <h1>Notes</h1>   <form method="post" action="{{ __actions_path['sign_out'] }}">    <button type="submit">Sign out</button>  </form>   {% if error %}  <p class="note-error" role="alert">{{ error | e }}</p>  {% endif %}   <form method="post" action="{{ __actions_path['create_note'] }}">    <input name="text" placeholder="A new note" maxlength="200" required>    <button type="submit">Add</button>  </form>   {% for note in notes %}  <NoteCard id="{{ note.id }}" text="{{ note.text }}" />  {% else %}  <p>No notes yet: write the first one above.</p>  {% endfor %}   <nav class="note-pages" aria-label="Pages of notes">    {% if number > 1 %}<a href="/notes?page={{ number - 1 }}" rel="prev">Newer</a>{% endif %}    {% if more %}<a href="/notes?page={{ number + 1 }}" rel="next">Older</a>{% endif %}  </nav></section>

src/apps/notes/note_view.nexy

---from nexy import useSearchParams note: prop[object]error = useSearchParams().get("error", "")---<section class="app-page">  <h1>{{ note.text | e }}</h1>   {% if error %}  <p class="note-error" role="alert">{{ error | e }}</p>  {% endif %}   <form method="post" action="{{ __actions_path['update_note'] }}">    <input type="hidden" name="note_id" value="{{ note.id }}">    <input name="text" value="{{ note.text | e }}" maxlength="200" required>    <button type="submit">Save</button>  </form>   <form method="post" action="{{ __actions_path['delete_note'] }}">    <input type="hidden" name="note_id" value="{{ note.id }}">    <button type="submit">Delete</button>  </form>   <a href="/notes">All notes</a></section>

src/apps/notes/notes_controller.py

from nexy import notFound, useSessionfrom nexy.decorators import Controllerfrom nexy.hooks import useViewsfrom nexy.http import HTTPExceptionfrom src.guard import signed_in_userfrom src.notes import all_notes, find LAYOUT = "src/apps/layout.nexy"  def page(view: str, **context: object):    return useViews(LAYOUT, context={"children": useViews(view, context=context)})  @Controller("/notes")class NotesController:    async def get(self):        return page("src/apps/notes/notes_view.nexy")  @Controller("/notes/{id}")class NoteController:    async def get(self, id: int):        note = find(signed_in_user(), id)        if note is None:            notFound()        return page("src/apps/notes/note_view.nexy", note=note)  @Controller("/api/notes")class NotesApiController:    async def get(self) -> list[dict]:        user_id = useSession().get("user_id")        if user_id is None:            raise HTTPException(status_code=401, detail="Sign in first")        return [note.model_dump() for note in all_notes(user_id)]

Sign out, then open http://localhost:3000/notes: you land on the sign-in page. Sign in, and the list is empty until you write a note. Create a second account in a private window: it does not see the notes of the first.

What you learned

A foreign_key ties a note to a user, and every query filters on that user.
Someone else's note is answered 404, like a note that does not exist.
redirect() stops a page or an Action; an API answers 401 with HTTPException.

Next steps

Part 12: reset a forgotten password by email
Exceptions and errors: 401, 403, 404 and how each one is shown

Edit on GitHub