pip install mysql-connector-python pymysql
数据库链接
创建src目录,里面创建db.py 代码如下:
# 导入mysql.connector模块,该模块提供了与MySQL数据库进行连接和交互的功能。
import mysql.connector # 定义一个函数get_db_connection,该函数用于获取与MySQL数据库的连接。
def get_db_connection(): # 使用mysql.connector.connect()方法创建一个数据库连接对象。 # 这个方法需要几个关键参数来指定如何连接到数据库。 connection = mysql.connector.connect( # host参数指定数据库服务器的地址,这里使用'localhost'表示数据库服务器运行在本机上。 host='localhost', # port参数指定数据库服务器监听的端口号,MySQL默认端口是3306。 port=3306, # user参数指定用于连接到数据库的用户名,这里使用的是"root"用户。 user="root", # password参数指定连接数据库的密码,出于安全考虑,实际使用时密码不应明文写在代码中。 password="XXXXXXX", # database参数指定要连接到的数据库名称,这里使用的是名为"database"的数据库。 database="database" ) # 返回创建的数据库连接对象,以供后续操作使用。 return connectiondef get_db():connection = get_db_connection()db = connection.cursor()try:yield dbfinally:db.close()connection.close()
main.py代码
# 导入FastAPI框架及相关依赖
from fastapi import FastAPI, Depends, Header, HTTPException
# 导入uvicorn,用于运行FastAPI应用
import uvicorn
# 从mysql.connector导入cursor,但这里可能存在问题,通常我们会导入connector然后从中获取cursor
from mysql.connector import cursor
# 导入自定义的数据库连接函数
from src.db import get_db # 初始化FastAPI应用实例
app = FastAPI() # 定义根路由,返回"Hello World"
.get("/")
async def root(): return {"message": "Hello World"} # 定义带参数的路由,返回带有输入名字的问候语
.get("/hello/{name}")
async def say_hello(name: str): return {"message": f"Hello {name}"} # 定义关于信息的路由,返回应用名称和版本
.get("/about")
async def about(): """ 项目信息 :return: 返回包含应用名称和版本信息的字典 """ return { "app_name": "人工智能识别", "app_version": "v0.0.1" } # 定义获取所有用户的路由,从数据库中查询并返回用户列表
.get("/users/")
async def get_users(db: cursor.MySQLCursor = Depends(get_db)): query = "SELECT * FROM view_set" db.execute(query) result = db.fetchall() if result: return {"users": result} else: return {"error": "User not found"} # 定义根据用户ID获取单个用户的路由
.get("/users/{user_id}")
async def get_user(user_id: int, db: cursor.MySQLCursor = Depends(get_db)): query = "SELECT * FROM view_set WHERE id = %s" db.execute(query, (user_id,)) result = db.fetchall() if result: return {"user_id": result[0][0], "username": result[0][1]} else: return {"error": "User not found"} # 定义插入用户的路由,但路由名称可能应该为/add_user/或其他更有意义的名称
.get("/user_name/{user_name}")
async def insert_user(user_name: str, db: cursor.MySQLCursor = Depends(get_db)): query = "INSERT INTO view_set (name) VALUES (%s)" db.execute(query, (user_name,)) # 注意:通常在执行INSERT操作后,我们不需要调用fetchone(),因为INSERT不返回结果集 # result = db.fetchone() # 这行代码是多余的 db.execute("COMMIT") # 注意:这里可能存在问题,通常我们不会在每个查询后都执行COMMIT,而是在适当的时候 return {"user_name": user_name} # 主程序入口
if __name__ == "__main__": # 使用uvicorn运行FastAPI应用 uvicorn.run(app, host="0.0.0.0", port=8000) # 另一种运行方式,使用命令行参数--reload实现热重载 # uvicorn main:app --reload