logo

开发者热搜

快速上手：Python — 使用 FastAPI 和 PostgreSQL 构建简单 API

作者：rousong2025.09.23 11:56浏览量：0

简介：本文将指导读者使用 FastAPI 和 PostgreSQL 构建一个简单的 RESTful API，涵盖环境配置、数据库连接、API 路由设计及测试方法，帮助开发者快速掌握关键技能。

快速上手：Python — 使用 FastAPI 和 PostgreSQL 构建简单 API

引言

在现代化软件开发中，API（应用程序接口）已成为连接不同系统、服务或组件的核心桥梁。无论是移动应用、Web 服务还是微服务架构，API 都承担着数据交互与功能调用的关键角色。随着 Python 生态的繁荣，FastAPI 以其高性能、易用性和类型安全特性迅速成为构建 API 的热门选择；而 PostgreSQL 作为功能强大的开源关系型数据库，以其稳定性、扩展性和丰富的数据类型支持，成为企业级应用的首选。本文将详细介绍如何结合 FastAPI 和 PostgreSQL 构建一个简单但完整的 RESTful API，涵盖环境配置、数据库连接、API 路由设计及测试方法，帮助开发者快速上手这一技术组合。

为什么选择 FastAPI 和 PostgreSQL？

FastAPI 的优势

FastAPI 是一个基于 Python 的现代 Web 框架，专为构建 API 设计。其核心优势包括：

  • 高性能：基于 Starlette（ASGI 框架）和 Pydantic（数据验证库），FastAPI 的响应速度接近 Node.js 和 Go，远超传统框架如 Flask 或 Django。
  • 自动生成 API 文档：内置支持 OpenAPI（Swagger）和 ReDoc，无需额外配置即可生成交互式文档，极大提升开发效率。
  • 类型注解支持：利用 Python 的类型提示（Type Hints）和 Pydantic 模型，实现编译时类型检查，减少运行时错误。
  • 异步支持：原生支持异步请求处理（async/await），适合高并发场景。

PostgreSQL 的优势

PostgreSQL 是一个功能全面的开源关系型数据库，其特点包括：

  • 事务支持：遵循 ACID（原子性、一致性、隔离性、持久性）原则，确保数据完整性。
  • 扩展性：支持 JSON、地理空间数据、全文搜索等高级功能，满足复杂业务需求。
  • 社区活跃：拥有庞大的开发者社区和丰富的第三方工具（如 psycopg2、SQLAlchemy），便于集成。

环境准备

安装 Python 和依赖

  1. Python 版本：确保系统已安装 Python 3.7+（FastAPI 和 Pydantic 的最低要求）。
  2. 虚拟环境：使用 venv 或 conda 创建隔离环境，避免依赖冲突。
    1. python -m venv fastapi_env
    2. source fastapi_env/bin/activate  # Linux/macOS
    3. fastapi_env\Scripts\activate     # Windows
  3. 安装依赖：通过 pip 安装 FastAPI、Uvicorn（ASGI 服务器）和 PostgreSQL 适配器。
    1. pip install fastapi uvicorn psycopg2-binary
    • psycopg2-binary 是 PostgreSQL 的 Python 适配器，适合开发环境；生产环境建议使用 psycopg2（需编译）。

配置 PostgreSQL 数据库

  1. 安装 PostgreSQL：从 官网 下载并安装，或使用云服务（如 AWS RDS、ElephantSQL）。
  2. 创建数据库和用户：
    • 登录 PostgreSQL：
      1. psql -U postgres
    • 创建数据库和用户（替换 your_db、your_user 和 your_password）：
      1. CREATE DATABASE your_db;
      2. CREATE USER your_user WITH PASSWORD 'your_password';
      3. GRANT ALL PRIVILEGES ON DATABASE your_db TO your_user;

构建 FastAPI 应用

项目结构

建议采用以下目录结构，便于维护：

  1. fastapi_postgres/
  2. ├── main.py            # 主应用文件
  3. ├── models.py          # 数据库模型
  4. ├── schemas.py         # Pydantic 模型（请求/响应体）
  5. ├── database.py        # 数据库连接
  6. └── requirements.txt   # 依赖列表

数据库连接（database.py）

使用 psycopg2 或 SQLAlchemy 建立连接。以下是基于 psycopg2 的示例：

  1. import psycopg2
  2. from psycopg2.extras import RealDictCursor
  4. DATABASE_URL = "postgresql://your_user:your_password@localhost:5432/your_db"
  6. def get_db_connection():
  7.     try:
  8.         conn = psycopg2.connect(DATABASE_URL, cursor_factory=RealDictCursor)
  9.         return conn
  10.     except Exception as e:
  11.         print(f"Database connection error: {e}")
  12.         raise
  • RealDictCursor 返回字典格式的结果，便于处理。

数据库模型（models.py）

定义数据库表结构。例如，创建一个 items 表：

  1. class Item:
  2.     def __init__(self, id: int, name: str, price: float):
  3.         self.id = id
  4.         self.name = name
  5.         self.price = price
  7.     @classmethod
  8.     def create_table(cls, conn):
  9.         with conn.cursor() as cursor:
  10.             cursor.execute("""
  11.                 CREATE TABLE IF NOT EXISTS items (
  12.                     id SERIAL PRIMARY KEY,
  13.                     name VARCHAR(100) NOT NULL,
  14.                     price DECIMAL(10, 2) NOT NULL
  15.                 )
  16.             """)
  17.             conn.commit()
  • SERIAL 是自增整数类型，DECIMAL 用于精确存储价格。

Pydantic 模型（schemas.py）

定义请求和响应的数据结构，利用类型注解实现验证：

  1. from pydantic import BaseModel
  3. class ItemCreate(BaseModel):
  4.     name: str
  5.     price: float
  7. class ItemResponse(BaseModel):
  8.     id: int
  9.     name: str
  10.     price: float
  • ItemCreate 用于接收创建请求，ItemResponse 用于返回响应。

主应用（main.py）

集成 FastAPI、数据库和路由：

  1. from fastapi import FastAPI, HTTPException
  2. from pydantic import BaseModel
  3. import psycopg2
  4. from database import get_db_connection
  5. from models import Item
  6. from schemas import ItemCreate, ItemResponse
  8. app = FastAPI()
  10. # 初始化数据库表
  11. conn = get_db_connection()
  12. Item.create_table(conn)
  13. conn.close()
  15. @app.post("/items/", response_model=ItemResponse)
  16. def create_item(item: ItemCreate):
  17.     conn = get_db_connection()
  18.     try:
  19.         with conn.cursor() as cursor:
  20.             cursor.execute(
  21.                 "INSERT INTO items (name, price) VALUES (%s, %s) RETURNING id",
  22.                 (item.name, item.price)
  23.             )
  24.             item_id = cursor.fetchone()["id"]
  25.             conn.commit()
  26.         return {"id": item_id, "name": item.name, "price": item.price}
  27.     except Exception as e:
  28.         conn.rollback()
  29.         raise HTTPException(status_code=500, detail=str(e))
  30.     finally:
  31.         conn.close()
  33. @app.get("/items/{item_id}", response_model=ItemResponse)
  34. def read_item(item_id: int):
  35.     conn = get_db_connection()
  36.     try:
  37.         with conn.cursor() as cursor:
  38.             cursor.execute("SELECT * FROM items WHERE id = %s", (item_id,))
  39.             item = cursor.fetchone()
  40.             if not item:
  41.                 raise HTTPException(status_code=404, detail="Item not found")
  42.             return item
  43.     finally:
  44.         conn.close()
  • response_model 指定返回的数据结构，FastAPI 会自动转换并验证。
  • 使用 try-except-finally 确保数据库连接正确关闭。

运行和测试 API

启动 FastAPI 服务

使用 Uvicorn 运行应用：

  1. uvicorn main:app --reload
  • --reload 启用开发模式，代码修改后自动重启。

测试 API

  1. 访问 Swagger 文档：打开 http://127.0.0.1:8000/docs，交互式测试 API。
  2. 使用 curl 或 Postman：
    • 创建 item：
      1. curl -X POST "http://127.0.0.1:8000/items/" -H "Content-Type: application/json" -d '{"name": "Laptop", "price": 999.99}'
    • 获取 item：
      1. curl "http://127.0.0.1:8000/items/1"

扩展与优化

使用 SQLAlchemy

对于复杂项目，推荐使用 SQLAlchemy（ORM）管理数据库：

  1. from sqlalchemy import create_engine, Column, Integer, String, Float
  2. from sqlalchemy.ext.declarative import declarative_base
  3. from sqlalchemy.orm import sessionmaker
  5. DATABASE_URL = "postgresql://your_user:your_password@localhost:5432/your_db"
  6. engine = create_engine(DATABASE_URL)
  7. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
  8. Base = declarative_base()
  10. class Item(Base):
  11.     __tablename__ = "items"
  12.     id = Column(Integer, primary_key=True, index=True)
  13.     name = Column(String(100), nullable=False)
  14.     price = Column(Float, nullable=False)
  • ORM 简化数据库操作，支持更复杂的查询。

添加认证和授权

使用 FastAPI 的 Dependency 和 OAuth2 实现 JWT 认证：

  1. from fastapi import Depends, HTTPException
  2. from fastapi.security import OAuth2PasswordBearer
  4. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  6. def get_current_user(token: str = Depends(oauth2_scheme)):
  7.     # 验证 token 并返回用户信息
  8.     pass

部署建议

  • 容器化：使用 Docker 打包应用和 PostgreSQL，便于部署。
  • 云服务：考虑 AWS RDS（PostgreSQL）和 Heroku/Vercel（FastAPI）。
  • 监控：集成 Prometheus 和 Grafana 监控 API 性能。

总结

本文详细介绍了如何使用 FastAPI 和 PostgreSQL 构建一个简单的 RESTful API，涵盖环境配置、数据库连接、API 路由设计及测试方法。FastAPI 的高性能和自动文档生成特性，结合 PostgreSQL 的稳定性和扩展性，为开发者提供了强大的工具链。通过实践，读者可以快速掌握这一技术组合，并进一步扩展为生产级应用。未来，可探索异步数据库操作、GraphQL 集成或微服务架构，以适应更复杂的业务场景。

相关文章推荐

发表评论

最热文章

关于作者

  • 被阅读数
  • 被赞数
  • 被收藏数