FastAPI 结合 SQLAlchemy 操作 MySQL 数据库

文章目录

1. 安装 SQLAlchemy
2. 创建数据库
3. SQLAlchemy 连接 MySQL
4. 创建数据模型
5. 创建 Pydantic 模型
6. crud 工具
7. main函数

learning from 《python web开发从入门到精通》

1. 安装 SQLAlchemy

pip install sqlalchemy

2. 创建数据库

mysql -u root -p 命令行登录 MySQL

创建数据库 fastapi_db

mysql> create database fastapi_db default charset utf8mb4 collate utf8mb4_unicode_ci;
Query OK, 1 row affected (0.04 sec)

3. SQLAlchemy 连接 MySQL

database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base# 数据库连接配置
SQLALCHEMY_DATABASE_URI = ("mysql+pymysql://root:123456@localhost/fastapi_db?charset=utf8mb4"#                用户:密码@服务器/数据库?参数
)# 创建数据库引擎
engine = create_engine(SQLALCHEMY_DATABASE_URI)
# 创建数据库会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 声明基类
Base = declarative_base()

4. 创建数据模型

models.py

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Base# 定义 User 类
class User(Base):__tablename__ = 'users'  # 定义表名id = Column(Integer, primary_key=True, index=True)email = Column(String(255), unique=True, index=True)hashed_password = Column(String(255))is_active = Column(Boolean, default=True)items = relationship("Item", back_populates="owner")# 关联 Item 表# 定义 Item 类
class Item(Base):__tablename__ = "items"id = Column(Integer, primary_key=True, index=True)title = Column(String(255), index=True)description = Column(String(255), index=True)owner_id = Column(Integer, ForeignKey('users.id'))owner = relationship("User", back_populates="items")# 关联 User 表

relationship 还不懂，有待学习 SQLAlchemy

5. 创建 Pydantic 模型

schemas.py

from typing import List
from pydantic import BaseModelclass ItemBase(BaseModel):title: strdescription: str = Noneclass ItemCreate(ItemBase):passclass Item(ItemBase):id: intowner_id: intclass Config:orm_mode = Trueclass UserBase(BaseModel):email: strclass UserCreate(UserBase):password: strclass User(UserBase):id: intis_active: boolitems: List[Item] = []class Config:orm_mode = True

6. crud 工具

crud.py

from sqlalchemy.orm import Sessionfrom . import models, schemasdef get_user(db: Session, user_id: int):"""根据id获取用户信息:param db: 数据库会话:param user_id: 用户id:return: 用户信息"""return db.query(models.User).filter(models.User.id == user_id).first()def get_user_by_email(db: Session, email: str):"""根据email获取用户信息:param db: 数据库会话:param email: 用户email:return: 用户信息"""return db.query(models.User).filter(models.User.email == email).first()def get_users(db: Session, skip: int = 0, limit: int = 100):"""获取特定数量的用户:param db: 数据库会话:param skip: 开始位置:param limit: 限制数量:return: 用户信息列表"""return db.query(models.User).offset(skip).limit(limit).all()def create_user(db: Session, user: schemas.UserCreate):"""创建用户:param db: 数据库会话:param user: 用户模型:return: 根据email和password登录的用户信息"""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_userdef get_items(db: Session, skip: int = 0, limit: int = 100):"""获取指定数量的item:param db: 数据库会话:param skip: 开始位置:param limit: 限制数量:return: item列表"""return db.query(models.Item).offset(skip).limit(limit).all()def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):"""创建用户item:param db: 数据库会话:param item: Item对象:param user_id: 用户id:return: Item模型对象"""db_item = models.Item(**item.dict(), owner_id=user_id)db.add(db_item)db.commit()db.refresh(db_item)return db_item

7. main函数

from typing import Listfrom fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Sessionfrom . import crud, models, schemas
from .database import SessionLocal, enginemodels.Base.metadata.create_all(bind=engine)app = FastAPI()# 依赖
def get_db():try:db = SessionLocal()yield dbfinally:db.close()@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):# 根据email查找用户db_user = crud.get_user_by_email(db, email=user.email)# 如果用户存在，提示该邮箱已经被注册if db_user:raise HTTPException(status_code=400, detail="Email already registered")# 返回创建的user对象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)):# 获取当前id的用户信息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)
):# 创建该用户的itemsreturn 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)):# 获取所有itemsitems = crud.get_items(db, skip=skip, limit=limit)return items

(pt19) D:\web_python_dev>uvicorn fastapi_mysql.main:app --reload
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [6988] using watchgod
INFO:     Started server process [2112]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

mysql> use fastapi_db
Database changed
mysql> show tables;
+----------------------+
| Tables_in_fastapi_db |
+----------------------+
| items                |
| users                |
+----------------------+
2 rows in set (0.00 sec)

在网页里发送一个 post 请求后，查询 sql

mysql> select * from users;
+----+----------------+---------------------+-----------+
| id | email          | hashed_password     | is_active |
+----+----------------+---------------------+-----------+
|  1 | michael@xx.com | abcdnotreallyhashed |         1 |
+----+----------------+---------------------+-----------+
1 row in set (0.00 sec)

FastAPI 结合 SQLAlchemy 操作 MySQL 数据库相关推荐

【tips】ORM - SQLAlchemy操作MySQL数据库
优先(官方文档SQLAlchemy-version1.2): sqlalchemy | 作者:斯芬克斯推荐一(长篇幅version1.2.0b3):python约会之ORM-sqlalchemy | ...
Python FastAPI 框架操作Mysql数据库增删改查
2 比 1 更容易理解,可以先看2(单文件级别) 1.FastAPI 框架操作Mysql数据库(项目多文件级别) FastAPI 可以使用任何您想要的关系型数据库. 在这里,让我们看一个使用着SQL ...
sqlalchemy 使用mysql_使用SQLALchemy连接MySql数据库并进行操作
"导语:这期主要学一下利用SQLALchemy连接MySql数据库以及对数据库的增删改查的操作,可能代码部分很多还不能理解,我觉得像一些固定写法先用着(比如引入的类等),再慢慢体会缘由&qu ...
pandas读取大文件（chunksize）并通过sqlalchemy写入MySQL数据库
pandas读取大文件(chunksize)并通过sqlalchemy写入MySQL数据库在pandas中读取表类文件的时候有一个参数chunksize,只要指定了这个参数的数值,那么得到的结果就不 ...
sqlalchemy mysql教程_python使用SQLAlchemy操作MySQL
SQLAlchemy是Python编程语言下的一款开源软件,提供了SQL工具包及对象关系映射(ORM)工具,使用MIT许可证发行.SQLAlchemy首次发行于2006年2月,并迅速地在Python社 ...
sqlalchemy mysql_使用SQLAlchemy操作MySQL
SQLAlchemy是Python编程语言下的一款开源软件,提供了SQL工具包及对象关系映射(ORM)工具,使用MIT许可证发行.SQLAlchemy首次发行于2006年2月,并迅速地在Python社 ...
python mysql工具类_Python工具类（一）—— 操作Mysql数据库
如何调用直接看__main__函数里如何调用此工具类就阔以啦! # encoding=utf-8 import pymysql # 导入所有Mysql配置常量,请自行指定文件 from conf.se ...
c语言连接数据库例子,c语言操作mysql数据库小例子_互帮互助（C language MySQL database operation example _ mutual help）.doc...
这是精心收集的精品经典资料,值得下载保存阅读! c语言操作mysql数据库小例子_互帮互助(C language MySQL database operation example _ mutual h ...
c 获取mysql列数据_转用C API 操作MySQL数据库
用C API 操作MySQL数据库参考MYSQL的帮助文档整理这里归纳了C API可使用的函数,并在下一节详细介绍了它们.请参见25.2.3节,"C API函数描述". 函数 ...

FastAPI 结合 SQLAlchemy 操作 MySQL 数据库

文章目录

1. 安装 SQLAlchemy

2. 创建数据库

3. SQLAlchemy 连接 MySQL

4. 创建数据模型

5. 创建 Pydantic 模型

6. crud 工具

7. main函数

FastAPI 结合 SQLAlchemy 操作 MySQL 数据库相关推荐

最新文章

热门文章