DocsUnité 3 : utilisateurs
Créez une application de notes avec Nexy, partie 11 : donnez à chaque utilisateur ses propres notes
La partie 11 lie chaque note à l'utilisateur qui l'a écrite : les pages de notes demandent une connexion, chaque utilisateur voit et modifie uniquement ses propres notes et l'API répond 401 à un visiteur.
Avant de commencer : l'application de la partie 10, avec un compte.
Attacher une note à son utilisateur
Une note obtient l'identifiant de son utilisateur. foreign_key fait que la base de données refuse un identifiant qui n'est pas un utilisateur :
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)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 peut être vide, car les notes écrites avant cette partie n'ont aucun utilisateur. Ils restent dans le tableau et personne ne les voit plus.
nx migrate -m "notes belong to users"nx migrate --upgrade
Demander à la base de données les notes d'un utilisateur
Chaque fonction prend désormais l'identifiant de l'utilisateur et le filtre. Une note de quelqu'un d'autre reçoit une réponse comme si elle n'existait pas, donc son identifiant ne dit rien au visiteur curieux :
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 Truefrom 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 TrueEnvoyer un visiteur vers la page de connexion
Une fonction lit la session et envoie toute personne sans utilisateur vers /signin :
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_idredirect arrête la page ou l'action juste là : rien après son exécution.
Protégez les actions
Chaque action demande d'abord l'utilisateur, donc un formulaire publié sans session ne fait rien :
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()Protégez les pages et l'API
Une page ne peut pas renvoyer un visiteur vers un formulaire, mais une API peut répondre pourquoi elle refuse : l'API répond 401, les pages redirigent.
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>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>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)]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>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>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)]Déconnectez-vous, puis ouvrez http://localhost:3000/notes : vous arrivez sur la page de connexion. Connectez-vous et la liste est vide jusqu'à ce que vous écriviez une note. Créez un deuxième compte dans une fenêtre privée : il ne voit pas les notes du premier.
Ce que tu as appris
- Un
foreign_keylie une note à un utilisateur et chaque requête filtre sur cet utilisateur. - La note de quelqu'un d'autre reçoit une réponse
404, comme une note qui n'existe pas. redirect()arrête une page ou une action ; une API répond401parHTTPException.
Prochaines étapes
- Partie 12 : réinitialiser un mot de passe oublié par email
- Exceptions et erreurs :
401,403,404et comment chacune est affichée