官方教程原文

https://fastapi.tiangolo.com/zh/tutorial/sql-databases/

SQLAlchemy的更多操作方法详见 https://·/sqlalchemy/orm/tutorial.html

目录结构

官方教程中，main.py放在sql_app文件夹下，考虑到一般习惯，本文将main.py文件放在与sql_app文件夹相同的位置。结构如下：

. ├── main.py └── sql_app ├── __init__.py ├── crud.py ├── database.py ├── models.py └── schemas.py

这里__init__.py文件留空即可，它的存在会让解释器将sql_app文件夹视为一个“包”。

数据库连接（database.py）

这里引入SQLAlchemy的主要组件，并初始化数据库连接。

#database.py from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker #数据库访问地址 SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" # SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db" #启动引擎 engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} ) #启动会话 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) #数据模型的基类 Base = declarative_base() 数据模型（models.py）

模型（models）用于建立供SQLAlchemy模块增删改查使用的对象，继承SQLAlchemy中的数据模型基类，根据项目需求建立数据模型。

#models.py from sqlalchemy import Boolean,Column,ForeignKey,Integer,String from sqlalchemy.orm import relationship from .database import Base class User(Base): __tablename__="users" id=Column(Integer,primary_key=True,index=True) email=Column(String,unique=True,index=True) hashed_password=Column(String) is_active=Column(Boolean,default=True) items=relationship('Item',back_populates='owner') class Item(Base): __tablename__ = "items" id = Column(Integer, primary_key=True, index=True) title = Column(String, index=True) description = Column(String, index=True) owner_id = Column(Integer, ForeignKey("users.id")) # 表a_user.items等于相应的Item表 owner = relationship("User", back_populates="items")

如果需要将模型存储到物理结构，在代码中添加：

Base.metadata.create_all(bind=engine)

这里的Base和engine，就是在上一段代码里面新建的对象，create_all()操作会生成一个*.db结尾的数据库文件并写入表结构，如果数据库文件和表结构已经存在，则不会破坏现有的结构，可以放心使用。

数据模式（schemas.py）

数据模式（schemas）是FastAPI模块用于数据传递的对象，通过继承pydantic中的类建立。

#schemas.py from typing import List, Optional from pydantic import BaseModel class ItemBase(BaseModel): title: str description: Optional[str] = None class ItemCreate(ItemBase): pass class Item(ItemBase): id: int owner_id: int class Config: orm_mode = True class UserBase(BaseModel): email: str class UserCreate(UserBase): password: str class User(UserBase): id: int is_active: bool items: List[Item] = [] class Config: orm_mode = True 数据传递（curd.py）

现在，FastAPI和SQLAlchemy两方面均建立了各自的数据模型或模式，接下来需要做的就是定义数据的传递和数据库的增删改查方法。

#curd.py from sqlalchemy.orm import Session from . import models, schemas def get_user(db: Session, user_id: int): return db.query(models.User).filter(models.User.id == user_id).first() def get_user_by_email(db: Session, email: str): return db.query(models.User).filter(models.User.email == email).first() def get_users(db: Session, skip: int = 0, limit: int = 100): return db.query(models.User).offset(skip).limit(limit).all() def create_user(db: Session, user: schemas.UserCreate): fake_hashed_password = user.password + "notreallyhashed" db_user = models.User(email=user.email, hashed_password=fake_hashed_password) db.add(db_user) db.commit() db.refresh(db_user) return db_user def get_items(db: Session, skip: int = 0, limit: int = 100): return db.query(models.Item).offset(skip).limit(limit).all() def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int): db_item = models.Item(**item.dict(), owner_id=user_id) db.add(db_item) db.commit() db.refresh(db_item) return db_item 数据查询方法

返回对象为一个字典

这里涉及的类和对象有：

db：Session类的对象；User：继承自Base = declarative_base()的类；User.email：User类的成员变量。 db.query(User).filter(User.email == '123@abc.com').first()

相当于

SELECT * FROM users WHRER email='123@abc.com' LIMIT 1 数据插入方法 #根据表模型实例化一个数据对象 db_user = models.User(email='123@abc.com', hashed_password="fake_hashed_password") #向数据库添加数据 db.add(db_user) #提交更改，只有执行commit()操作才会真正把数据添加到数据库中 db.commit() #刷新数据库连接 db.refresh(db_user) 数据提取与转化

数据库查询的结果是对象，用点号访问即可，如db_user.email。

提取schemas中的数据，可以用点号，如user.email，也可以转化成字典，如user.dict()。

相似的模型和模式之间的数据交换可以直接用解包后的字典（如**item.dict()）作为构造时的参数。

db_item = models.Item(**item.dict(), owner_id=user_id) 在API 中应用（main.py） #main.py from typing import List#用于定义对象数组 from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from sql_app import crud, models, schemas from sql_app.database import SessionLocal, engine #在数据库中生成表结构 models.Base.metadata.create_all(bind=engine) app = FastAPI() # Dependency def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.post("/users/", response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): db_user = crud.get_user_by_email(db, email=user.email) if db_user: raise HTTPException(status_code=400, detail="Email already registered") return crud.create_user(db=db, user=user) @app.get("/users/", response_model=List[schemas.User]) def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): users = crud.get_users(db, skip=skip, limit=limit) return users @app.get("/users/{user_id}", response_model=schemas.User) def read_user(user_id: int, db: Session = Depends(get_db)): db_user = crud.get_user(db, user_id=user_id) if db_user is None: raise HTTPException(status_code=404, detail="User not found") return db_user @app.post("/users/{user_id}/items/", response_model=schemas.Item) def create_item_for_user( user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) ): return crud.create_user_item(db=db, item=item, user_id=user_id) @app.get("/items/", response_model=List[schemas.Item]) def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): items = crud.get_items(db, skip=skip, limit=limit) return items 代码参数解析

以下面这段代码为例：

@app.post("/users/", response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): db_user = crud.get_user_by_email(db, email=user.email) if db_user: raise HTTPException(status_code=400, detail="Email already registered") return crud.create_user(db=db, user=user) response_model=schemas.User ，响应模板，限制API返回数据的格式；user: schemas.UserCreate，请求模板，限制前端请求的格式；db: Session = Depends(get_db)，注入依赖，执行此函数前先执行函数“get_db”。 启动应用

在命令行输入

$ unicorn main:app --reload

默认端口为8000，运行后可以通过浏览器访问

API访问端口： http://127.0.0.1:8000

API文档和交互调试： http://127.0.0.1:8000/docs

API可选文档：http://127.0.0.1:8000/redoc

API原始数据：http://127.0.0.1:8000/openapi.json

---

1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，会注明原创字样,如未注明都非原创，如有侵权请联系删除！；3.作者投稿可能会经我们编辑修改或补充；4.本站不提供任何储存功能只提供收集或者投稿人的网盘链接。