FastAPI可与任何数据库和任何样式的库一起使用,以与数据库进行通信。

一种常见的模式是使用 ORM :“对象关系映射”库。

数据库连接使用有多种映射模型( ORM.) 如 Peewee, Django-ORM, SQLAIchemy...

本文主要介绍 AQLAIchemy 的连接配置使用

SQLAIchemy 的新建数据库表,新建数据库映射

文件结构

图

1 创建数据库的链接,及配置

这里我们将链接到数据库,并初始化一些配置

# database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

import pymysql
pymysql.install_as_MySQLdb()

# 指定路径
# SQLALCHEMY_DATABASE_URL = "mysql:///./hrs.sql"
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:xxxxxx@192.168.3.33:3306/comment_test"

# 建一个SQLAlchemy engine“引擎”
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    # connect_args={"check_same_thread": False}  # 仅适用于 AQLite
)


# 建立本地会话 类,该类的每个 实例都是一个数据库会话。
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


# 创建 Base类, 将用来创建每个数据库的模型MOdel (一般都是一个类
# 真正的项目中一般都是将此对象,抽离成一个单独的文件,同时定义一些 公共字段
Base = declarative_base()

2 创建数据库模型 ( ORM 映射

这里是建设 SQLAIchemy 与数据库的映射关系

# models.oy 
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


# 创建了 两个表的ORM 映射模型

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    # 官网的 演示是没有声明长度的。而我使用的 maridb必须声明 String(16)
    email = Column(String(16), unique=True, index=True)
    hashed_password = Column(String(128))
    is_active = Column(Boolean, default=True)

    # 关联两个表,SQLAIchemy 会自动按照外键约束处理。flask 中只用定义一个一般在表关系中 多 的那一个表
    # 四天后更新,主要原因是使用了  back_populates, 使用 backerf则只需要再一边定义就行
    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(128), index=True)
    description = Column(String(256), index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

3 创建 ORM 映射模型和 Pydantic(fastapi 的模型)的映射

为了避免 SQLAlchemy 模型和 Pydantic 模型之间的混淆,使用带有 Pydantic 模型的 schemas.py( 来自于 SQLAIchemy 模型的映射)文件, 而非是直接使用 数据库的映射文件。

同时也可以 细致定义一个表的某个或多个字段为返回值,而非是 整个表的全部字段

同时也避免了在使用两者时产生混淆

简单理解就是:将 ORM 映射表拆分成不同的字段进行拆分,以供路由 请求参数,及返回参数 的验证(筛选。

如:创建一个 user 只需要 emial 和 password, 而不需要 is_active ,所以对请求体使用 类 UserCreate。而创建成功则需要返回整个用户,但不需要返回 password 所以使用类User 此类继承自 UserBase 不具有参数 password.

# schemas.py 
from typing import List, Optional
from pydantic import BaseModel

# 注意属性的名字,是和ORM的属性名一样的

# 创建一个ItemBase和UserBasePydantic模型(或称“方案”)以在创建或读取数据时具有共同的属性。
class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None

# 创建一个ItemCreate和并UserCreate从它们继承(这样它们将具有相同的属性),以及创建所需的所有其他数据(属性)。
class ItemCreate(ItemBase):
    pass

# 创建Pydantic模型/模式以进行读取/返回, Item, User
class Item(ItemBase):
    id: int
    owner_id: int

    # **tips: 外部   
    class Config:
        # 将要做为 限制响应体 数据库ORM的数据时,请务必开启此配置
        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

tips: 此类 Config 用于为 Pydantic 提供配置

orm_mode = True 允许使用 ORM 模式: 可以将任意的类实例创建 Pydantic 模型,支持映射到 ORM 对象的模型。(白话就是可以将可以使用此类的一个方法(User.from_orm(BaseModel.User()),直接将 ORM 的数据转为 Pydantic 模型的数据)

例子: 内部中对于传输数据的取值假如是这样的 User(objName),

内部默认的取值方式: id = objName['id'].

因为设置了值为 True, 所以他还会尝试 id = ObjName.id

注意:如果设置的 Pydantic 模型将要作为 返回响应体(response_model )的限制, 同时返回的数据是 数据库ORM映射模型时。请必须设置 ore_mode属性。否则将会因为不能解析数据而报错

4 CRUD utils 于数据库操作的抽离

对于大部分情况,我们不会将对数据库的交互(增删查改),直接放在路由函数中。而是应该单独抽离出来,以便于以后的修改,及减少代码的重复量...

# curd.py

# 导入此参数,便于声明函数参数类型(已得到 语法提示,及 python声明形参类型的一些特性
from sqlalchemy.orm import Session

# models 交互数据库,schems声明证形参类型
from . import models, schemas


# 创建对数据库操作函数(类似原子类, 以减少路由操作函数中 对数据库操作的重复代码

def get_user(db: Session, user_id: int):
    # 查询此 id的用户
    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):
    # 添加用户
    # schemas.UserCreate 使用这个类来筛选参数?? python本身不会对声明了参数类型的参数做验证
    # 同时本身这个参数是由外部传入路由过来的对象,本身就能保证参数的正确
    # 经过简单的验证 python 对于声明了类型的形参不会做出验证
    fake_hashed_password = user.password + "notreallyhashed"
    # 获得 模型实例
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    # 添加到数据库会话中
    db.add(db_user)
    # 提交此处操作
    db.commit()
    # 刷新模型实例,以获得 数据库返回的 id等字段
    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

5 创建 FastApi 应用,以使用这些定义的数据映射

# mian.py 
from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

# 在主文件使用 . 相对路径导入 文件总是报错,所以此处使用绝对路径
# from . import crud, models, schemas
from spL_app import models, schemas, crud
# 导入 会话 及 引擎 
from spL_app.database import SessionLocal, engine

# 按照创建的 ORM模型,创建数据库表(存在则不创建,当然一般都是 使用数据库 迁移工具进行数据库的管理,此处仅为演示
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

对于路由函数的解释

在此处每个路由函数仅相当于对 需要的操作按照条件,转发至相应的 crud 中的函数中,并没有对数据库进行直接操作

# 定义返回体类型
@app.post("/users/", response_model=schemas.User)
# 接收 UserCreate中定义的字段 email,password 封装入 user, 同时使用 依赖,注入了一个 db 对象
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:
        # 查询到数据库中 已经存在了 此user ,直接返回错误 400
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)
  • response_model 声明了和限制了 响应体的数据 :

    • 将输出数据转换为其类型声明。
    • 将输出数据限制为模型的数据.
    • 在 OpenAPI 路径操作中为响应添加一个 JSON 模式。
    • 将由自动文档系统使用。

    基本在各种类型 路由函数装饰器上都可以使用此属性( get, post, ...

    可以的接收参数: Paydantic 模型, Paydantic 模型的列表( List[item])

  • db , 来自 Dependes(get_db) 依赖 注入的 数据库, 他将传递给相应的 crud 文件中的函数进行相应操作

6 总结

其实从头慢慢看下来,还是很简单的。就是有很多细节。所以 简单按步骤总结下

  1. database 建立数据库的连接,同时初始化了一些类: engine (连接数据库, SessionLocal (建立一次会话, Base(用于建立 ORM 模型

  2. models 建立数据库的映射模型,(可以手动创建然后再数据库中生成。或者使用 迁移工具从数据库映射到本地

  3. schemas.py 创建 Pydantic 模型, 此模型是根据 数据库的 ORM 映射而来(注:再程序里本身并无继承或者引用 models 的内容)

    • 此 Pydantic 模型建立的目的主要使用,请求参数的接收,及响应内容的返回。(并不会直接再 路由的形参列表里直接使用 ORM 来接收验证数据)
    • 注意当一个路由需要直接响应一个 ORM 对象实例时,必须要在 Pydantic 模型配置里面声明orm_mode = True
  4. curd.py 将对数据库的操作代码抽离出来,作为一个个的函数(这些函数只进行操作,不触及连接数据库及路由错误响应)

  5. main.py 路由文件(实际操作中,此文件 应该只用来进行 初始化,再将一些路由,依赖都抽离出去

    • 单个路由的参数请求参数和响应内容都是 使用的 pydantic 模型,而非是使用 数据库的 ORM
    • 此文件并没有直接操作数据库,而是使用 curd 定义一些函数(方法)。但是数据库的连接(会话)却是在路由中使用依赖打开的(解耦合,复用性

数据库迁移 Alembic 的使用

alembic 是 sqlalchemy 的作者开发的。用来做 OMR 模型与数据库的迁移与映射。alembic 使用方式跟 git 有点类似.

Alembic 使用 SQLAlchemy 作为数据库引擎,为关系型数据提供创建、管理、更改和调用的管理脚本,协助开发和运维人员在系统上线后对数据库进行在线管理。

1 安装

pip install alembic

2 创建迁移环境

alembic init alembic

生成了的文件目录如下:示例

3 配置 env.py 文件

  1. 导入数据库的 OMR

  2. 定义 数据库的连接

  3. 修改 env.py 配置文件,使用定义的数据库连接

# 省略了没有修改的代码,毕竟是在原基础上进行修改

# ……
import os
import sys

# 将上级目录添加 导入(import的搜索路径中,否则会找不到 包 spL_app. 迁移工具的运行位置为此文件当前目录
sys.path.append(os.path.dirname(os.pardir))


# 注意此处导入 Base的位置,如果使用 database里导入,则会因为检测不到ORM发生 错误
# 一般项目中的做方法是 专门用一个文件导入既定ORM类和Base 供此和其他地方导入
# from spL_app.database import Base
from spL_app.models import Base
target_metadata = Base.metadata

#... 

# 创建此函数以便于方便的修改 url(数据库地址, 如果直接将 url放在alembic.ini 中,则此处及下面都可以不用修改
def get_url():
    # 一般的项目中:是将这些配置放在配置文件中,然后导入至此
    SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:xxxxx@192.168.3.33:3306/comment_test"
    return SQLALCHEMY_DATABASE_URL

# 修改自动生成的两个函数,以 覆盖alembic.ini 中定义的 url. 

def run_migrations_offline():   
    # url = config.get_main_option("sqlalchemy.url")
    url = get_url()
    context.configure(
        url=url,
        ...
    )
    # ...

def run_migrations_online():
    # 增加三条语句和删除一条: 目的是 将配置文件中 url替换
    configuration = config.get_section(config.config_ini_section)  # 1
    configuration['sqlalchemy.url'] = get_url()  # 2
    connectable = engine_from_config(
        # config.get_section(config.config_ini_section), #del
        configuration,  # 3
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    # ...

sys.path.append:

python 程序中使用 import XXX 时, python 解析器会在当前目录、已安装和第三方模块中搜索 xxx ,如果都搜索不到就会报错.

使用 sys.path.append()方法可以临时添加搜索路径,方便更简洁的 import 其他包和模块。这种方法导入的路径会在 python 程序退出后失效。

4 生成迁移的脚本文件

alembic revision --autogenerate -m "msg"

5 提交此次迁移至数据库

alembic upgrade head

此后每次想要修改数据库表,在 ORM 的类里面修改后,进行 4 , 5 的操作就行了,注意每次生成的迁移文件必须提交,否则会因为版本的不同,而发生错误。

Tips: 这些配置都是基于上面的演示项目而来,所以在很多地方,都不是特别合理。

同时上面 main.py 里面的 models.Base.metadata.create_all(bind=engine) 这条代码,也可以删除不用了


下一篇,将介绍 如何将数据库的表映射至本地,并自动生成 ORM 文件


相关推荐:

来自系列:fastapi

分类 python下文章:

1.0 爬虫的介绍,和requests模块的简单使用

1.1 数据解析的三种方式。正则表达式, bs4, xpath

2.0 多任务(进程,协程,线程)爬虫:验证码识别,返回头储存,ip代理 介绍。 异步是什么,爬虫异步的方式。线程,进程,介绍

2.0.1 协程的 async/await 实现 爬虫 单线程 + 异步协程的实现

3.0 基于selenium 模块的 爬虫操作。 selenium 是一个用于Web应用程序测试的工具。Selenium测试直接运行在浏览器中,就像真正的用户在操作一样。

更多...

aid 加一试试->

评论([[comments.sum]])

发表

加载更多([[item.son.length - 2]])...

发表

2020-11 By Hchuan.

flask & bootstrap-flask

© 2021 HChuan. All rights reserved.

随机占位图来自:fghrsh

互联网ICP备案号:蜀ICP备2020031846号