FastAPI 快速开发 Web API 项目: 通过 SQLAlchemy 进行数据操作

举报
宇宙之一粟 发表于 2023/04/24 11:41:17 2023/04/24
【摘要】 FastAPI 快速开发 Web API 项目学习笔记：第一篇：通过 Python FastAPI 开发一个快速的 Web API 项目第二篇：FastAPI 的路由介绍与使用第三篇：FastAPI 开发中数据校验利器 Pydantic 介绍与集成使用第四篇：FastAPI 快速开发 Web API 项目: 路径参数和查询参数第五篇：从 Flask 转换到 FastAPI（翻译）第六篇：Fa...

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 表的数据库元数据：

  1. 映射从一个基类开始，上面称为 Base，它是通过针对 DeclarativeBase 类创建一个简单的子类来创建的。
  2. 然后通过创建 Base 的子类来创建单独的映射类。映射类通常指单个特定的数据库表，其名称通过使用 __tablename__ 类级属性来指示。
  3. 接着定义表的列字段：字段类型、是否自增、是否主键、是否为空等等
  4. 更多知识请查看 SQLAlchemy 的文档…

利用 SQLAlchemy 创建 User 的模型，在 model.py 中写入如下的代码：

from sqlalchemy import Column, TEXT, INT, BIGINT, DATETIME, INTEGER
from 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_engine
from 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 = 20
POOL_RECYCLE = 3600
POOL_TIMEOUT = 15
MAX_OVERFLOW = 2
CONNECT_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 APIRouter
from sqlalchemy import and_
from db.db import Database
from models.model import User
from 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 APIRouter
from endpoints import user

router = APIRouter()
router.include_router(user.router)

main.py 文件

此时，我们通过编写 main.py :

import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from fastapi import FastAPI
from 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 Users 快速将注册和身份验证系统添加到您的 FastAPI 项目。

  • Fief

Fief 是一个基于 FastAPI Users 框架的用户认证和授权系统。它是开源的，可以自己免费搭建，也可以使用他们提供的托管服务。用户可以选择自己的数据库，Fief 会负责与数据库的交互。此外，还提供了预先构建好的登录和注册页面，使得用户可以方便快捷地实现身份验证功能。最后，还提供了官方 Python 客户端，内置了 FastAPI 集成，可以帮助用户更加方便地使用。

总结

本文将 SQLAlchemy 结合使用 FastAPI 实现了数据库的连接和操作，更多 ORM 操作数据库大同小异，如果学习过 Django 框架的同学本身也不陌生，不懂就查官方文档。优秀的框架总少不了优秀的文档，也少不了大量的使用者。

希望本文能对你有所帮助，如果喜欢本文，可以点个关注。

下一篇文章见！宇宙古今无有穷期，一生不过须臾，当思奋争。

参考链接：

【版权声明】本文为华为云社区用户原创内容，转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息， 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
API
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入