尧图网站建设 
demo03 采购管理 采购单号 全局唯一 采购日期 商品名称 商品数量 采购总金额 状态待审核、已下单、已入库、已取消 增 post json格式的参数 /purchase_add 删 post json格式的参数 /purchase_delete 改 post json格式的参数 /purchase_modify 全量查 get /purchase_list 根据采购单号查询 get 依赖注入 采购增删改查的功能血药登录后才能使用每个接口需要校验用户是否登录。 将是否登录提取出来作为一个公共的函数其他接口中使用 采购增删改查的功能需要连接数据库使用完成后断开连接 将数据库连接提取出来作为一个公共的函数其他的 import uuid import pymysql import uvicorn from fastapi import FastAPI, Form, Header from fastapi.params import Depends from langgraph_sdk.auth.exceptions import HTTPException from pydantic import BaseModel, Field from starlette.responses import HTMLResponse app FastAPI() # 创建一个实例 class PurchaseMode(BaseModel): id: str Field(...,min_length16,max_length16,description采购单号)# ... 必填项 date: str product: str Field(...,min_length2,max_length16,description商品名称)# ... 必填项 num: int Field(gt0,description数量) totalAmount: float status: str class DeleteMode(BaseModel): id: str Field(...,min_length16,max_length16,description采购单号) DB { host: 127.0.0.1, port: 3306, user: root, password: 123456, charset: utf8 } class UserModel(BaseModel): phone:str Field(...,min_length11,max_length11,description手机号码) pwd:str Field(...,min_length6,max_length18,description密码) def init_db(): try: db pymysql.connect(**DB) # 解包 keyvalue,keyvalue # db pymysql.connect(host127.0.0.1,port 3306,user root,password 123456,) cursor db.cursor() cursor.execute(fcreate database if not exists fastapi38 character set utf8;) db.commit() db.select_db(fastapi38) cursor.execute(fcreate table if not exists purchase (id varchar(16) primary key, date date, fproduct varchar(16) not null, num int, totalAmount float, status varchar(16));) cursor.execute(create table if not exists user(id int primary key auto_increment, phone varchar(11), pwd varchar(20), token varchar(32));) db.commit() cursor.close() db.close() print(初始化数据库完成) except Exception as e: print(f初始化数据库异常{e}) #####依赖1数据库连接 def get_db(): db pymysql.connect(**DB) db.select_db(fastapi38) yield db db.close() ##### 依赖2是否登录 def check_login(token: str Header(None)): if token is None: return {msg: 请先登录,code: 20101} db pymysql.connect(**DB) db.select_db(fastapi38) cursor db.cursor() sql select * from user where token %s cursor.execute(sql,(token,)) a cursor.fetchone() if a is not None: return True else: raise HTTPException(status_code401,detail登录状态已经失效) app.post(/user/register) def register(data:UserModel, db: pymysql.Connection Depends(get_db)): cursor db.cursor() sql select * from user where phone %s cursor.execute(sql, (data.phone,)) a cursor.fetchone() if a is not None: return {msg: 手机号已被注册, code: 20101} else: sql insert into user (phone, pwd) values (%s, %s) cursor.execute(sql, (data.phone, data.pwd)) db.commit() return {msg: 注册成功, code: 10001} app.post(/user/login) def login(data:UserModel, db: pymysql.Connection Depends(get_db)): cursor db.cursor() sql select * from user where phone %s and pwd %s cursor.execute(sql, (data.phone,data.pwd,)) a cursor.fetchone() if a is not None: token str(uuid.uuid4()).replace(-,) # 生成一个token,表示用户登录状态 print(token,token) cursor.execute(update user set token %s where phone %s and pwd %s,(token,data.phone,data.pwd)) db.commit() return {msg: 登录成功, code: 10001,token: token} else: return {msg: 手机号或密码错误, code: 20101} app.post(/purchase_add,dependencies[Depends(check_login)]) def add(data: PurchaseMode,db: pymysql.Connection Depends(get_db)): cursor db.cursor() # 查询结果是元祖 sql select * from purchase where id %s cursor.execute(sql,(data.id,)) a cursor.fetchone() if a is not None: return {msg: 采购单号重复, code: 20101} else: sql insert into purchase (id, date, product, num, totalAmount, status) values (%s, %s, %s, %s, %s, %s) cursor.execute(sql,(data.id,data.date,data.product,data.num,data.totalAmount,data.status)) db.commit() cursor.close() return {msg: 添加成功, code: 10001} app.post(/purchase_delete, dependencies[Depends(check_login)]) def delete(data: DeleteMode, db: pymysql.Connection Depends(get_db)): cursor db.cursor() sql select * from purchase where id %s cursor.execute(sql, (data.id,)) a cursor.fetchone() if a is None: cursor.close() db.close() return {msg: 采购单号不存在, code: 20102} else: sql1 delete from purchase where id%s cursor.execute(sql1,(data.id,)) db.commit() cursor.close() return {msg: 删除成功, code: 10001} app.post(/purchase_modify, dependencies[Depends(check_login)]) def modify(data: PurchaseMode, db: pymysql.Connection Depends(get_db)): cursor db.cursor() cursor.execute(select * from purchase where id%s, (data.id,)) a cursor.fetchone() if a is None: cursor.close() db.close() return {msg:单据不存在无法修改,code:20103} else: sql update purchase set date%s, product%s, num%s, totalAmount%s, status%s where id%s cursor.execute(sql,(data.date,data.product,data.num,data.totalAmount,data.status,data.id)) db.commit() cursor.close() return {msg:修改成功,code:10001} app.get(/purchase_list, dependencies[Depends(check_login)]) def purchase_list(db: pymysql.Connection Depends(get_db)): cursor db.cursor(pymysql.cursors.DictCursor) sql select * from purchase cursor.execute(sql) a cursor.fetchall() cursor.close() return {msg: 查询成功, code: 10001, data: a} app.get(/purchase_get,dependencies[Depends(check_login)]) def get_one(id:str, db: pymysql.Connection Depends(get_db)): cursor db.cursor(pymysql.cursors.DictCursor) sql select * from purchase where id %s cursor.execute(sql, (id,)) a cursor.fetchone() cursor.close() if a is None: return {msg: 查询失败,无该采购单号, code: 20104} else: return {msg: 查询成功, code: 10001, data:a} # 页面路由读取static文件夹html文件 app.get(/register.html, response_classHTMLResponse) def page_register(): with open(static/register.html, r, encodingutf-8) as f: return f.read() app.get(/login.html, response_classHTMLResponse) def page_login(): with open(static/login.html, r, encodingutf-8) as f: return f.read() app.get(/purchase.html,response_classHTMLResponse) def purchase_page(): with open(static/purchase.html, r, encodingutf-8) as f: return f.read() if __name__ __main__: init_db() # 初始化数据库 uvicorn.run(demo03:app,port8000,reloadTrue,log_leveldebug)test_demo03 pip install faker 构造测试数据的一个库 import random import requests from faker import Faker def test_001(): f Faker(zh-CN) phone f.phone_number() pwd f.password() print(手机号,phone) print(密码,pwd) url http://127.0.0.1:8000/user/register data {phone: phone, pwd: pwd} r requests.post(url, jsondata) print(r.text) assert r.status_code 200 assert r.json()[code] 10001 assert r.json()[msg] 注册成功 url http://127.0.0.1:8000/user/login r requests.post(url, jsondata) print(r.text) token r.json()[token] print(token) assert r.status_code 200 assert r.json()[code] 10001 assert r.json()[msg] 登录成功 url http://127.0.0.1:8000/purchase_add data { id: str(random.randint(1000, 9999)) * 4, date: 2026-08-21, product: 办公打印机, num: 2, totalAmount: 5600.00, status: 待审核 } headers {token: token} r requests.post(url, jsondata, headersheaders) print(r.text) assert r.status_code 200 assert r.json()[code] 10001 assert r.json()[msg] 添加成功demo01_调用大模型 pip install langchain langchain-openai python-dotenv langchain 用来开发基于大模型的应用程序 api-key 1. 可以写死代码中 2. 写到env文件中-我们使用这种 3. 写到环境变量 import os import dotenv from langchain.chat_models import init_chat_model from langchain_community.chat_models.tongyi import ChatTongyi from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url os.getenv(BASE_URL) api_key os.getenv(API_KEY) model_name os.getenv(MODEL_NAME) # 方式1创建一个大模型的实例任何大模型都可以使用这种方式。 model init_chat_model( model model_name, api_key api_key, base_url base_url, model_provider openai, temperature 0.9, # 数字越高越有创意 max_tokens 300, # 回答的最大的token数量 max_retries 3 # 最大重试的次数 ) # 调用大模型 r model.invoke(你好你是谁) print(r) print(type(r)) # class langchain_core.messages.ai.AIMessage print(r.content) # 方式2针对OpenAI的模型 model ChatOpenAI( model model_name, api_key api_key, base_url base_url ) r model.invoke(你好你是谁) print(r.content) # 方式3针对通义千问、deepseek等 model ChatTongyi(model model_name,base_url base_url,dashscope_api_key api_key) r model.invoke(你好你是谁) print(r.content)demo02_消息的几种方式 1. 字符串 适用于不用保留历史会话的场景。 2. AIMessage、HumanMessage、SystemMessage定义消息列表 使用于保留历史会话的场景。 3. 字典列表 [{role: , content: },{role: , content: }] import os import dotenv from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url os.getenv(BASE_URL) api_key os.getenv(API_KEY) model_name os.getenv(MODEL_NAME,qwen3.7-plus) # 方式1创建一个大模型的实例任何大模型都可以使用这种方式。 model ChatOpenAI( model model_name, api_key api_key, base_url base_url, ) # 1. 字符串 # r model.invoke(你好你是谁) # print(r.content) # 2. 消息列表 msgs [ SystemMessage(content你是一个人工智能的助手擅长帮用户解决数字问题你的名字叫小A), AIMessage(content好的我叫小A), HumanMessage(content你是谁) ] r model.invoke(msgs) print(r.content) # 小A # 3. 字典列表 msgs [ {role: system,content:你是一个人工智能的助手擅长帮用户解决数字问题你的名字叫小A}, {role: ai,content: 好的我叫小A}, {role: human,content: 你是谁}, ] r model.invoke(msgs) print(r.content) # 小A msgs [ (system,你是一个人工智能的助手擅长帮用户解决数字问题你的名字叫小A), (ai,好的我叫小A), (human,你是谁), ] r model.invoke(msgs) print(r.content) # 小Ademo03_调用大模型的几种方式 1. invoke 非流式输出大模型一次性把回答返回 2. stream 流式输出大模型逐字逐句的返回 3. batch 批量调用 4. ainvoke 异步调用 import os import dotenv from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url os.getenv(BASE_URL) api_key os.getenv(API_KEY) model_name os.getenv(MODEL_NAME,qwen3.7-plus) model ChatOpenAI( model model_name, api_key api_key, base_url base_url, ) ## 2.stream流式输出。大模型逐字逐句的返回 r model.stream(给我一首李白的诗) print(type(r)) # class generator 生成器 # 列表推导式 [i for i in range(10)] 生成器i for i in range(10) # 普通函数加yield生成器 for item in r: print(,item.content) 智能体测开Day59
demo03 采购管理 采购单号 全局唯一 采购日期 商品名称 商品数量 采购总金额 状态待审核、已下单、已入库、已取消 增 post json格式的参数 /purchase_add 删 post json格式的参数 /purchase_delete 改 post json格式的参数 /purchase_modify 全量查 get /purchase_list 根据采购单号查询 get 依赖注入 采购增删改查的功能血药登录后才能使用每个接口需要校验用户是否登录。 将是否登录提取出来作为一个公共的函数其他接口中使用 采购增删改查的功能需要连接数据库使用完成后断开连接 将数据库连接提取出来作为一个公共的函数其他的 import uuid import pymysql import uvicorn from fastapi import FastAPI, Form, Header from fastapi.params import Depends from langgraph_sdk.auth.exceptions import HTTPException from pydantic import BaseModel, Field from starlette.responses import HTMLResponse app FastAPI() # 创建一个实例 class PurchaseMode(BaseModel): id: str Field(...,min_length16,max_length16,description采购单号)# ... 必填项 date: str product: str Field(...,min_length2,max_length16,description商品名称)# ... 必填项 num: int Field(gt0,description数量) totalAmount: float status: str class DeleteMode(BaseModel): id: str Field(...,min_length16,max_length16,description采购单号) DB { host: 127.0.0.1, port: 3306, user: root, password: 123456, charset: utf8 } class UserModel(BaseModel): phone:str Field(...,min_length11,max_length11,description手机号码) pwd:str Field(...,min_length6,max_length18,description密码) def init_db(): try: db pymysql.connect(**DB) # 解包 keyvalue,keyvalue # db pymysql.connect(host127.0.0.1,port 3306,user root,password 123456,) cursor db.cursor() cursor.execute(fcreate database if not exists fastapi38 character set utf8;) db.commit() db.select_db(fastapi38) cursor.execute(fcreate table if not exists purchase (id varchar(16) primary key, date date, fproduct varchar(16) not null, num int, totalAmount float, status varchar(16));) cursor.execute(create table if not exists user(id int primary key auto_increment, phone varchar(11), pwd varchar(20), token varchar(32));) db.commit() cursor.close() db.close() print(初始化数据库完成) except Exception as e: print(f初始化数据库异常{e}) #####依赖1数据库连接 def get_db(): db pymysql.connect(**DB) db.select_db(fastapi38) yield db db.close() ##### 依赖2是否登录 def check_login(token: str Header(None)): if token is None: return {msg: 请先登录,code: 20101} db pymysql.connect(**DB) db.select_db(fastapi38) cursor db.cursor() sql select * from user where token %s cursor.execute(sql,(token,)) a cursor.fetchone() if a is not None: return True else: raise HTTPException(status_code401,detail登录状态已经失效) app.post(/user/register) def register(data:UserModel, db: pymysql.Connection Depends(get_db)): cursor db.cursor() sql select * from user where phone %s cursor.execute(sql, (data.phone,)) a cursor.fetchone() if a is not None: return {msg: 手机号已被注册, code: 20101} else: sql insert into user (phone, pwd) values (%s, %s) cursor.execute(sql, (data.phone, data.pwd)) db.commit() return {msg: 注册成功, code: 10001} app.post(/user/login) def login(data:UserModel, db: pymysql.Connection Depends(get_db)): cursor db.cursor() sql select * from user where phone %s and pwd %s cursor.execute(sql, (data.phone,data.pwd,)) a cursor.fetchone() if a is not None: token str(uuid.uuid4()).replace(-,) # 生成一个token,表示用户登录状态 print(token,token) cursor.execute(update user set token %s where phone %s and pwd %s,(token,data.phone,data.pwd)) db.commit() return {msg: 登录成功, code: 10001,token: token} else: return {msg: 手机号或密码错误, code: 20101} app.post(/purchase_add,dependencies[Depends(check_login)]) def add(data: PurchaseMode,db: pymysql.Connection Depends(get_db)): cursor db.cursor() # 查询结果是元祖 sql select * from purchase where id %s cursor.execute(sql,(data.id,)) a cursor.fetchone() if a is not None: return {msg: 采购单号重复, code: 20101} else: sql insert into purchase (id, date, product, num, totalAmount, status) values (%s, %s, %s, %s, %s, %s) cursor.execute(sql,(data.id,data.date,data.product,data.num,data.totalAmount,data.status)) db.commit() cursor.close() return {msg: 添加成功, code: 10001} app.post(/purchase_delete, dependencies[Depends(check_login)]) def delete(data: DeleteMode, db: pymysql.Connection Depends(get_db)): cursor db.cursor() sql select * from purchase where id %s cursor.execute(sql, (data.id,)) a cursor.fetchone() if a is None: cursor.close() db.close() return {msg: 采购单号不存在, code: 20102} else: sql1 delete from purchase where id%s cursor.execute(sql1,(data.id,)) db.commit() cursor.close() return {msg: 删除成功, code: 10001} app.post(/purchase_modify, dependencies[Depends(check_login)]) def modify(data: PurchaseMode, db: pymysql.Connection Depends(get_db)): cursor db.cursor() cursor.execute(select * from purchase where id%s, (data.id,)) a cursor.fetchone() if a is None: cursor.close() db.close() return {msg:单据不存在无法修改,code:20103} else: sql update purchase set date%s, product%s, num%s, totalAmount%s, status%s where id%s cursor.execute(sql,(data.date,data.product,data.num,data.totalAmount,data.status,data.id)) db.commit() cursor.close() return {msg:修改成功,code:10001} app.get(/purchase_list, dependencies[Depends(check_login)]) def purchase_list(db: pymysql.Connection Depends(get_db)): cursor db.cursor(pymysql.cursors.DictCursor) sql select * from purchase cursor.execute(sql) a cursor.fetchall() cursor.close() return {msg: 查询成功, code: 10001, data: a} app.get(/purchase_get,dependencies[Depends(check_login)]) def get_one(id:str, db: pymysql.Connection Depends(get_db)): cursor db.cursor(pymysql.cursors.DictCursor) sql select * from purchase where id %s cursor.execute(sql, (id,)) a cursor.fetchone() cursor.close() if a is None: return {msg: 查询失败,无该采购单号, code: 20104} else: return {msg: 查询成功, code: 10001, data:a} # 页面路由读取static文件夹html文件 app.get(/register.html, response_classHTMLResponse) def page_register(): with open(static/register.html, r, encodingutf-8) as f: return f.read() app.get(/login.html, response_classHTMLResponse) def page_login(): with open(static/login.html, r, encodingutf-8) as f: return f.read() app.get(/purchase.html,response_classHTMLResponse) def purchase_page(): with open(static/purchase.html, r, encodingutf-8) as f: return f.read() if __name__ __main__: init_db() # 初始化数据库 uvicorn.run(demo03:app,port8000,reloadTrue,log_leveldebug)test_demo03 pip install faker 构造测试数据的一个库 import random import requests from faker import Faker def test_001(): f Faker(zh-CN) phone f.phone_number() pwd f.password() print(手机号,phone) print(密码,pwd) url http://127.0.0.1:8000/user/register data {phone: phone, pwd: pwd} r requests.post(url, jsondata) print(r.text) assert r.status_code 200 assert r.json()[code] 10001 assert r.json()[msg] 注册成功 url http://127.0.0.1:8000/user/login r requests.post(url, jsondata) print(r.text) token r.json()[token] print(token) assert r.status_code 200 assert r.json()[code] 10001 assert r.json()[msg] 登录成功 url http://127.0.0.1:8000/purchase_add data { id: str(random.randint(1000, 9999)) * 4, date: 2026-08-21, product: 办公打印机, num: 2, totalAmount: 5600.00, status: 待审核 } headers {token: token} r requests.post(url, jsondata, headersheaders) print(r.text) assert r.status_code 200 assert r.json()[code] 10001 assert r.json()[msg] 添加成功demo01_调用大模型 pip install langchain langchain-openai python-dotenv langchain 用来开发基于大模型的应用程序 api-key 1. 可以写死代码中 2. 写到env文件中-我们使用这种 3. 写到环境变量 import os import dotenv from langchain.chat_models import init_chat_model from langchain_community.chat_models.tongyi import ChatTongyi from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url os.getenv(BASE_URL) api_key os.getenv(API_KEY) model_name os.getenv(MODEL_NAME) # 方式1创建一个大模型的实例任何大模型都可以使用这种方式。 model init_chat_model( model model_name, api_key api_key, base_url base_url, model_provider openai, temperature 0.9, # 数字越高越有创意 max_tokens 300, # 回答的最大的token数量 max_retries 3 # 最大重试的次数 ) # 调用大模型 r model.invoke(你好你是谁) print(r) print(type(r)) # class langchain_core.messages.ai.AIMessage print(r.content) # 方式2针对OpenAI的模型 model ChatOpenAI( model model_name, api_key api_key, base_url base_url ) r model.invoke(你好你是谁) print(r.content) # 方式3针对通义千问、deepseek等 model ChatTongyi(model model_name,base_url base_url,dashscope_api_key api_key) r model.invoke(你好你是谁) print(r.content)demo02_消息的几种方式 1. 字符串 适用于不用保留历史会话的场景。 2. AIMessage、HumanMessage、SystemMessage定义消息列表 使用于保留历史会话的场景。 3. 字典列表 [{role: , content: },{role: , content: }] import os import dotenv from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url os.getenv(BASE_URL) api_key os.getenv(API_KEY) model_name os.getenv(MODEL_NAME,qwen3.7-plus) # 方式1创建一个大模型的实例任何大模型都可以使用这种方式。 model ChatOpenAI( model model_name, api_key api_key, base_url base_url, ) # 1. 字符串 # r model.invoke(你好你是谁) # print(r.content) # 2. 消息列表 msgs [ SystemMessage(content你是一个人工智能的助手擅长帮用户解决数字问题你的名字叫小A), AIMessage(content好的我叫小A), HumanMessage(content你是谁) ] r model.invoke(msgs) print(r.content) # 小A # 3. 字典列表 msgs [ {role: system,content:你是一个人工智能的助手擅长帮用户解决数字问题你的名字叫小A}, {role: ai,content: 好的我叫小A}, {role: human,content: 你是谁}, ] r model.invoke(msgs) print(r.content) # 小A msgs [ (system,你是一个人工智能的助手擅长帮用户解决数字问题你的名字叫小A), (ai,好的我叫小A), (human,你是谁), ] r model.invoke(msgs) print(r.content) # 小Ademo03_调用大模型的几种方式 1. invoke 非流式输出大模型一次性把回答返回 2. stream 流式输出大模型逐字逐句的返回 3. batch 批量调用 4. ainvoke 异步调用 import os import dotenv from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url os.getenv(BASE_URL) api_key os.getenv(API_KEY) model_name os.getenv(MODEL_NAME,qwen3.7-plus) model ChatOpenAI( model model_name, api_key api_key, base_url base_url, ) ## 2.stream流式输出。大模型逐字逐句的返回 r model.stream(给我一首李白的诗) print(type(r)) # class generator 生成器 # 列表推导式 [i for i in range(10)] 生成器i for i in range(10) # 普通函数加yield生成器 for item in r: print(,item.content)