文档第三单元:用户
使用 Nexy 构建笔记应用程序,第 11 部分:为每个用户提供自己的笔记
第 11 部分将每条注释与编写该注释的用户联系起来:注释页面要求登录,每个用户只能看到和更改自己的注释,API 会向访问者做出响应 401。
开始之前:第 10 部分的应用程序,带有帐户。
将注释与其用户联系起来
注释获取其用户的 ID。 foreign_key 使数据库拒绝不是用户的 ID:
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 可能为空,因为这部分之前写的注释没有用户。他们留在桌子上,没有人再看到他们。
nx migrate -m "notes belong to users"nx migrate --upgrade
向数据库询问一位用户的注释
现在每个函数都获取用户的 ID,并对其进行过滤。别人的一条笔记会被当作不存在来回答,所以它的 id 不会告诉好奇的访问者任何信息:
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 True将访客引导至登录页面
一个函数读取会话,并将没有用户的任何人发送到 /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 立即停止页面或操作:运行后没有任何内容。
守护行动
每个操作首先请求用户,因此没有会话的表单发布不会执行任何操作:
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()保护页面和 API
页面无法将访问者发送到表单,但 API 可以回答其拒绝的原因:API 回答 401,页面重定向。
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)]注销,然后打开 http://localhost:3000/notes:您将进入登录页面。登录后,列表为空,直到您写下注释。在私人窗口中创建第二个帐户:它看不到第一个帐户的注释。
你学到了什么
foreign_key将注释与用户联系起来,每个查询都会过滤该用户。- 别人的笔记被回复
404,就像一条不存在的笔记一样。 redirect()停止页面或操作; API 用HTTPException回答401。
后续步骤
- 第 12 部分:通过电子邮件重置忘记的密码
- 异常和错误:
401、403、404以及每一个的显示方式