Python 使用 FastAPI 和 PostgreSQL 构建简单 API
最近一年公司也在卷 LLM 的应用项目，所以我们也从 goper => Pythoner。 这一年使用最多的就是 Python 的 FastAPI 框架。下面一个简易项目让你快速玩转 Python API Web。
API代表应用程序编程接口，是软件开发中最重要的概念之一。它允许程序通过发送和接收数据与其他服务进行交互。API Web 通信最广泛使用的标准之一是 REST，它依赖于JSON 格式或键值对，类似于 Python 的字典。
如果想用 Python 构建一个，那么可以从几个框架中选择。Flask -RESTful、Django Rest Framework 和 FastAPI 是最受欢迎的。
然而，FastAPI 专为快速 API 应用程序设计。
- 自动文档：FastAPI 根据 OpenAPI 标准自动生成全面的文档，节省时间和精力。
- 快捷、简便：专为高性能、高效执行而设计。
- 数据验证： FastAPI建立在 Pydantic 之上，提供了一批有用的功能，如数据验证和异步支持。
虽然Django Rest Framework在许多公司中也很常用，但让从FastAPI开始，了解它的好处。话虽如此，在简短的理论介绍之后，让我们继续编写 FastAPI 代码的第一步。
总体简介
端点
如前所述，API是不同服务和应用程序之间的一种通信方式。这种通信通常由服务器端基础设施和 API 的后端实现来促进。此后端的目的可能是提供对数据的访问，而无需直接连接到数据库，从而最大限度地减少对数据库的请求数量。
在 API 通信中，我们通常有一个指定的 URL，该端点可能在网站URL显示为/api/item/5。这种方法非常有利，因为它允许我们利用同一个URL从各种设备请求数据并接收相同的响应。
HTTP 方法
尽管我不想深入讨论理论概念，但了解使用 API 时不同方法之间的区别非常重要。让我们快速回顾一下最常用的方法：
- GET — 用于检索数据
- POST — 写入新数据
- DELETE— 删除数据
- PUT — 更新现有数据
虽然还有其他选项，但这些是我们应该从中入门的基本选项。POST 和 PUT 之间的区别很重要。两者都可用于添加数据，但 POST 用于写入新项目，而 PUT 用于使用较新的值更新现有数据。
构建你的第一个 API
运行第一个 FastAPI 服务器
首先，我们需要安装所有依赖项。至少需要将fastapi包与uvicorn服务器和一起安装pydantic。第一行应该安装所有列出的库。
?pip install fastapi[all]
?pip install uvicorn
?pip install pydantic现在，让我们使用带有必要方法和端点的装饰器创建一个基本的FastAPI应用程序。在此示例中，我们将使用 GET 方法检索数据。在本文中，我们不会讨论装饰器，因为它超出了 API 主题的范围。
在该函数之前，可能会看到下面一行代码，它负责将所有FastAPI进程添加到我们的方法中。
?from fastapi import FastAPI
?
?app = FastAPI()
??
?@app.get('/')
?def first_response():
? ? ?return {"response": "first"}要查看响应，可以使用 uvicorn 运行服务器。默认情况下，服务器在端口 8000 上，并可通过 http://127.0.0.1:8000 访问。在开发过程中，可以使用 --reload 选项确保服务器对代码所做的任何更改都会自动重启。
?uvicorn main:app
?uvicorn main:app --reload通常我们可以使用 terminal 可以使用curl测试一下响应。
?curl 127.0.0.1:8000
?#{"response": "first"}
??
?curl 127.0.0.1:8000 -X POST
?#{"detail":"Method Not Allowed"}最后，我们可以使用 requests 库来访问我们的端点，并在 Python 中打印响应。
?import requests
??
?print(requests.get('http://127.0.0.1:8000').json())
?#{'response': 'first'}使用JSON文件中的数据
让我们继续下一步，开始处理一些实际数据。为了简化目前的情况，我们可以创建一个带有一些输入内容的 JSON 文件。
我们随便找一段 json 数据：
?{
? ? ?"stocks": [
? ? ? ? {
? ? ? ? ? ? ?"symbol": "TSLA",
? ? ? ? ? ? ?"stockname": "Tesla Inc. Common Stock",
? ? ? ? ? ? ?"lastsale": "$235.45",
? ? ? ? ? ? ?"country": "United States",
? ? ? ? ? ? ?"ipoyear": 2010
? ? ? ? },
? ? ? ? {
? ? ? ? ? ? ?"symbol": "NVDA",
? ? ? ? ? ? ?"stockname": "NVIDIA Corporation Common Stock",
? ? ? ? ? ? ?"lastsale": "$477.76",
? ? ? ? ? ? ?"country": "United States",
? ? ? ? ? ? ?"ipoyear": 1999
? ? ? ? },
? ? ? ? {
? ? ? ? ? ? ?"symbol": "AMZN",
? ? ? ? ? ? ?"stockname": "Amazon.com Inc. Common Stock",
? ? ? ? ? ? ?"lastsale": "$146.74",
? ? ? ? ? ? ?"country": "United States",
? ? ? ? ? ? ?"ipoyear": 1997
? ? ? ? }
? ? ]
?}现在我们可以修改我们的代码，不再检索所有项目，而是仅检索包含我们想要的特定符号的项目。也就是在后端执行过滤操作。为了实现这一点，我们执行以下操作：
- 导入 pydantic 的模块。
- 读取 JSON 文件并将其内容存储在字典中。
- 使用过滤器，仅从该字典中检索所需项目。 如果未找到则引发带有 404 状态码（Not Found）的异常。
看起来很简单，但这是每个 FastAPI 应用程序的基础部分 — 模式、数据和方法。
?from fastapi import FastAPI, HTTPException, Query
?from pydantic import BaseModel
?from typing import Optional
?import json
??
?app = FastAPI()
??
?class Stock(BaseModel):
? ? ?symbol: str
? ? ?stockname: str
? ? ?lastsale: str
? ? ?country: str
? ? ?ipoyear: Optional[int] = None
? ? ?
?with open('stocks.json', 'r') as f:
? ? ?stocks = json.load(f)['stocks']
??
?@app.get('/stock/{stock_symbol}', status_code=200)
?def get_stock(stock_symbol: str) -> Stock:
? ? ?stock = [stock for stock in stocks if stock['symbol'] == stock_symbol]
? ? ?if len(stock) == 0:
? ? ? ? ?raise HTTPException(
? ? ? ? ? ? ?status_code=404, detail=f"No stock {stock_symbol} found."
? ? ? ? ? )
??
? ? ?return stock[0]当然，现在我们需要更改获数据请求的 url。
?import requests
??
?print(requests.get('http://127.0.0.1:8000/stock/AMZN').json())如上面所述，我们的文档现在应该是可用的。可以通过在浏览器的地址栏中输入http://127.0.0.1:8000/docs 来访问它。
FastAPI 内置了 OpenAPI 的文档系统，可基于我们的代码生成标准 API 文档。
随着我们继续添加更多方法，可以轻松地导航到同一位置找到它们所有。这份文档对刚开始使用 API 的人尤其有帮助。
在我们的代码中确定了 Pydantic 模型之后，现在我们可以确定输出响应的模式。这也是更好地理解各种 API 方法及其返回内容的一个更好的方式。
链接 Postgres 数据库
过去，我们只使用本地文件。然而，在大多数情况下将需要在后端使用数据库。为了实现这一点，我们将连接PostgreSQL数据库，并尝试在我们的GET方法中使用它。这是一个简单的SELECT语句。但是我们需要正确识别所有内容以供FastAPI使用。
这个过程涉及到 SQLAlchem 库，它是 Python 中最流行的用于对象关系映射（ORM）操作的包之一。
为了将数据存储在单独的文件中，创建一个名为config.py的文件，并添加以下代码。下面是示例代码：
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
sqlalchemy_string: str = "postgresql://user:passwordp@host/db"
settings = Settings()让我们通过将这部分内容结构化到 database.py 文件中来创建引擎并为 FastAPI 会话准备数据库。我们使用 config.py 文件中的设置。
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from config import settings
engine = create_engine(
settings.sqlalchemy_string, connect_args={'sslmode':'require'}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()接下来，我们需要将模型与数据库中相关的表格进行关联。我们具体处理数据的表格，并且将为其使用一个简单的模型。我们会使用基本模型。
from sqlalchemy import Column, Integer, String, Float, BigInteger
from database import Base
class Stock(Base):
__tablename__ = "nasdaq_stocks"
symbol = Column(String, primary_key=True)
stockname = Column(String)
lastsale = Column(String)
netchange = Column(Float)
percentchange = Column(String)
marketcap = Column(BigInteger)
country = Column(String, nullable=True)
ipoyear = Column(Integer, nullable=True)
volume = Column(Integer)
sector = Column(String, nullable=True)
industry = Column(String, nullable=True)我们将 Pydantic 模型存储在单独的文件中。让我们将文件命名为 schemas.py，并在其中添加相关模型。正确配置 orm_mode，因为我们正在使用 SQLAlchemy 数据库。
from pydantic import BaseModel
from typing import Optional
class StockBase(BaseModel):
symbol: str
stockname: str
lastsale: str
country: str
ipoyear: Optional[int] = None
volume: int
class StockCreate(StockBase):
pass
class Stock(StockBase):
class Config:
orm_mode = True我们必须在数据库指定 CRUD（Create, Read, Update, Delete） 操作的代码。以后在主脚本中仅使用 FastAPI 应用程序中的函数会更方便。 对于我们基本的 GET 方法，仅需要通过符号进行简单的过滤查询即可, 下面是一个最基础的 crud.py 文件示例。
from sqlalchemy.orm import Session
import models, schemas
def get_stock(db: Session, symbol: str):
return db.query(models.Stock).filter(models.Stock.symbol == symbol).first()我们已经完成了实现 API 所需的所有准备工作。由于正在操作数据库，需要在脚本中包含一些额外的细节。
get_db 函数负责与数据库建立连接，并已将其包含在 Depends FastAPI 类中。下面是可运行代码的最终示例。
from fastapi import FastAPI, HTTPException, Query, Depends
from sqlalchemy.orm import Session
import crud, models, schemas
from database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI(
title="NASDAQ stocks",
description="Start using FastAPI in development",
version="0.1"
)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get('/stock/{symbol}', response_model=schemas.Stock, status_code=200)
def get_stock(symbol: str, db: Session = Depends(get_db)) -> models.Stock:
db_stock = crud.get_stock(db, symbol=symbol)
if db_stock is None:
raise HTTPException(
status_code=404, detail=f"No stock {symbol} found."
)
return db_stock虽然没有明显变化，但我们现在可以搜索整个数据库，而不仅仅是来自 JSON 文件中的数据。修改输出也很简单，因为我们可以在 Pydantic 模型中添加或删除字段。
import requests
print(requests.get('http://127.0.0.1:8000/stock/AAL').json())ORM 与市场上提供的各种数据库选项无缝配合，实现高效集成不需要进行任何修改。可以参考关于如何在 FastAPI 中使用 SQL 数据库的相关文档。
总结
在本文中，我们描述了 FastAPI 及其简化 REST API 实现的能力。 与其他有用的 Python 依赖项一起，FastAPI 提供许多必要的功能：
- Pydantic 使用Pydantic 对数据进行验证。
- SQLAlchemy 使用SQLAlchemy ORM 对数据进行数据库操作。
FastAPI 不仅局限于使用 GET 方法返回数据作为响应。它提供了完整的 REST API 功能，包括POST、PUT 和 DELETE 等其他有价值的方法。
作者：demo007x
链接：
https://juejin.cn/post/7387550356375994378