备案控制台
开发者社区 数据库 文章 正文

FastAPI数据库系列(一) MySQL数据库操作 一、简介

2022-11-19 700
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云数据库 RDS MySQL Serverless，0.5-2RCU 50GB
推荐场景：
学生管理系统数据库设计 搭建个人博客
云数据库 RDS MySQL Serverless，价值2615元额度，1个月
简介： FastAPI中你可以使用任何关系型数据库，可以通过SQLAlchemy将其轻松的适应于任何的数据库，比如：PostgreSQLMySQLSQLiteOracleMicrosoft SQL Server...

一、简介
FastAPI中你可以使用任何关系型数据库，可以通过SQLAlchemy将其轻松的适应于任何的数据库，比如：

PostgreSQL
MySQL
SQLite
Oracle
Microsoft SQL Server
...
　　SQLAlchemy是一个ORM（object-relational mapping）的框架。在ORM中，你创建一个类就会通过SQLAlchemy将其自动转成一张表，在类中的每一个属性就会将其转成表中的字段。

这里有一些实例，假如有一个大的项目，里面包含一个子包叫做sql_app：

复制代码
.
└── sql_app

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

复制代码
__init__.py 是一个空文件，但是说明sql_app是一个package
database.py 数据库配置相关
models.py 数据库模型表
schemas.py 模型验证
crud.py 数据库操作相关
main.py 主文件
二、简单实例
该实例以MySQL为例，SQLAlchemy需要借助于pymysql连接数据库，所以需要进行安装这两个工具包：

pip install sqlalchemy
pip install pymysql
1、database.py
复制代码
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@127.0.0.1:3306/test"

echo=True表示引擎将用repr()函数记录所有语句及其参数列表到日志

engine = create_engine(

SQLALCHEMY_DATABASE_URL, encoding='utf8', echo=True

)

SQLAlchemy中，CRUD是通过会话进行管理的，所以需要先创建会话，

每一个SessionLocal实例就是一个数据库session

flush指发送到数据库语句到数据库，但数据库不一定执行写入磁盘

commit是指提交事务，将变更保存到数据库文件中

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

创建基本映射类

Base = declarative_base()
复制代码
在数据库相关的配置文件中，首先创建一个SQLAlchemy的"engine"，然后创建SessionLocal实例进行会话，最后创建模型类的基类。

2、models.py
复制代码
from sqlalchemy import Boolean, Column, Integer, String
from database import Base

class User(Base):

__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String(32), unique=True, index=True)
hashed_password = Column(String(32))
is_active = Column(Boolean, default=True)

复制代码
通过数据库配置文件中的基类来创建模型类。

3、schemas.py
复制代码
from pydantic import BaseModel

class UserBase(BaseModel):

email: str

class UserCreate(UserBase):

"""
请求模型验证：
email:
password:
"""
password: str

class User(UserBase):

"""
响应模型：
id:
email:
is_active
并且设置orm_mode与之兼容
"""
id: int
is_active: bool

class Config:
    orm_mode = True

复制代码
　　定义请求参数模型验证与响应模型验证的Pydantic模型，其中响应模型中设置orm_mode=True参数，表示与ORM模型兼容，因为后续中返回的数据库查询是orm模型，通过设置这个参数可以将orm模型通过pydantic模型进行验证。

4、crud.py
复制代码
from sqlalchemy.orm import Session
import models, schemas

通过id查询用户

def get_user(db: Session, user_id: int):

return db.query(models.User).filter(models.User.id == user_id).first()

新建用户

def db_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

复制代码
通过传入数据库连接以及参数等进行数据库操作，包括创建用户、查询用户等，返回的是orm模型对象。

5、main.py
复制代码
from fastapi import FastAPI, Depends, HTTPException
import crud, schemas
from database import SessionLocal, engine, Base
from sqlalchemy.orm import Session
import uvicorn

Base.metadata.create_all(bind=engine) #数据库初始化，如果没有库或者表，会自动创建

app = FastAPI()

Dependency

def get_db():

"""
每一个请求处理完毕后会关闭当前连接，不同的请求使用不同的连接
:return:
"""
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)):

return crud.db_create_user(db=db, user=user)

通过id查询用户

@app.get("/user/{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 not db_user:
    raise HTTPException(status_code=404, detail="User not found")
return db_user

if name == '__main__':

uvicorn.run(app=app, host="127.0.0.1", port=8000)

复制代码
主文件进行数据库初始化、FastAPI实例创建以及处理各种请求。

进入到交互文档查看：

http://127.0.0.1:8000/users/
复制代码

请求

{
"email": "hhh@example113.com",
"password": "ss123456"
}

响应

{
"email": "hhh@example113.com",
"id": 7,
"is_active": true
}
复制代码
http://127.0.0.1:8000/user/7
复制代码

响应

{
"email": "hhh@example113.com",
"id": 7,
"is_active": true
}
复制代码
三、复杂实例
在之前的基础上再加一个模型类Item，User与之是一对多的关系。

1、models.py
复制代码
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey
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(32), unique=True, index=True)
hashed_password = Column(String(32))
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(32), index=True)
description = Column(String(32), index=True)
owner_id = Column(Integer, ForeignKey("users.id"))

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

复制代码
2、schemas.py
复制代码
from typing import Optional,List
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):

"""
请求模型验证：
email:
password:
"""
password: str

class User(UserBase):

"""
响应模型：
id:
email:
is_active
并且设置orm_mode与之兼容
"""
id: int
is_active: bool
items: List[Item] = []

class Config:
    orm_mode = True

复制代码
3、crud.py
复制代码
from sqlalchemy.orm import Session
import models, schemas

通过id查询用户

def get_user(db: Session, user_id: int):

return db.query(models.User).filter(models.User.id == user_id).first()

新建用户

def db_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

获取用户拥有的item

def get_item(db: Session, skip: int = 0, limit: int = 100):

return db.query(models.Item).offset(skip).limit(limit).all()

新建用户的item

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

复制代码
4、main.py
复制代码
from typing import List
from fastapi import FastAPI, Depends, HTTPException
import crud, schemas
from database import SessionLocal, engine, Base
from sqlalchemy.orm import Session
import uvicorn

Base.metadata.create_all(bind=engine)

app = FastAPI()

Dependency

def get_db():

"""
每一个请求处理完毕后会关闭当前连接，不同的请求使用不同的连接
:return:
"""
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)):

return crud.db_create_user(db=db, user=user)

通过id查询用户

@app.get("/user/{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 not db_user:
    raise HTTPException(status_code=404, detail="User not found")
return db_user

读取用户拥有的item

@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 0, db: Session = Depends(get_db)):

items = crud.get_item(db=db, skip=skip, limit=limit)
return items

创建用户的item

@app.post("/users/{user_id}/items", response_model=schemas.Item)
def create_item_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)

if name == '__main__':

uvicorn.run(app=app, host="127.0.0.1", port=8000)

复制代码
当启动项目后，会生成新的Item数据表，以及与User表之间建立关系：

复制代码

User表

create table users
(

id              int auto_increment
    primary key,
email           varchar(32) null,
hashed_password varchar(32) null,
is_active       tinyint(1)  null,
constraint ix_users_email
    unique (email)

);

create index ix_users_id

on users (id);

Item表

create table items
(

id          int auto_increment
    primary key,
title       varchar(32) null,
description varchar(32) null,
owner_id    int         null,
constraint items_ibfk_1
    foreign key (owner_id) references users (id)

);

create index ix_items_description

on items (description);

create index ix_items_id

on items (id);

create index ix_items_title

on items (title);

create index owner_id

on items (owner_id);

复制代码
最后进入交互文档进行测试。

文章标签：
云数据库 RDS MySQL 版
Python
SQL
关系型数据库
数据库管理
MySQL
Oracle
PostgreSQL
数据库
数据库连接
关键词：
云数据库 RDS MySQL 版数据库
数据库操作
数据库mysql
云数据库 RDS MySQL 版操作
数据库mysql数据库
相关实践学习
基于CentOS快速搭建LAMP环境
本教程介绍如何搭建LAMP环境，其中LAMP分别代表Linux、Apache、MySQL和PHP。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心，利用清洁计算为万物互联的新世界提供源源不断的能源动力，目前开服的区域包括中国（华北、华东、华南、香港）、新加坡、美国（美东、美西）、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程，来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品：云服务器ECS 云服务器 ECS（Elastic Compute Service）是一种弹性可伸缩的计算服务，助您降低 IT 成本，提升运维效率，使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
游客t5nljtsvdwfa4
目录
相关文章
宁波阿成.
|
3天前
|
关系型数据库 MySQL 数据库
docker MySQL删除数据库时的错误（errno: 39）
docker MySQL删除数据库时的错误（errno: 39）
宁波阿成.
10 0
默不会开发
|
5天前
|
存储 数据处理 数据库
深入了解达梦数据库的增删查改操作：从入门到精通
深入了解达梦数据库的增删查改操作：从入门到精通
默不会开发
16 4
xin在这
|
13天前
|
DataWorks 关系型数据库 调度
DataWorks操作报错合集之DataWorks 数据库同步任务中，如果遇到表情符报错的现象，怎么处理
DataWorks是阿里云提供的一站式大数据开发与治理平台，支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中，可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
xin在这
29 0
xin在这
|
13天前
|
DataWorks Oracle 关系型数据库
DataWorks操作报错合集之尝试从Oracle数据库同步数据到TDSQL的PG版本，并遇到了与RAW字段相关的语法错误，该怎么处理
DataWorks是阿里云提供的一站式大数据开发与治理平台，支持数据集成、数据开发、数据服务、数据质量管理、数据安全管理等全流程数据处理。在使用DataWorks过程中，可能会遇到各种操作报错。以下是一些常见的报错情况及其可能的原因和解决方法。
xin在这
30 0
火威网络
|
1天前
|
关系型数据库 MySQL 数据库
mysql 设置环境变量与未设置环境变量连接数据库的区别
设置与未设置MySQL环境变量在连接数据库时主要区别在于命令输入方式和系统便捷性。设置环境变量后，可直接使用`mysql -u 用户名 -p`命令连接，而无需指定完整路径，提升便利性和灵活性。未设置时，需输入完整路径如`C:\Program Files\MySQL\...`，操作繁琐且易错。为提高效率和减少错误，推荐安装后设置环境变量。[查看视频讲解](https://www.bilibili.com/video/BV1vH4y137HC/)。
火威网络
17 3
mysql 设置环境变量与未设置环境变量连接数据库的区别
1529723250799936
|
3天前
|
Oracle 关系型数据库 Java
java操作多数据源将oracle数据同步达梦数据库
java操作多数据源将oracle数据同步达梦数据库
1529723250799936
9 1
宁波阿成.
|
3天前
|
关系型数据库 MySQL 数据库连接
用Navicat备份Mysql演示系统数据库的时候出：Too Many Connections
用Navicat备份Mysql演示系统数据库的时候出：Too Many Connections
宁波阿成.
10 0
源码星辰
|
5天前
|
存储 Oracle 关系型数据库
oracle 数据库 迁移 mysql数据库
将 Oracle 数据库迁移到 MySQL 是一项复杂的任务，因为这两种数据库管理系统具有不同的架构、语法和功能。
源码星辰
15 0
然然学长
|
5天前
|
SQL Java 数据库连接
Java数据库编程实践：连接与操作数据库
Java数据库编程实践：连接与操作数据库
然然学长
9 0
阿六_Dexter
|
7天前
|
SQL JSON 关系型数据库
[UE虚幻引擎插件DTPostgreSQL] PostgreSQL Connector 使用蓝图连接操作 PostgreSQL 数据库说明
本插件主要是支持在UE蓝图中连接和操作PostgreSQL 数据库。
阿六_Dexter
16 2

热门文章

最新文章

  • 1
    嵌入式数据库sqlite3【基础篇】基本命令操作，小白一看就懂（C/C++）
  • 2
    阿里云数据库使用方法，从购买、创建数据库账号密码到连接数据库全流程
  • 3
    TiDB与MySQL、PostgreSQL等数据库的比较分析
  • 4
    TiDB的优势：为何选择TiDB作为您的数据库解决方案
  • 5
    Windows公网远程连接MongoDB数据库【无公网IP】
  • 6
    宝塔数据库崩溃解决方案详解
  • 7
    【mysql】—— 数据库的操作
  • 8
    悦数图数据库推出 AI 知识图谱构建器及图语言生成助手
  • 9
    OceanBase数据库的主备库参照的配置文件
  • 10
    面对多种国产数据库，如何选择合适的技术栈和发展方向
  • 1
    连接和管理RDS
    30
  • 2
    购买和初始化阿里云RDS
    30
  • 3
    rds教程
    28
  • 4
    TiDB与MySQL、PostgreSQL等数据库的比较分析
    189
  • 5
    OBCP实践 - 迁移 MySQL 数据到 OceanBase 集群
    51
  • 6
    Flink CDC产品常见问题之mysql整库同步到starrock时任务挂掉如何解决
    48
  • 7
    TIDB和MySQL的区别
    213
  • 8
    Flink CDC产品常见问题之mysql整库同步到starrock时任务挂掉如何解决
    75
  • 9
    [MySQL]事务原理之redo log,undo log
    127
  • 10
    [MySQL]SQL优化之索引的使用规则
    42

    • 相关课程

    更多
  • 数据库的前世今生
  • 数据库核心概念
  • 从传统数据库到云数据库演进
  • 数据库常见问题排查
  • 数据库及SQL/MySQL基础
  • 高校精品课-西安交通大学 -数据库理论与技术

    • 相关电子书

    更多
  • 2022 DTCC-阿里云一站式数据库上云最佳实践
  • 云时代的数据库技术趋势
  • 超大型金融机构国产数据库全面迁移成功实践

    • 相关实验场景

    更多
  • MySQL引擎及架构优化
  • PolarDB for AI：在数据库中通过SQL实现AI能力
  • 云原生HTAP数据库，让你的交易和分析一库搞定
  • Excel文件转存到RDS数据库
  • 快速体验PolarDB开源数据库
  • 5分钟上手Flink MySQL连接器

    • 推荐镜像

    更多
  • mysql
  • mariadb
  • mongodb
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）