pythonsqlitesqlalchemyjinja2fastapi

How to trigger a POST request API to add a record in a SQLite database table using FastAPI and HTML forms using Jinja2?


I am trying to submit a HTML form from the browser to create a new user in a SQLite database table. Clicking on the Submit button triggers a POST request using FastAPI and Sqlalchemy 2.0. The API works perfectly when executed from the Swagger UI. But it does not work when triggered from an actual HTML form, returning a 422 Unprocessable Entity error. Below is the code I have used for the same along with the error I am seeing on the browser.

I can see that the error is pointing to an id which does not exist in my Pydantic model and also in my html form. Any help on how to handle this error would be greatly appreciated.

I am using:

core/database.py

from os import getenv
from dotenv import load_dotenv
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

load_dotenv()  # Needed to load full path of the .env file

engine = create_engine(
    getenv("DATABASE_URL"),
    connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Database dependency
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# declarative base class
class Base(DeclarativeBase):
    pass

users/models.py

from datetime import datetime, timezone
from sqlalchemy import String, Enum
from sqlalchemy.orm import Mapped, mapped_column
from typing import Optional, List
from enum import Enum as pyEnum
from .database import Base


class Gender(str, pyEnum):
    default = ""
    male = "M"
    female = "F"


def time_now():
    return datetime.now(timezone.utc).strftime("%b %d, %Y %I:%M:%S %p")


class AbstractBase(Base):
    __abstract__ = True

    id: Mapped[Optional[int]] = mapped_column(primary_key=True, index=True)
    created_by: Mapped[Optional[str]] = mapped_column(String(50), nullable=True, default="")
    updated_by: Mapped[Optional[str]] = mapped_column(String(50), nullable=True, default="")
    created: Mapped[Optional[str]] = mapped_column(nullable=True, default=time_now)
    updated: Mapped[Optional[str]] = mapped_column(nullable=True, default=time_now, onupdate=time_now)


class User(AbstractBase):
    __tablename__ = "users"

    first_name: Mapped[str] = mapped_column(String(25), nullable=False)
    last_name: Mapped[Optional[str]] = mapped_column(String(25), default="")
    gender: Mapped[Optional[Gender]] = mapped_column(Enum(Gender), nullable=False, default=Gender.default.value)
    email: Mapped[str] = mapped_column(String(50), unique=True, nullable=False, index=True)

users/schemas.py

from typing import Optional
from pydantic import BaseModel, EmailStr, Field
from .models import Gender


class UserCreate(BaseModel):
    first_name: str = Field(min_length=1, max_length=25)
    last_name: Optional[str] = Field(min_length=0, max_length=25, default="")
    gender: Optional[Gender] = Gender.default.value
    email: EmailStr = Field(min_length=7, max_length=50)


class UserUpdate(UserCreate):
    pass


class UserResponse(UserUpdate):
    id: Optional[int]
    created_by: Optional[EmailStr] = Field(min_length=7, max_length=50)
    updated_by: Optional[EmailStr] = Field(min_length=7, max_length=50)
    created: Optional[str]
    updated: Optional[str]

    class Config:
        from_attributes = True

users/routers.py

from fastapi import APIRouter, Depends, HTTPException, Request, Form
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from sqlalchemy.orm import Session
from core.database import get_db
from users.models import User
from users.schemas import UserCreate, UserUpdate, UserResponse


templates = Jinja2Templates(directory="templates")

users_router = APIRouter()


# API to redirect user to the Register page
@users_router.get("/register", response_class=HTMLResponse, status_code=200)
async def redirect_user(request: Request):
    return templates.TemplateResponse(
        name="create_user.html",
        context={
            "request": request,
            "title": "FastAPI - Create User",
            "navbar": "create_user"
        }
    )


# API to create new user
@users_router.post("/create", response_model=UserCreate, response_class=HTMLResponse, status_code=201)
async def create_user(request: Request, user: UserCreate=Form(), db: Session = Depends(get_db)):
    if db.query(User).filter(User.email == user.email).first():
        raise HTTPException(
            status_code=403,
            detail=f"Email '{user.email}' already exists. Please try with another email."
        )
    obj = User(
        first_name = user.first_name,
        last_name = user.last_name,
        gender = user.gender.value,
        email = user.email.lower(),
        created_by = user.email.lower(),
        updated_by = user.email.lower()
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return (
        templates.TemplateResponse(
            name="create_user.html",
            context={
                "request": request,
                "item": obj,
                "title": "FastAPI - Create User",
                "navbar": "create_user"
            }
        )
    )

main.py

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.templating import Jinja2Templates
from core.database import Base, engine
from users.routers import users_router    

templates = Jinja2Templates(directory="templates")


# Create FastAPI instance
app = FastAPI()
app.include_router(users_router, prefix='/users', tags = ['Users'])

# Specify URLS that are allowed to connect to the APIs
origins = [
    "http://localhost",
    "http://127.0.0.1",
    "http://localhost:8000",
    "http://127.0.0.1:8000"
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Create tables in database
Base.metadata.create_all(bind=engine)

templates/create_user.py

{% extends 'base.html' %}

{% block title %} {{ title }} {% endblock title %}

{% block content %}
    <form method="POST" action="/create">
        <div class="mb-3">
            <label for="first_name" class="form-label">First Name</label>
            <input type="text" class="form-control" id="first_name" name="first_name">
        </div>
        <div class="mb-3">
            <label for="last_name" class="form-label">Last Name</label>
            <input type="text" class="form-control" id="last_name" name="last_name">
        </div>
        <div class="mb-3">
            <label for="email" class="form-label">Email</label>
            <input type="email" class="form-control" id="email" name="email" aria-describedby="emailHelp">
        </div>
        <a href="{{ url_for('create_user') }}" type="submit" class="btn btn-outline-success float-end">Submit</a>
    </form>
{% endblock content %}

Error Message on Browser:

{
  "detail": [
    {
      "type": "int_parsing",
      "loc": [
        "path",
        "id"
      ],
      "msg": "Input should be a valid integer, unable to parse string as an integer",
      "input": "create"
    }
  ]
}

Solution

  • I could see here couple of potential problems:

    1. User Form dependency for each form field:
    @users_router.post("/create", response_class=HTMLResponse, status_code=201)
    async def create_user(
        request: Request,
        first_name: str = Form(),
        last_name: str = Form(),
        gender: str = Form(),
        email: str = Form(),
        db: Session = Depends(get_db)
    ):
        ...
    

    or parse FormData into Pydantic Model:

    class UserCreate(BaseModel):
        first_name:str = Field(min_length=1, max_length=25)
        last_name: str | None= Field(default="", min_length=0, max_length=25)
        gender: Gender | None = Field(default=Gender.default.value)
        email: EmailStr =Field(min_length=7, max_length=50)
    
        @classmethod
        def as_form(
            cls,
            first_name: str = Form(...),
            last_name: str | None =Form(''),
            gender: Gender | None = Form(Gender.default.value),
            email:EmailStr =Form(...)
        ):
            ...
    
    1. In your form you using <a> tag with href attribute and type="submit" attribute, but <a> tag doesntt support type attribute and clicking it will cause GET request to the URL specified in href attribute, not POST request with form data. This means your form data isnt being submitted. So you could change change attribute of your tag:
    <form method="POST" action="/users/create">
    

    or

    <form method="POST" action="{{request.url_for('create_user')}}">
    

    but make sure to add name='create_user' to your route so that url_for can find it

    @users_router.post("/create", ..., name='create_user')
    async def create_user(...):
        ...
    

    after this, replace <a> tag with element of type submit

    <button type="submit" class="btn btn-outline-success float-end">Submit</button>