最近一年公司也在卷 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() .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'] .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() .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 等其他有价值的方法。
