phpyieldsqlite技巧_开拓者必看用FastAPI SQLite构建API的极速秘籍

文章目录 [+]

作为一个开拓者，如果想要快速的开拓 API，可能会有很多的选择，不同的措辞都有自己最为大略的框架。
他们都有自己的优缺陷，有的企业级受欢迎，有的发展历史悠久各种开拓问题都有案例，有的原生高并发，有的原生内存安全等等。

但是如果要快速开拓 API 并且有一定的性能和开拓体验，考虑到生态对接，这必须是 Python 的 FastAPI + SQLite。

phpyieldsqlite技巧_开拓者必看用FastAPI SQLite构建API的极速秘籍

为什么是 FastAPI?

（图片来自网络侵删）

Python 足够大略：措辞层面没有心智包袱。
想要学习编程措辞都可以学的会。
FastAPI 性能刁悍：基于 Starlette 和 Pydantic，这两个库都非常快。
性能直接比拟 Node.js 和 Golang。
适用于处理大量情形的运用，我们自己的小运用就更加不用说了。
FastAPI 自带文档：集成了 swagger 和 reDoc。
可以轻松的在路由中配置和查看并且测试 API。
FastAPI 类型系统：FastAPI 利用了 typings 也便是 Type Hints 最类型检讨，减少开拓过程中不必要的缺点。
FastAPI 异步支持：FastAPI 原生支持了异步编程，使得它非常适宜 I/O 密集型任务。
并且支持并发。
处理数据库要乞降外部 API 调用。
FastAPI 代码简洁当代：Python 本身语法极为简洁。
FastAPI 利用装饰器处理路由，并且装饰器能够直接装饰函数。
使得函数式编程变成非常方便。
FastAPI依赖注入：内部供应了依赖注入 Depend, 这位函数的扩展体统了良好的扩展。
FastAPI 社区生态：FastAPI 拥抱社区对各种库支持来那个号，并且内置了 OAuth2.o 和 JWT 方便的实现授权。
其他数据干系的库 SQLAlchemy 等也能非常方便的集成到。
Python 内置SQLite 支持。
安装依赖

我们默认你已经有了 Python 的开拓环境。
直接安装 FastAPI 及干系依赖。
（当然你也可利用其他的 Python 环境安装）

pip install fastapi uvicorn sqlalchemypip install databases # 异步支持数据库（可选安装）pip install sqlite # （可选安装）代码组织

一个良好的代码组织是管理好代码的关键，这里利用 python 模块的办法进行管理。

/app//main.py/db/ /__init__.py/db.py/models.py

实在也非常大略，将 db 与 sqlalchemy 模型分开。
以模块的形式供应给 fastapi 的路由利用。

定义 db 根本

from sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker# 定义数据库的连接字符串，这里利用的是 SQLite 数据库DATABASE_URL = "sqlite:///./test.db"# 创建一个数据库引擎 (Engine) 实例，用于与数据库通信engine = create_engine(DATABASE_URL)# 创建一个会话 (Session) 类，用于管理数据库连接# 这里 `autocommit=False` 表示会话须要手动提交# `autoflush=False` 表示在每个查询之前不会自动刷新会话SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 创建一个根本类，用于声明数据库模型 (ORM 模型)Base = declarative_base()

db 创建引擎和 session，同时利用对 ORM 供应 Base 数据模型。

定义模型和 Schema

本次定义 User 模型和更新模型，要知道他们的根本模型是不一样的。

from sqlalchemy import Column, Integer, Stringfrom db.db import Basefrom pydantic import BaseModel# 用户模型class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) name = Column(String, index=True) email = Column(String, unique=True, index=True)# 用户更新 Schemaclass UserUpdate(BaseModel): name: str | None = None email: str | None = None

模型基于 Class, Base class 是模型根本模型，每一个模型都须要继续它。
和更新 User 事后的 Schema。
如果你是初学 models 和 schema 不要搞混了，由于项目示例，放在一起。

供应 app 和 CRUD 的 db

from fastapi import FastAPI, Depends, HTTPExceptionfrom sqlalchemy.orm import Sessionfrom db.db import Base, SessionLocal, enginefrom db.models import User, UserUpdateBase.metadata.create_all(bind=engine)app = FastAPI()def get_db(): db = SessionLocal() try: yield db finally: db.close()

我们须要在启动 app 之前，导入依赖，同时要初利用 Base 初始化数据库表。
同时利用 Session 获取到 db，后面我们就能利用 db 操作数据库了。
此处没有利用异步，这里利用 yield 语句获取。

须要把稳的是：在启动做事自后，会自动创建 SQLite 数据库文件和对应的表。

基于 sqlalchemy ORM 的 CRUD

得益于 sqlalchemy 库支持 sql 和 ORM 两种模式，我们可以利用 ORM 方便链式调用操作数据库。

# 测试接口@app.get("/")def home(): return { "node": "123"}# 根据 user_id 获取参数@app.get("/users/{user_id}")def get_user_all(user_id: int, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if user is None: raise HTTPException(status_code=404, detail="User not found") return user# 获取所有的用户@app.get("/users/all")def get_user_all(db: Session = Depends(get_db)): users = db.query(User).all() return users# 分页获取数据@app.get("/users/list")def get_user_page(limit: int = 10, skip: int = 0, db: Session = Depends(get_db)): users = db.query(User).limit(limit).offset(skip).all() return users# 创建用户@app.post("/users/")def create_user(name: str, email: str, db: Session = Depends(get_db)): user = User(name=name, email=email) db.add(user) db.commit() db.refresh(user) return user# 更新用户@app.put("/users/{user_id}")def create_user(user_id: int, user_update: UserUpdate, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if user is None: raise HTTPException(status_code=404, detail="User not found") if user_update.name is not None: user.name = user_update.name if user_update.email is not None: user.email = user_update.email db.commit() db.refresh(user) return user# 删除用户@app.delete("/users/{user_id}")def delete_user(user_id: int, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if user is None: raise HTTPException(status_code=404, detail="User not found") db.delete(user) db.commit() return {"message": "User deleted successfully"}

crud 中 db 通过依赖注入的办法获取。
fastapi 供应 Depnds 注入 get_db 函数。

db.query(Model)：常用 filter()、filter_by()、order_by()、limit()、offset()db.add(instance)：用于将一个新的实例（常日是模型工具）添加到当前的数据库会话中。
db.commit()：用于提交当前数据库会话中的所有变动，将它们持久化到数据库。
db.rollback()：当操作失落败或须要撤销未提交的变动时，可以利用 rollback() 来回滚当前事务中的所有变动。
db.delete(instance)：从数据库中删除指定的实例。
db.refresh(instance)：重新加载数据库中的给定实例，确保它包含最新的数据。
db.close()：关闭当前的数据库会话。
常日不须要手动调用，由于 FastAPI 的依赖注入机制会自动处理会话的创建和关闭。
db.flush()：将当前会话中的变动发送到数据库，但不提交这些变动。
flush() 会将天生的 SQL 语句发送到数据库，确保可以在同一事务中引用新天生的主键。
启动做事

uvicorn main:app --reload

启动做事，可以查看数据库表是否创建，然后 fastapi 供应 docs 接口，进行接口测试。

http://127.0.0.1:8000/docs

你可以利用多种工具访问 SQLite 数据库：vscode 插件， SQLite Studio 等等

小结

本文旨在快速适用 fastapi + sqlite 完成 api 的 CRUD，同时就被 API 文档和测试环境.如果你还不熟习 fastapi 这篇文章一定会帮助到你，更加快速的实战。
当然，数据库其他操作如何迁移等、支配不是这篇文章的关注中央，其他的内容在后续文章中连续，感谢关注。

#如何做一个api接口# #Python# #API#