深圳龙华网站开发,页面跳转自动更新,凌晨三点看的片免费,杭州小程序制作公司排行榜pip install mysql-connector-python pymysql
数据库链接
创建src目录#xff0c;里面创建db.py 代码如下#xff1a;
# 导入mysql.connector模块#xff0c;该模块提供了与MySQL数据库进行连接和交互的功能。
import mysql.connector # 定义一个函数get_db_connectio…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表示数据库服务器运行在本机上。 hostlocalhost, # port参数指定数据库服务器监听的端口号MySQL默认端口是3306。 port3306, # user参数指定用于连接到数据库的用户名这里使用的是root用户。 userroot, # password参数指定连接数据库的密码出于安全考虑实际使用时密码不应明文写在代码中。 passwordXXXXXXX, # database参数指定要连接到的数据库名称这里使用的是名为database的数据库。 databasedatabase ) # 返回创建的数据库连接对象以供后续操作使用。 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
app.get(/)
async def root(): return {message: Hello World} # 定义带参数的路由返回带有输入名字的问候语
app.get(/hello/{name})
async def say_hello(name: str): return {message: fHello {name}} # 定义关于信息的路由返回应用名称和版本
app.get(/about)
async def about(): 项目信息 :return: 返回包含应用名称和版本信息的字典 return { app_name: 人工智能识别, app_version: v0.0.1 } # 定义获取所有用户的路由从数据库中查询并返回用户列表
app.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获取单个用户的路由
app.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/或其他更有意义的名称
app.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, host0.0.0.0, port8000) # 另一种运行方式使用命令行参数--reload实现热重载 # uvicorn main:app --reload
