★ FEATURED ARTICLE FastAPi入门web服务启功1 ——直接点击运行符号不用重启就能更新代码2 -——输入 uvicorn main:app --reloadpythonweb的热启动不用重启就能更新代码交互文档路径后加/docs路由Java里的controller路径请求方法路径参数from fastapi import FastAPI,Path再形参后加Path查询参数导入from fastapi import FastAPI,QueryQUery请求体先导包from pydantic import BaseModelField一参见实体类类注解导入FieldFiled二定义请求体class User(BaseModel): usernamr:str password:str app.post(/user) async def create_user(user:User): return user响应格式导入这里导入的为html格式from starlette.responses import HTMLResponse创建接口app.get(/html,response_classHTMLResponse) async def html(): return h1 这是个html格式/h1自定义响应格式要先基于BaseModel创建实体类然后创建接口异常处理先导包from fastapi import HTTPException然后定义接口用raise HTTPException()抛出问题中间件类似拦截器先进的最后执行代码 后进的的 先执行代码依赖注入一导入Dependsfrom fastapi import FastAPI,Depends2抽取充分的异步函数3依赖注入形参名Depends抽取充分的异步函数 orm建表1创建异步引擎2定义数据类型 基类提取共用的属性如创建时间和更改时间和对应表的数据模型3定义函数来调用异步引擎引入的包 from datetime import datetime from fastapi import FastAPI from sqlalchemy import func, DateTime,String from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column #,Mapped为声明类型 mapped_column为映射字段#创建异步引擎 FastAPI_urlmysqlaiomysql://root:123456localhost:3306/fastapi_first engine create_async_engine(FastAPI_url, echoTrue) #2定义数据类型 创建基类表对应的数据模型 class Datatime: pass class Base(DeclarativeBase): create_time :Mapped[datetime] mapped_column(DateTime,insert_defaultfunc.now,defaultfunc.now,comment创建时间) update_time : Mapped[datetime] mapped_column(DateTime,insert_defaultfunc.now,defaultfunc.now,comment更新时间) # 定义数据模型 书籍id class book(Base): __tablename__ book # 表名 id :Mapped[int] mapped_column( primary_keyTrue,commentid) name :Mapped[str] mapped_column(String(255),comment书名) author :Mapped[str] mapped_column(String(50),comment作者) #3定义函数建表——》fastAP调用函数建表 async def create_table(): # 获取异步请求引擎 async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) app.on_event(startup) async def startup(): await create_table()orm的使用# 定义会话工厂 sessionmaker async_sessionmaker(bindengine, class_AsyncSession, expire_on_commitFalse ) # 依赖项 async def get_session(): async with sessionmaker() as session: try: yield session await session.commit() except: await session.rollback() raise finally: await session.close() app.get(/book) async def get_book(sessionDepends(get_session)): book_list await session.execute(select(book)) return book_list.scalars().all()这段代码实现FastAPI 异步 SQLAlchemy 数据库操作包含会话工厂、依赖注入、事务管理、查询数据二、逐段知识点讲解1️⃣ async_sessionmaker异步会话工厂python运行sessionmaker async_sessionmaker( bindengine, # 绑定数据库引擎 class_AsyncSession, # 使用异步会话 expire_on_commitFalse # 提交后不过期 )知识点sessionmaker不是会话本身是制造会话的工厂AsyncSessionSQLAlchemy 异步会话必须配合await使用expire_on_commitFalse提交后查询到的对象不会失效可以继续使用不会报错作用统一创建数据库会话避免重复写代码2️⃣ 依赖项 get_session () —— 数据库连接管理器python运行async def get_session(): async with sessionmaker() as session: try: yield session await session.commit() except: await session.rollback() raise finally: await session.close()核心知识点async with异步上下文管理器自动创建 / 关闭会话yield session生成器依赖先给接口提供会话请求结束后再执行后续代码commit/rollback事务机制commit()成功就提交rollback()出错就回滚保证数据安全finally无论成功失败一定关闭会话防止数据库连接泄漏3️⃣ Depends 依赖注入python运行async def get_book(sessionDepends(get_session)):知识点DependsFastAPI 依赖注入作用自动调用get_session()自动创建会话自动提交 / 回滚自动关闭让接口只关心业务不关心数据库连接4️⃣ 异步数据库查询python运行book_list await session.execute(select(book)) return book_list.scalars().all()知识点await所有异步数据库操作必须加session.execute()执行 SQL 语句的标准方法select(book)SQLAlchemy 2.0 新式查询语法等价于SELECT * FROM bookscalars().all()scalars()只取模型对象去掉元组包装all()返回所有结果列表结果[Book对象, Book对象...orm查询app.get(/book) async def get_book(db: AsyncSession Depends(get_session)): # book_list await db.execute(select(book))# 查询——》返回一个orm对象 # resultbook_list.scalars().all()# 获取所有数据 # result book_list.scalars().first() # 获取一条数据 books await db.get(book, 2) return booksorm条件查询模糊查询示例代码 app.get(/book/{id}) async def getbookByid(id:int,db:AsyncSession Depends(get_session)): resultawait db.execute(select(book).where(book.idid)) books result.scalar_one_or_none() return books 1. app.get(/book/{id}) 作用定义一个 GET 接口 /book/{id}动态路由 {id}表示可以传入任意数字比如 /book/2、/book/3 效果用户访问 /book/2 就会执行下面的函数 2. async def getbookByid(id:int, db:AsyncSession Depends(get_session)): async def异步函数FastAPI 必须这么写 getbookByid函数名根据 ID 获取书籍 id:int接收前端传来的 id 参数并限定必须是数字 db:AsyncSession Depends(get_session) 从依赖项获取数据库会话 db 就是你和数据库通信的 “通道” 3. result await db.execute(select(book).where(book.id id)) 这一行是核心查询语句 await必须加 异步操作等待执行 db.execute(...)执行 SQL 语句 select(book)等价于 SELECT * FROM book .where(book.id id)条件查询 意思只查 book 表里 id 传入参数的那条数据 整体等价于 SQL sql SELECT * FROM book WHERE id ? 4. books result.scalar_one_or_none() result是数据库返回的原生结果集 .scalar_one_or_none() 取出一条结果 有数据 → 返回对象 没数据 → 返回 None 最安全、最常用的取值方法 5. return books 把查询到的书籍数据返回给前端 如果有数据 → 返回 JSON 没有 → 返回 null聚合查询分页查询示例 # 1. 定义 GET 接口地址/book/select app.get(/book/select) async def select_book( page1, # 默认第 1 页 page_size2, # 默认每页 2 条数据 db: AsyncSession Depends(get_session) # 数据库会话 ): # 计算从第几条开始查 # 第1页(1-1)*2 0 → 从第0条开始 # 第2页(2-1)*2 2 → 从第2条开始 skil (page - 1) * page_size # 构建查询语句分页查询 # select(book) → 查询 book 表 # offset(skil) → 跳过 skil 条数据 # limit(page_size) → 只取 page_size 条 stment select(book).offset(skil).limit(page_size) # 2. 执行数据库查询 result await db.execute(stment) # 3. 获取所有查询到的书籍列表 # scalars() → 提取 ORM 对象 # all() → 转成列表 books result.scalars().all() # 4. 返回数据给前端 return {book: books}新增book在数据库层面是表bookBase你可以把它理解为表单校验class bookBase(BaseModel): id:int name:str author:str app.post(/book/add) async def add_booK(book_data: bookBase ,db:AsyncSession Depends(get_session)): book_obj book(**book_data.__dict__)# 将数据模型转换为orm对象 db.add(book_obj)# 添加数据 await db.commit()# 提交会话 return book_data更新查找——》重新赋值app.put(/book/update/{id}) async def update_book(id:int,book_data:bookBase,db:AsyncSession Depends(get_session)): #查询数据 books await db.get(book,id) if books is None: return {message: 数据不存在} #更新数据 books.namebook_data.name books.authorbook_data.author await db.commit() return books删除数据查找——》删除app.delete(/book/delete/{id}) async def delete_book(id:int,db:AsyncSession Depends(get_session)): #查询数据 books await db.get(book,id) if books is None: return {message: 数据不存在} #删除数据 await db.delete(books) await db.commit() return {message: 删除成功}redis引用封装redis方法新闻项目实战用户接口用户注册接口实现流程问题添加insert 和add的区别方法 insert 是对 sql的insert方法的封装接近sql语句属于orm的低级语句add是orm的高级语句需要有模型类add是 ORM 的高层接口用于将一个模型对象添加到当前的会话中进行管理。在 SQLAlchemy 中它与Session绑定将模型实例置于会话的“脏”状态允许后续的事务提交来实际完成插入操作。特点需要通过 ORM 的模型类创建一个对象实例并将实例与会话绑定。隐式地处理对象间关系例如外键约束和其他高级机制。暂时将对象保存到会话中只有当commit或flush时插入操作才会真正被发送到数据库。支持更丰富的功能比如对象状态跟踪transient、pending、persistent等。适用场景需要利用 ORM 提供的高层功能例如关系自动管理、级联保存等。通常用于需要动态创建和插入单个或少量对象的情况。SQLAlchemy 示例# 假设 User 是一个 ORM 模型 user User(nameAlice, age30) # 创建一个新的用户对象 session.add(user) # 将用户添加到会话 session.commit() # 提交事务将对象数据插入数据库总结对比特性insertadd操作级别偏底层接近原生 SQL高级基于 ORM 的对象管理插入方式批量或直接插入 SQL 语句创建模型对象后添加到 ORM 会话关系处理不处理自动处理对象关系性能性能较高单纯插入性能稍慢带有对象跟踪和关系管理应用场景大规模插入或对简单数据的直接操控对象化管理、动态插入及 ORM 全功能支持问题pyanctic作用解决做请求体响应体数据校验和类型转化的模型类一、核心作用专门用来接口请求体校验、响应体格式化、自动类型转换、字段规则约束FastAPI 标配。二、请求体入参模型继承BaseModel定义字段自动接收并校验前端传参。python运行from pydantic import BaseModel class UserReq(BaseModel): name: str age: int email: str | None None三、响应体出参模型同样用BaseModel规范返回给前端的字段、过滤多余字段、统一格式。python运行class UserResp(BaseModel): id: int name: str # 只返回定义的字段自动过滤多余属性四、数据转换自动类型强转字符串数字自动转int/float时间字符串自动转datetime字典自动解析成嵌套模型模型可一键转字典 / JSONmodel.model_dump()→ 字典model.model_dump_json()→ JSON 字符串五、数据约束字段限制用Field做规则约束数值gt大于、ge大于等于、lt小于、le小于等于字符串min_length/max_length长度限制、regex正则格式校验EmailStr邮箱、HttpUrl网址固定枚举Literal[男,女]只能选指定值六、自定义约束装饰器field_validator手写校验规则不符合直接抛异常。七、额外配置model_config {extra: forbid}禁止传多余未知字段防脏数据。问题什么时间用await一、必须用 await 的 4 种场景调用 async 定义的函数python运行async def func(): ... await func() # 必须加操作数据库异步驱动SQLAlchemy asyncDatabasesMotor (MongoDB)python运行await db.execute(...) await db.fetch_all(...)发送 HTTP 请求异步客户端httpx.AsyncClientaiohttppython运行await client.get(https://xxx)异步 I/O 操作文件、Redis、消息队列等异步库python运行await redis.get(key)问题redis 使用1下载redis2连接redis同步普通连接python运行import redis r redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue # 自动字节转字符串不用手动encode/decode )异步连接FastAPI 常用bash运行pip install redis[asyncio]python运行from redis.asyncio import Redis redis_client Redis(hostlocalhost, port6379, db0, decode_responsesTrue)3封装方法调用方法前面链接起的变量名 redis_client,方法名1. 字符串 String最简单存普通文本、验证码、token存set设值setex顺便设过期时间取get删delete查有没有exists2. 哈希 Hash专门存对象适合存用户、商品信息不用拆成多个 key单个字段存 / 取hset/hget一次性拿全部信息hgetall删除某个字段hdel3. 列表 List排队顺序用有序、可重复适合消息队列、记录流水左右加数据lpush、rpush左右弹出拿走数据lpop、rpop查全部lrange 0 -14. 集合 Set自动去重无序、自动去重适合签到、好友列表、黑名单添加sadd查所有smembers删元素srem判断在不在集合里sismember5. 有序集合 ZSet做排行榜带分数自动排序适合榜单、积分排名加数据带分数zadd按排名查zrange删元素zrem问题pydantic 模型类调用BaseModelorm基本类调用 基类存放一些通用字段和模型类DeclarativeBase问题scalars作用scalars会给结果剥壳scalars()只拿查询的第一列字段自动把外层列表、元组拆掉直接拿到实体对象 / 单个值一、基础作用查出结果只取第一个字段自动剥离(对象,)这种元组包裹配合 ORM 查询直接得到模型对象列表不用自己循环拆元组代码更简洁二、常用链式方法 作用1.result.scalars().all()作用查出全部数据返回列表返回[模型对象1, 模型对象2, ...]场景查多条所有记录2.result.scalars().first()作用只拿第一条数据没有返回None场景查第一条、最新一条3.result.scalars().one()作用必须有且只能有一条多一条 / 少一条 直接报错场景确定唯一数据比如按主键查4.result.scalars().one_or_none()作用有一条就返回没有返回None多条报错场景不确定有没有防止报错5.result.scalars().limit(n)作用限制只取前 n 条6.result.scalars().unique()作用去重配合分组、联表查询去重7. 直接遍历for obj in result.scalars()作用直接循环每一个模型对象不用拆包三、对比有无 scalars () 区别不用 scalars ()python运行res db.execute(select(User)) all_data res.all() # 结果[(user1,), (user2,)] 带元组还要拆用 scalars ()python运行res db.execute(select(User)).scalars().all() # 结果[user1, user2] 直接是对象列表四、一句话口诀scalars()剥元组、只取首列、直拿对象.all()拿全部.first()拿第一条.one()必须唯一一条.one_or_none()有就拿没有为空.unique()去重问题orm应用二、配置数据库连接 URL不同数据库格式不同填你的账号密码python运行# 同步URL # MySQL: mysqlpymysql://账号:密码地址:端口/库名 DATABASE_URL mysqlpymysql://root:123456localhost:3306/test_db # SQLite: sqlite:///本地文件路径 # DATABASE_URL sqlite:///./test.db # 异步URL把驱动换成async版 ASYNC_DATABASE_URL mysqlaiomysql://root:123456localhost:3306/test_db三、创建数据库引擎连接数据库的核心入口python运行# 同步引擎 from sqlalchemy import create_engine # echoTrue 会打印执行的SQL调试用生产关 engine create_engine(DATABASE_URL, echoTrue) # 异步引擎 from sqlalchemy.ext.asyncio import create_async_engine async_engine create_async_engine(ASYNC_DATABASE_URL, echoTrue)四、创建 ORM 基类所有模型的父类继承它才能被 ORM 识别python运行from sqlalchemy.orm import declarative_base Base declarative_base()五、定义 ORM 模型表结构类 表属性 字段对象 行python运行from sqlalchemy import Column, Integer, String # 这就是你的表 class User(Base): __tablename__ users # 数据库里的表名 # 字段定义 id Column(Integer, primary_keyTrue, indexTrue) # 主键 name Column(String(50), indexTrue) # 字符串最大50 age Column(Integer)六、自动建表把你定义的模型同步到数据库生成真实的表python运行# 同步版直接跑 Base.metadata.create_all(bindengine) # 异步版要在异步函数里跑 async def init_db(): async with async_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)七、创建会话操作数据库的句柄有了会话才能真正增删改查python运行# 同步会话 from sqlalchemy.orm import sessionmaker SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) # FastAPI 常用依赖写法直接用 def get_db(): db SessionLocal() try: yield db # 把会话给接口用 finally: db.close() # 用完自动关 # 异步会话 from sqlalchemy.ext.asyncio import async_sessionmaker AsyncSessionLocal async_sessionmaker(autocommitFalse, autoflushFalse, bindasync_engine) # FastAPI 异步依赖 async def get_db(): async with AsyncSessionLocal() as session: yield session
阅读完成 · 觉得有帮助？
