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:

src/model/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 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:

src/catatan.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/catatan.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

Kirim pengunjung ke halaman masuk

Satu fungsi membaca sesi, dan mengirimkan siapa pun tanpa pengguna ke /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 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:

src/aksi/catatan.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()

Jaga halaman dan API

Halaman tidak dapat mengirim pengunjung ke formulir, namun API dapat menjawab mengapa ia menolak: API menjawab 401, halaman dialihkan.

src/rute/catatan/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/rute/catatan/[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)]

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_key mengikat 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 menjawab 401 dengan HTTPException.

Langkah selanjutnya

Sunting di GitHub

Enter membuka hasil, Esc menutup.