DokumentasiUnit 3: pengguna
Membangun aplikasi catatan dengan Nexy, bagian 11: memberikan catatan mereka sendiri kepada setiap pengguna
Bagian 11 menghubungkan setiap catatan dengan pengguna yang menulisnya: halaman catatan meminta untuk masuk, setiap pengguna hanya melihat dan mengubah catatan mereka sendiri, dan API menjawab 401 kepada pengunjung.
Sebelum Anda memulai: aplikasi bagian 10, dengan akun.
Ikat catatan ke penggunanya
Sebuah catatan mendapatkan id penggunanya. foreign_key membuat database menolak id yang bukan pengguna:
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 mungkin kosong, karena catatan yang ditulis sebelum bagian ini tidak ada penggunanya. Mereka tetap berada di meja, dan tidak ada yang melihat mereka lagi.
nx migrate -m "notes belong to users"nx migrate --upgrade
Minta database untuk catatan satu pengguna
Setiap fungsi sekarang mengambil id pengguna, dan memfilternya. Catatan orang lain dijawab seolah-olah tidak ada, jadi idnya tidak memberitahukan apa pun kepada pengunjung yang penasaran:
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 TrueKirim pengunjung ke halaman masuk
Satu fungsi membaca sesi, dan mengirimkan siapa pun tanpa pengguna ke /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 menghentikan halaman atau Tindakan di sana: tidak ada apa pun setelah dijalankan.
Jaga Tindakannya
Setiap Tindakan meminta pengguna terlebih dahulu, sehingga formulir yang diposting tanpa sesi tidak menghasilkan apa pun:
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()Jaga halaman dan API
Halaman tidak dapat mengirim pengunjung ke formulir, namun API dapat menjawab mengapa ia menolak: API menjawab 401, halaman dialihkan.
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)]Keluar, lalu buka http://localhost:3000/notes: Anda masuk ke halaman masuk. Masuk, dan daftarnya kosong sampai Anda menulis catatan. Buat akun kedua di jendela pribadi: ia tidak melihat catatan yang pertama.
Apa yang Anda pelajari
foreign_keymengikat catatan ke pengguna, dan setiap kueri memfilter pengguna tersebut.- Catatan orang lain dijawab
404, seperti catatan yang tidak ada. redirect()menghentikan halaman atau Tindakan; sebuah API menjawab401denganHTTPException.
Langkah selanjutnya
- Bagian 12: mereset kata sandi yang terlupa melalui email
- Pengecualian dan kesalahan:
401,403,404dan cara masing-masing ditampilkan