DocsUnidade 3: usuários

Crie um aplicativo de notas com Nexy, parte 11: dê a cada usuário suas próprias notas

A Parte 11 vincula cada nota ao usuário que a escreveu: as páginas de notas pedem um login, cada usuário vê e altera apenas suas próprias notas e a API responde 401 a um visitante.

Antes de começar: o aplicativo da parte 10, com uma conta.

Amarre uma nota ao seu usuário

Uma nota obtém o ID de seu usuário. foreign_key faz com que o banco de dados recuse um id que não seja de usuário:

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 pode estar vazio, pois as notas escritas antes desta parte não possuem usuário. Eles ficam na mesa e ninguém mais os vê.

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

Peça ao banco de dados as notas de um usuário

Cada função agora pega o id do usuário e o filtra. Uma nota de outra pessoa é respondida como se não existisse, então seu id não diz nada a um visitante curioso:

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

Uma função lê a sessão e envia qualquer pessoa sem usuário para /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 interrompe a página ou a ação ali mesmo: nada depois de executada.

Proteja as ações

Cada Action solicita primeiro o usuário, então um formulário postado sem sessão não faz nada:

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()

Proteja as páginas e a API

Uma página não pode enviar um visitante para um formulário, mas uma API pode responder por que ela recusa: a API responde 401, as páginas redirecionam.

src/rotas/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/rotas/notas/[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)]

Saia e abra http://localhost:3000/notes: você acessa a página de login. Faça login e a lista ficará vazia até você escrever uma nota. Crie uma segunda conta em uma janela privada: ela não verá as notas da primeira.

O que você aprendeu

Um foreign_key vincula uma nota a um usuário e cada consulta é filtrada nesse usuário.
A nota de outra pessoa é respondida 404, como uma nota que não existe.
redirect() interrompe uma página ou uma ação; uma API responde 401 com HTTPException.

Próximas etapas

Parte 12: redefinir uma senha esquecida por e-mail
Exceções e erros: 401, 403, 404 e como cada um é mostrado

Editar no GitHub