FastAPI 快速开发 Web API 项目学习笔记：
前言
在 FastAPI 中管理数据库模型的一个有用选项是 SQLAlchemy。我将使用它的 ORM 来方便访问带有 Python 熟悉的对象的数据库。
SQLAlchemy 介绍
SQLAlchemy 是 Python SQL 工具包和对象关系映射器，它为应用程序开发人员提供了 SQL 的完整功能和灵活性。
它提供了一整套众所周知的企业级持久性模式，专为高效和高性能的数据库访问而设计，适用于简单的 且非常 Pythonic 语言。
安装 pip install sqlalchemy:
User 模型
创建数据库和表
在上一篇文章中，我们已经创建了一 example_db 的数据库，然后我们在这一篇文章中直接进行使用。
创建一个 user 表：
CREATE TABLE `user` ( `id` int unsigned NOT NULL AUTO_INCREMENT, `name` varchar(255) DEFAULT NULL, `age` int DEFAULT NULL, `created_by` varchar(255) DEFAULT NULL, `created_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (`id`) ) ENGINE = InnoDB AUTO_INCREMENT = 2 DEFAULT CHARSET = utf8mb3
复制代码
文章结构
如下所示创建文件夹和文件，使项目的文件结构如下所示：
创建 User 模型
在 SQLAlchemy 定义了模块级结构，这些结构将形成从数据库中查询的结构。这种称为声明式映射的结构同时定义了 Python 对象模型以及描述特定数据库中存在或将存在的真实 SQL 表的数据库元数据：
映射从一个基类开始，上面称为 Base，它是通过针对 DeclarativeBase 类创建一个简单的子类来创建的。
然后通过创建 Base 的子类来创建单独的映射类。映射类通常指单个特定的数据库表，其名称通过使用 __tablename__ 类级属性来指示。
接着定义表的列字段：字段类型、是否自增、是否主键、是否为空等等
更多知识请查看 SQLAlchemy 的文档...
利用 SQLAlchemy 创建 User 的模型，在 model.py 中写入如下的代码：
from sqlalchemy import Column, TEXT, INT, BIGINT, DATETIME, INTEGERfrom sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base): __tablename__ = 'user' id = Column(BIGINT, autoincrement=True, primary_key=True) name = Column(TEXT, nullable=False) age = Column(INT, nullable=False) created_by = Column(INTEGER, nullable=False) created_at = Column(DATETIME, nullable=False)
复制代码
连接 MySQL 数据库
Engine 引擎是一个工厂，可以为我们创建新的数据库连接，它也可以在连接池中保存连接，以便快速重复使用。
通过 Session = sessionmaker(bind=engine)创建 session 持久会话对象
基本的 SQLite 引擎创建结构如下：
>>> from sqlalchemy import create_engine>>> engine = create_engine("sqlite://", echo=True)
复制代码
在 db/db.py 中写入如下代码：
from sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmaker
# DB_URL = 'mysql+pymysql://{USERNAME}:{PASSWORD}@{PORT}/{DBNAME}'MYSQL_URL = 'mysql+pymysql://root:123456@localhost:3306/example_db?charset=utf8'POOL_SIZE = 20POOL_RECYCLE = 3600POOL_TIMEOUT = 15MAX_OVERFLOW = 2CONNECT_TIMEOUT = 60
class Database(): def __init__(self) -> None: self.connection_is_active = False self.engine = None def get_db_connection(self): if self.connection_is_active == False: connect_args = {"connect_timeout": CONNECT_TIMEOUT} try: self.engine = create_engine(MYSQL_URL, pool_size=POOL_SIZE, pool_recycle=POOL_RECYCLE, pool_timeout=POOL_TIMEOUT, max_overflow=MAX_OVERFLOW, connect_args=connect_args) return self.engine except Exception as e: print("Error connecting to MySQL DB:", e) return self.engine def get_db_session(self, engine): try: Session = sessionmaker(bind=engine) session = Session() return session except Exception as e: print("Error getting DB session:", e) return None
复制代码
创建 User API 终端
在 endpoints/user.py 中写入如下的代码：
from fastapi import APIRouterfrom sqlalchemy import and_from db.db import Databasefrom models.model import Userfrom models.response import Response
router = APIRouter( prefix="/users", tags=["User"], responses={404: {"description": "404 Not Found"}},)
database = Database()engine = database.get_db_connection()
@router.get("/")async def read_all_users(): session = database.get_db_session(engine) data = session.query(User).all() return Response(data, 200, "All User retrieved successfully.", False)
@router.get('/{user_id}')async def get_user(user_id: int): session = database.get_db_session(engine) response_message = "User retrieved successfully" data = None try: data = session.query(User).filter( and_(User.id == user_id)).one() except Exception as ex: print("Error:", ex) response_message = "User Not found" error = False return Response(data, 200, response_message, error)
复制代码
创建路由
把刚刚的 User API 终端导入到 FastAPI 的路由中，因此需要在 routers.api.py 文件中，写入如下代码：
from fastapi import APIRouterfrom endpoints import user
router = APIRouter()router.include_router(user.router)
复制代码
main.py 文件
此时，我们通过编写 main.py :
import uvicornfrom fastapi.middleware.cors import CORSMiddlewarefrom fastapi import FastAPIfrom routers.api import router as api_router
app = FastAPI()
origins = ["http://localhost:8000"]
app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"],)
app.include_router(api_router)
if __name__ == '__main__': uvicorn.run("main:app", host='127.0.0.1', port=8000, log_level="info", reload = True) print("running")
复制代码
运行 main.py 文件：
访问 http://127.0.0.1:8000/redoc 会自动帮我们重定向到 User 的接口文档界面：
访问 http://127.0.0.1:8000/users/:
{ "data":[ { "name":"Atom", "id":1, "created_by":"Admin", "created_at":"2023-04-20T16:17:30", "age":18 } ], "code":200, "message":"All User retrieved successfully.", "error":false}
复制代码
访问 http://127.0.0.1:8000/users/1：
{ "data":{ "name":"Atom", "id":1, "created_by":"Admin", "created_at":"2023-04-20T16:17:30", "age":18 }, "code":200, "message":"User retrieved successfully", "error":false}
复制代码
至此，我们实现了与上一篇文章中相同的效果。感兴趣的朋友，可以完善其他 CRUD 操作。感觉把知识结合起来学习效果更佳。
第三方库学习
在思考 User 的注册和登录时，发现了下面两个框架，先记录下来，今后进行学习。
FastAPI Users 快速将注册和身份验证系统添加到您的 FastAPI 项目。
Fief 是一个基于 FastAPI Users 框架的用户认证和授权系统。它是开源的，可以自己免费搭建，也可以使用他们提供的托管服务。用户可以选择自己的数据库，Fief 会负责与数据库的交互。此外，还提供了预先构建好的登录和注册页面，使得用户可以方便快捷地实现身份验证功能。最后，还提供了官方 Python 客户端，内置了 FastAPI 集成，可以帮助用户更加方便地使用。
总结
本文将 SQLAlchemy 结合使用 FastAPI 实现了数据库的连接和操作，更多 ORM 操作数据库大同小异，如果学习过 Django 框架的同学本身也不陌生，不懂就查官方文档。优秀的框架总少不了优秀的文档，也少不了大量的使用者。
希望本文能对你有所帮助，如果喜欢本文，可以点个关注。
下一篇文章见！宇宙古今无有穷期，一生不过须臾，当思奋争。
参考链接：
评论