我正在尝试从浏览器提交 HTML 表单，以在 SQLite 数据库表中创建新用户。单击“提交”按钮将触发使用 FastAPI 和 Sqlalchemy 2.0 的 POST 请求。从 Swagger UI 执行时，该 API 可以完美运行。但当从实际的 HTML 表单触发时，它不起作用，返回
422 Unprocessable Entity我可以看到错误指向一个
id我正在使用：
核心/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
用户/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)
用户/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
用户/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)
模板/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 %}
浏览器上的错误消息：
{
"detail": [
{
"type": "int_parsing",
"loc": [
"path",
"id"
],
"msg": "Input should be a valid integer, unable to parse string as an integer",
"input": "create"
}
]
}
我可以在这里看到几个潜在的问题：
/createusers/createtemplates/create_user.py<form method="POST" action="/users/create">
Form@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)
):
...