FastAPI整合MySQL实现商品管理API

前言

学无止境，无止境学。坚持每天学点编程知识，坚持每天写点小文章，坚持每天进步一点点。大家好，我是张大鹏，喜欢学习和分享，希望能够通过此公众号，将自己学到的东西分享给大家，和大家一起交流，一起成长，一起进步。

乾坤未定，你我皆是黑马。大鹏一日同风起，扶摇直上九万里。宝剑锋从磨砺出，梅花香自苦寒来。不积跬步，无以至千里，不积小流无以成江海。

如果有多余的时间，就坚持学习吧，小时候学习改变命运，长大了学习丰富内涵，老了学习带来宁静。活到老，学到老，我能行！

课程介绍

《Python+Vite+Vue3前后端分类企业商城系统》实战课

本教程属于后端部分第5个教程，之前的教程还有：

《Python异步接口开发框架FastAPI入门教程》

《FastAPI实现Oauth2和JWT登录》

《FastAPI组件化实现Oauth2和JWT登录》

《FastAPI实现商品管理API》

peewee基本用法

安装

pip install peewee

pip install pymysql

创建模型

import pymysql

from peewee import *

from playhouse.db_url import connect

pymysql.install_as_MySQLdb()

db = connect('mysql://root:root@localhost:3306/gaoxiang_shop_admin')

class Goods(Model):

name = CharField()

price = FloatField()

class Meta:

database = db

db.connect()

创建表

# 创建表格

try:

db.create_tables([Goods])

except Exception as e:

pass

新增商品

# 创建商品

goods = Goods(name='《React学习手册》', price=128)

goods.save()

根据ID查询商品

# 根据ID查询

goods = Goods.get(1)

print(goods.name, goods.price)

修改商品

# 修改

goods = Goods.get(1)

goods.name = "《React学习手册111》"

goods.save()

分页查询商品

# 分页查询

data = Goods.select().paginate(2, 2)

print(list(data))

完整代码

import pymysql

from peewee import *

from playhouse.db_url import connect

pymysql.install_as_MySQLdb()

db = connect('mysql://root:root@localhost:3306/gaoxiang_shop_admin')

class Goods(Model):

name = CharField()

price = FloatField()

class Meta:

database = db

db.connect()

# 创建表格

try:

db.create_tables([Goods])

except Exception as e:

pass

# 创建商品

goods = Goods(name='《React学习手册》', price=128)

goods.save()

# 根据ID查询

goods = Goods.get(1)

print(goods.name, goods.price)

# 修改

goods = Goods.get(1)

goods.name = "《React学习手册111》"

goods.save()

goods = Goods.get(1)

print(goods.name, goods.price)

# 分页查询

data = Goods.select().paginate(1, 2)

print(list(data))

基于MySQL的商品管理接口

数据准备

import pymysql

from fastapi import FastAPI

from peewee import *

from playhouse.db_url import connect

from pydantic import BaseModel

pymysql.install_as_MySQLdb()

db = connect('mysql://root:root@localhost:3306/gaoxiang_shop_admin')

app = FastAPI()

class GoodsModel(Model):

name = CharField()

price = FloatField()

class Meta:

database = db

db_table = "goods"

class GoodsSchema(BaseModel):

name: str # 商品名称

price: float # 商品价格

def get_response(data=None):

response = {"code": 10000, "msg": "success", "status": True}

if data is not None:

response["data"] = data

return response

def get_response_error(code, msg):

return {"code": code, "msg": msg, "status": False}

新增商品接口

@app.post("/goods")

async def post_goods(goods: GoodsSchema):

try:

goods_model = GoodsModel(**goods.dict())

goods_model.save()

return get_response()

except Exception as e:

print(e)

return get_response_error(10003, "保存商品数据失败")

根据ID获取商品接口

@app.get("/goods/")

async def get_goods_by_id(gid: int):

try:

goods = GoodsModel.get(gid)

return get_response(goods.__data__)

except Exception as e:

print(e)

return get_response_error(10004, "商品不存在")

@app.get("/goods")

async def get_goods(page: int = 1, size: int = 20):

query = GoodsModel.select()

count = query.count()

result = query.paginate(page, size).dicts()

data = {

"total": count,

"data": list(result)

}

return get_response(data)

修改商品接口

@app.put("/goods/")

async def put_goods(gid: int, goods: GoodsSchema):

try:

goods_model = GoodsModel.get(gid)

goods_model.name = goods.name

goods_model.price = goods.price

goods_model.save()

return get_response()

except Exception as e:

return get_response_error(10004, "商品不存在")

删除商品接口

@app.delete("/goods/")

async def delete_goods(gid: int):

try:

GoodsModel.get(gid).delete_instance()

except Exception as e:

pass

return get_response()

完整代码

import pymysql

from fastapi import FastAPI

from peewee import *

from playhouse.db_url import connect

from pydantic import BaseModel

pymysql.install_as_MySQLdb()

db = connect('mysql://root:root@localhost:3306/gaoxiang_shop_admin')

app = FastAPI()

class GoodsModel(Model):

name = CharField()

price = FloatField()

class Meta:

database = db

db_table = "goods"

class GoodsSchema(BaseModel):

name: str # 商品名称

price: float # 商品价格

def get_response(data=None):

response = {"code": 10000, "msg": "success", "status": True}

if data is not None:

response["data"] = data

return response

def get_response_error(code, msg):

return {"code": code, "msg": msg, "status": False}

# http://localhost:8000/docs

# http://localhost:8000/goods

@app.post("/goods")

async def post_goods(goods: GoodsSchema):

try:

goods_model = GoodsModel(**goods.dict())

goods_model.save()

return get_response()

except Exception as e:

print(e)

return get_response_error(10003, "保存商品数据失败")

@app.get("/goods/")

async def get_goods_by_id(gid: int):

try:

goods = GoodsModel.get(gid)

return get_response(goods.__data__)

except Exception as e:

print(e)

return get_response_error(10004, "商品不存在")

# http://localhost:8000/goods

@app.get("/goods")

async def get_goods(page: int = 1, size: int = 20):

query = GoodsModel.select()

count = query.count()

result = query.paginate(page, size).dicts()

data = {

"total": count,

"data": list(result)

}

return get_response(data)

@app.put("/goods/")

async def put_goods(gid: int, goods: GoodsSchema):

try:

goods_model = GoodsModel.get(gid)

goods_model.name = goods.name

goods_model.price = goods.price

goods_model.save()

return get_response()

except Exception as e:

return get_response_error(10004, "商品不存在")

@app.delete("/goods/")

async def delete_goods(gid: int):

try:

GoodsModel.get(gid).delete_instance()

except Exception as e:

pass

return get_response()

if __name__ == '__main__':

import uvicorn

uvicorn.run("test:app")

总结