ДокументацияБлок 3: пользователи
Создайте приложение для заметок с помощью Nexy, часть 11: дайте каждому пользователю свои заметки
Часть 11 привязывает каждую заметку к написавшему ее пользователю: страницы заметок запрашивают вход в систему, каждый пользователь видит и изменяет только свои собственные заметки, а API отвечает 401 посетителю.
Прежде чем начать: приложение из части 10 с учетной записью.
Привяжите заметку к ее пользователю
Заметка получает идентификатор своего пользователя. foreign_key заставляет базу данных отклонять идентификатор, который не является пользователем:
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
Запросить у базы данных заметки одного пользователя
Каждая функция теперь принимает идентификатор пользователя и фильтрует его. На чужую записку отвечают так, как будто ее не существует, поэтому ее идентификатор ничего не говорит любопытному посетителю:
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 отвечает401с помощьюHTTPException.
Следующие шаги
- Часть 12: сброс забытого пароля по электронной почте
- Исключения и ошибки:
401,403,404и способ отображения каждого из них