Model Forms & Formsets

from django.conf import settings
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
from django.urls import reverse


class Genre(models.Model):
    name = models.CharField(max_length=60, unique=True)

    class Meta:
        ordering = ["name"]

    def __str__(self):
        return self.name


class Author(models.Model):
    name = models.CharField(max_length=200)
    slug = models.SlugField(unique=True)

    class Meta:
        ordering = ["name"]

    def __str__(self):
        return self.name


class Book(models.Model):
    title = models.CharField(max_length=200)
    slug = models.SlugField(unique=True)
    isbn = models.CharField(max_length=13, unique=True)
    author = models.ForeignKey(Author, on_delete=models.PROTECT, related_name="books")
    genres = models.ManyToManyField(Genre, blank=True, related_name="books")
    price = models.DecimalField(max_digits=6, decimal_places=2)
    published = models.DateField()
    # Set by the server, never by the form:
    added_by = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
        related_name="+",
    )

    class Meta:
        ordering = ["title"]

    def __str__(self):
        return self.title

    def get_absolute_url(self):
        return reverse("catalog:book-detail", kwargs={"slug": self.slug})


class Review(models.Model):
    book = models.ForeignKey(Book, on_delete=models.CASCADE, related_name="reviews")
    reviewer = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name="reviews"
    )
    rating = models.PositiveSmallIntegerField(
        validators=[MinValueValidator(1), MaxValueValidator(5)]
    )
    body = models.TextField(blank=True)

    class Meta:
        ordering = ["id"]
        constraints = [
            models.UniqueConstraint(
                fields=["book", "reviewer"], name="one_review_per_reader_and_book"
            )
        ]

    def __str__(self):
        return f"{self.reviewer} on {self.book}: {self.rating}/5"

False {'isbn': ['An ISBN has 10 or 13 digits.'], 'slug': ['Another book already uses this slug.']}
ValueError - The Book could not be created because the data didn't validate.

PriceFormSet = modelformset_factory(
    Book,
    fields=["price"],
    extra=0,
    edit_only=True,
)

@login_required
def price_list(request):
    """Model formset limited to existing rows (edit_only=True)."""
    queryset = Book.objects.select_related("author").order_by("title")
    if request.method == "POST":
        formset = PriceFormSet(request.POST, queryset=queryset)
        if formset.is_valid():
            formset.save()
            return redirect("catalog:price-list")
    else:
        formset = PriceFormSet(queryset=queryset)
    return render(request, "catalog/price_list.html", {"formset": formset})

def test_edit_only_ignores_extra_forms(self):
    qs = Book.objects.order_by("title")
    rows = [{"id": b.pk, "price": "2.50"} for b in self.books]
    rows.append({"price": "9"})  # an extra, new-object row
    formset = PriceFormSet(formset_data("form", rows, initial=3), queryset=qs)
    self.assertTrue(formset.is_valid(), formset.errors)
    formset.save()
    self.assertEqual(Book.objects.count(), 3)  # nothing created
    self.assertEqual(set(Book.objects.values_list("price", flat=True)), {Decimal("2.50")})

def test_row_outside_queryset_is_ignored(self):
    qs = Book.objects.filter(pk=self.books[0].pk)
    rows = [{"id": self.books[1].pk, "price": "99"}]
    formset = PriceFormSet(formset_data("form", rows, initial=1), queryset=qs)
    self.assertTrue(formset.is_valid())  # not an error: the row is just not "initial"
    self.assertEqual(formset.save(), [])
    self.assertEqual(Book.objects.get(pk=self.books[1].pk).price, Decimal("1.00"))

class BookInlineFormSet(BaseInlineFormSet):
    """Books of one author; refuses two forms with the same title."""

    def clean(self):
        super().clean()  # runs the unique checks across the forms
        if any(self.errors):
            return
        titles = [
            f.cleaned_data["title"].casefold()
            for f in self.forms
            if f.cleaned_data and not f.cleaned_data.get("DELETE")
        ]
        if len(titles) != len(set(titles)):
            raise forms.ValidationError("Two books of one author cannot share a title.")


AuthorBooksFormSet = inlineformset_factory(
    Author,
    Book,
    form=BookForm,
    formset=BookInlineFormSet,
    fk_name="author",
    fields=["title", "slug", "isbn", "price", "published"],
    extra=2,
    max_num=5,
    validate_max=True,
    can_delete=True,
)

@login_required
def author_books(request, slug):
    """Inline formset: an author and their books on one page."""
    author = get_object_or_404(Author, slug=slug)
    if request.method == "POST":
        formset = AuthorBooksFormSet(request.POST, instance=author)
        if formset.is_valid():
            with transaction.atomic():
                formset.save()
            messages.success(request, "Books saved.")
            return redirect("catalog:author-books", slug=author.slug)
    else:
        formset = AuthorBooksFormSet(instance=author)
    return render(
        request, "catalog/author_books.html", {"author": author, "formset": formset}
    )

def test_add_edit_and_delete_in_one_post(self):
    data = formset_data(
        "books",
        [
            {"id": self.dune.pk, "author": self.author.pk, "title": "Dune (1st ed.)",
             "slug": "dune", "isbn": "9780441172719", "price": "10.00", "published": "1965-08-01"},
            {"title": "Messiah", "slug": "messiah", "isbn": "9780441172696",
             "price": "8.00", "published": "1969-01-01"},
        ],
        initial=1, maximum=5,
    )
    formset = AuthorBooksFormSet(data, instance=self.author)
    self.assertTrue(formset.is_valid(), formset.errors)
    saved = formset.save()
    self.assertEqual(len(saved), 2)
    self.assertEqual(formset.changed_objects[0][1], ["title", "price"])
    self.assertEqual([b.title for b in formset.new_objects], ["Messiah"])
    self.assertEqual(Book.objects.get(slug="messiah").author, self.author)  # FK set by the formset

    data = formset_data(
        "books",
        [{"id": self.dune.pk, "author": self.author.pk, "title": "Dune", "slug": "dune",
          "isbn": "9780441172719", "price": "10.00", "published": "1965-08-01", "DELETE": "on"}],
        initial=1, maximum=5,
    )
    formset = AuthorBooksFormSet(data, instance=self.author)
    self.assertTrue(formset.is_valid(), formset.errors)
    formset.save()
    self.assertEqual([b.slug for b in formset.deleted_objects], ["dune"])
    self.assertFalse(Book.objects.filter(slug="dune").exists())

def test_default_prefix_and_empty_form(self):
    formset = AuthorBooksFormSet(instance=self.author)
    self.assertEqual(formset.prefix, "books")  # the related_name
    self.assertEqual(formset.initial_form_count(), 1)
    self.assertEqual(len(formset.forms), 3)  # 1 existing + extra=2
    self.assertEqual(formset.empty_form.prefix, "books-__prefix__")
    self.assertIn('name="books-TOTAL_FORMS"', str(formset.management_form))

const total = document.querySelector('#id_books-TOTAL_FORMS');
document.querySelector('#add-row').addEventListener('click', () => {
  const index = Number(total.value);
  const html = document.querySelector('#empty-form').innerHTML.replaceAll('__prefix__', index);
  document.querySelector('#rows').insertAdjacentHTML('beforeend', html);
  total.value = index + 1;   // never exceed MAX_NUM_FORMS in the UI; the server enforces it anyway
});

class UserReviewFormSet(BaseInlineFormSet):
    """Alternative to form_kwargs=...: the formset owns the user and hands it down."""

    def __init__(self, *args, user, **kwargs):
        self.user = user
        super().__init__(*args, **kwargs)

    def get_form_kwargs(self, index):
        # index is None for empty_form; the dict is per form.
        kwargs = super().get_form_kwargs(index)
        kwargs["user"] = self.user
        return kwargs


ReviewFormSet = inlineformset_factory(
    Book,
    Review,
    form=ReviewForm,
    formset=UserReviewFormSet,
    fields=["reviewer", "rating", "body"],
    extra=1,
    can_delete=True,
)

@login_required
def book_reviews(request, slug):
    """Inline formset whose forms receive request.user, saved with commit=False."""
    book = get_object_or_404(Book, slug=slug)
    mine = book.reviews.filter(reviewer=request.user)
    if request.method == "POST":
        formset = ReviewFormSet(
            request.POST, instance=book, queryset=mine, user=request.user
        )
        if formset.is_valid():
            reviews = formset.save(commit=False)
            for review in reviews:
                review.reviewer = request.user  # belt and braces; the form already did
                review.save()
            for review in formset.deleted_objects:
                review.delete()  # commit=False does not delete by itself
            return redirect(book)
    else:
        formset = ReviewFormSet(instance=book, queryset=mine, user=request.user)
    return render(request, "catalog/book_reviews.html", {"book": book, "formset": formset})

def test_get_form_kwargs_reaches_every_form_and_the_empty_form(self):
    formset = ReviewFormSet(instance=self.book, user=self.ann)
    self.assertTrue(all(f.user == self.ann for f in formset.forms))
    self.assertEqual(formset.empty_form.user, self.ann)

def test_form_kwargs_shorthand_on_a_plain_formset(self):
    from django.forms import inlineformset_factory

    Plain = inlineformset_factory(Book, Review, form=ReviewForm, fields=["rating", "body"])
    formset = Plain(instance=self.book, form_kwargs={"user": self.bob})
    self.assertEqual(formset.forms[0].user, self.bob)

def test_commit_false_view_flow(self):
    self.client.force_login(self.ann)
    url = reverse("catalog:book-reviews", args=["b"])
    data = formset_data("reviews", [{"rating": "5", "body": "Loved it"}], initial=0, maximum=1000)
    response = self.client.post(url, data)
    self.assertRedirects(response, self.book.get_absolute_url())
    review = Review.objects.get()
    self.assertEqual((review.reviewer, review.book, review.rating), (self.ann, self.book, 5))

    # Bob cannot see or edit ann's review: the queryset is scoped to the user.
    self.client.force_login(self.bob)
    data = formset_data("reviews", [{"id": review.pk, "book": self.book.pk, "rating": "1"}], initial=1)
    response = self.client.post(url, data)
    self.assertEqual(response.status_code, 302)  # valid, but the row is ignored
    review.refresh_from_db()
    self.assertEqual((review.rating, review.reviewer), (5, self.ann))

    # Ann deletes hers; with commit=False the view deletes deleted_objects itself.
    self.client.force_login(self.ann)
    data = formset_data("reviews", [{"id": review.pk, "book": self.book.pk, "rating": "5", "DELETE": "on"}], initial=1)
    self.client.post(url, data)
    self.assertFalse(Review.objects.exists())

async def book_create_async(request):
    """Async view: form code is synchronous, so the DB-touching calls are wrapped."""
    user = await request.auser()
    if not user.is_authenticated:
        return HttpResponse(status=401)
    form = BookForm(request.POST or None)
    if request.method == "POST":
        # is_valid() runs ModelChoiceField and unique queries: synchronous ORM.
        if await sync_to_async(form.is_valid)():
            book = form.save(commit=False)  # touches no database
            book.added_by = user
            await book.asave()
            await book.genres.aset(form.cleaned_data["genres"])  # instead of save_m2m()
            return HttpResponse(f"created {book.slug}", status=201)
        return HttpResponse(form.errors.as_json(), status=400, content_type="application/json")
    return HttpResponse("POST only", status=405)

SynchronousOnlyOperation - You cannot call this from an async context - use a thread or sync_to_async.
<select name="author" required id="id_author">   <option val
SynchronousOnlyOperation - You cannot call this from an async context - use a thread or sync_to_async.

async def test_async_view(self):
    author = await Author.objects.acreate(name="A", slug="a")
    genre = await Genre.objects.acreate(name="SF")
    user = await User.objects.acreate_user("ann", password="s3cret-pass-1")
    await self.async_client.aforce_login(user)
    response = await self.async_client.post(
        reverse("catalog:book-create-async"),
        book_data(author=author.pk, genres=[genre.pk]),
    )
    self.assertEqual(response.status_code, 201)
    book = await Book.objects.select_related("added_by").aget(slug="dune")
    self.assertEqual(book.added_by, user)
    self.assertEqual([g async for g in book.genres.all()], [genre])
    bad = await self.async_client.post(reverse("catalog:book-create-async"), book_data(author=author.pk, isbn="x"))
    self.assertEqual(bad.status_code, 400)

System check identified no issues (0 silenced).
Creating test database for alias 'default'...
................................
----------------------------------------------------------------------
Ran 32 tests in 0.980s

OK
Destroying test database for alias 'default'...
Found 32 test(s).
System check identified no issues (0 silenced).

def book_data(**extra):
    data = {
        "title": "Dune",
        "slug": "dune",
        "isbn": "9780441172719",
        "price": "9.99",
        "published": "1965-08-01",
    }
    data.update(extra)
    return data

def formset_data(prefix, forms_data, initial=0, minimum=0, maximum=1000):
    data = {
        f"{prefix}-TOTAL_FORMS": str(len(forms_data)),
        f"{prefix}-INITIAL_FORMS": str(initial),
        f"{prefix}-MIN_NUM_FORMS": str(minimum),
        f"{prefix}-MAX_NUM_FORMS": str(maximum),
    }
    for i, row in enumerate(forms_data):
        for key, value in row.items():
            data[f"{prefix}-{i}-{key}"] = value
    return data

$ curl -s -b cookies.txt -c cookies.txt http://127.0.0.1:8136/catalog/authors/herbert/books/ | grep -o '<input type="hidden"[^>]*>'
<input type="hidden" name="csrfmiddlewaretoken" value="...">
<input type="hidden" name="books-TOTAL_FORMS" value="3" id="id_books-TOTAL_FORMS">
<input type="hidden" name="books-INITIAL_FORMS" value="1" id="id_books-INITIAL_FORMS">
<input type="hidden" name="books-MIN_NUM_FORMS" value="0" id="id_books-MIN_NUM_FORMS">
<input type="hidden" name="books-MAX_NUM_FORMS" value="5" id="id_books-MAX_NUM_FORMS">
<input type="hidden" name="books-0-id" value="1" id="id_books-0-id">
<input type="hidden" name="books-0-author" value="1" id="id_books-0-author">
<input type="hidden" name="books-1-id" id="id_books-1-id">
...
$ curl ... -X POST .../authors/herbert/books/ -d books-TOTAL_FORMS=2 -d books-INITIAL_FORMS=1 ... -d books-0-price=10.50 -d books-1-title=Messiah ...
302 -> http://127.0.0.1:8136/catalog/authors/herbert/books/
$ curl ... -X POST ... (two forms titled "Same" and "same")
<ul class="errorlist nonform"><li>Two books of one author cannot share a title.</li></ul>
$ curl -X POST ... (no csrfmiddlewaretoken)
403
[('dune', Decimal('10.50')), ('messiah', Decimal('8.00'))]   <- database state after the first POST

from django.urls import path

from . import views

app_name = "catalog"

urlpatterns = [
    path("books/new/", views.BookCreateView.as_view(), name="book-create"),
    path("books/<slug:slug>/", views.BookDetailView.as_view(), name="book-detail"),
    path("books/<slug:slug>/edit/", views.book_update, name="book-update"),
    path("books/<slug:slug>/reviews/", views.book_reviews, name="book-reviews"),
    path("reviews/new/", views.ReviewCreateView.as_view(), name="review-create"),
    path("authors/<slug:slug>/books/", views.author_books, name="author-books"),
    path("prices/", views.price_list, name="price-list"),
    path("async/books/new/", views.book_create_async, name="book-create-async"),
    path("", views.price_list, name="index"),
]