字数 1552，阅读大约需 8 分钟

Python之FastAPI的入门到精通系列：从零开始创建第一个API

代码地址：[email protected]:FunkyGod/fastapi-demo.git
目标功能：教学和分享Python开发，将0开始，将全部实践代码提交到代码仓；
范围包括：基础语法、web构建、机器学习、深度学习、AI实践。

本篇核心

分享如何从零开始，创建第一个API，以创建和更新系统账户维护为例，从model开始，维护schema，crud, 并开发业务逻辑的API。

1. 客户端发送请求（如 POST /users/ 带 JSON 数据）。
2. api 层接收请求，通过 UserCreate（schema）验证 JSON 数据格式。
3. api 层调用 create_user（crud 层），传入数据库会话和验证后的 UserCreate 对象。
4. crud 层将 UserCreate 转换为 User（model），执行数据库插入操作。
5. crud 层返回数据库生成的 User 对象（含 id 等）。
6. api 层通过 UserResponse（schema）将 User 对象序列化为 JSON，返回给客户端。

数据库Model层(User)CRUD层Schema验证(UserCreate)API层客户端数据库Model层(User)CRUD层Schema验证(UserCreate)API层客户端1. POST /users/(JSON数据)2. 验证数据格式验证通过(UserCreate对象)3. create_user()(session, UserCreate)4. 转换为User模型执行INSERT操作返回插入结果(含id等)5. 返回User对象User对象(含id)6. 序列化(UserResponse)JSON响应返回JSON响应

1. GET / - 获取用户列表（需超级用户权限）
2. POST / - 创建新用户（需超级用户权限）
3. PUT /me - 更新当前用户信息
4. GET /me - 获取当前用户信息
5. POST /open - 开放注册（无需登录）
6. GET /{id} - 根据ID获取用户信息
7. PUT /{id} - 更新用户（需超级用户权限）

model：数据库表定义

最核心的用户表，可以理解为后续的数据库表结构: 通过alembic 将model生成对应的DDL等sqlalchemy支持的SQL迁移python代码，实现无痕迁移。

基于 ORM 框架（如 SQLAlchemy）定义，包含表名、字段类型、约束（主键、外键、索引等）、关系（一对一、一对多等）;

作为数据库操作的 “蓝图”，ORM 通过 model 将 Python 类与数据库表关联，实现数据的持久化存储。


    
    
    
  from sqlalchemy import Boolean, Column, Integer, String

from app.db.base_class import Base


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean(), default=True)
    is_superuser = Column(Boolean(), default=False)

alembic 迁移表

使用alembic进行表迁移：Welcome to Alembic’s documentation! — Alembic 1.17.1 documentation^[1]

alembic 是轻量化的表迁移工具，usage with the SQLAlchemy Database 管理数据库表结构和数据信息。

执行alembic revision --autogenerate -m 生成迁移文件！


    
    
    
  root@fastapi:/service# alembic revision --autogenerate -m "v 1.0.0用户信息表"
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'user'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_user_email' on '('email',)'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_user_full_name' on '('full_name',)'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_user_id' on '('id',)'
  Generating /service/alembic/versions/2025-11-06_v_1_0_0用户信息表_905e5bb0f65c.py ...  done


root@fastapi:/service# alembic upgrade head
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade ec699efbc75a -> 905e5bb0f65c, v 1.0.0用户信息表

查询数据库，对应user表成功生成

schema

定义 API 接口的输入 / 输出数据格式，负责数据的验证、序列化（对象转 JSON）和反序列化（JSON 转对象）。

在FastAPI，推荐使用pydantic v2管理校验，v1版本太旧了。


    
    
    
  from typing import Optional

from pydantic import BaseModel, EmailStr


# Shared properties
class UserBase(BaseModel):
    email: Optional[EmailStr] = None
    is_active: Optional[bool] = True
    is_superuser: bool = False
    full_name: Optional[str] = None


# Properties to receive via API on creation
class UserCreate(UserBase):
    email: EmailStr
    password: str


# Properties to receive via API on update
class UserUpdate(UserBase):
    password: Optional[str] = None


class UserInDBBase(UserBase):
    id: Optional[int] = None

    class Config:
        orm_mode = True


# Additional properties to return via API
class User(UserInDBBase):
    pass


# Additional properties stored in DB
class UserInDB(UserInDBBase):
    hashed_password: str

crud

核心职责：封装数据库的具体操作逻辑（Create、Read、Update、Delete），是业务逻辑与数据库交互的中间层。


    
    
    
  from typing import Any, Dict, Optional, Union

from sqlalchemy.orm import Session

from app.core.security import get_password_hash, verify_password
from app.crud.base import CRUDBase
from app.model.user import User
from app.schema.user import UserCreate, UserUpdate


class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
    def get_by_email(self, db: Session, *, email: str) -> Optional[User]:
        return db.query(User).filter(User.email == email).first()

    def create(self, db: Session, *, obj_in: UserCreate) -> User:
        db_obj = User(
            email=obj_in.email,
            hashed_password=get_password_hash(obj_in.password),
            full_name=obj_in.full_name,
            is_superuser=obj_in.is_superuser,
        )
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def update(
        self, db: Session, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
    ) -> User:
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.dict(exclude_unset=True)
        if update_data.get("password", ""):
            hashed_password = get_password_hash(update_data["password"])
            del update_data["password"]
            update_data["hashed_password"] = hashed_password
        return super().update(db, db_obj=db_obj, obj_in=update_data)

    def authenticate(self, db: Session, *, email: str, password: str) -> Optional[User]:
        user = self.get_by_email(db, email=email)
        if not user:
            return None
        if not verify_password(password, user.hashed_password):
            return None
        return user

    def is_active(self, user: User) -> bool:
        return user.is_active

    def is_superuser(self, user: User) -> bool:
        return user.is_superuser


user = CRUDUser(User)

api

以创建用户为例，演示如何开发一个API接口，核心职责：定义 API 路由（URL 路径）和请求方法（GET/POST/PUT/DELETE 等），接收客户端请求，协调 schema 验证和 crud 操作，返回响应。


    
    
    
  @router.post("/", response_model=schema.User)
def create_user(
    *,
    db: Session = Depends(deps.get_db),
    user_in: schema.UserCreate,
    current_user: model.User = Depends(deps.get_current_active_superuser),
) -> Any:
    """
    Create new user.
    """
    user = crud.user.get_by_email(db, email=user_in.email)
    if user:
        raise HTTPException(
            status_code=400,
            detail="The user with this username already exists in the system.",
        )
    user = crud.user.create(db, obj_in=user_in)
    if settings.EMAILS_ENABLED and user_in.email:
        send_new_account_email(
            email_to=user_in.email, username=user_in.email, password=user_in.password
        )
    return user