正文

快速上手FastAPI：轻松实现数据库数据的导入与导出技巧

/0 浏览量
1106

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，与 Python 3.6+ 类型提示一起使用。它具有异步处理能力，这意味着它可以同时处理多个请求，而不需要为每个请求创建新的线程或进程。在本文中，我们将探讨如何使用 FastAPI 实现数据库数据的导入与导出。

数据库选择

在开始之前，我们需要选择一个数据库。FastAPI 支持多种数据库，如 PostgreSQL、MySQL、SQLite 等。这里，我们将使用 SQLite，因为它轻量级、易于设置，且不需要额外的数据库服务器。

安装依赖

首先，确保你已经安装了 fastapi 和 uvicorn。如果你还没有安装，可以使用以下命令：

pip install fastapi uvicorn

然后，安装 sqlalchemy 和 databases，用于数据库操作：

pip install sqlalchemy databases

创建数据库模型

使用 SQLAlchemy 定义数据库模型。以下是一个简单的用户模型示例：

from pydantic import BaseModel
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base, BaseModel):
    __tablename__ = "users"
    id: int = Column(Integer, primary_key=True, index=True)
    name: str = Column(String, index=True)
    age: int = Column(Integer, index=True)

初始化数据库

使用 SQLAlchemy 创建数据库和表：

from sqlalchemy import create_engine

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

Base.metadata.create_all(bind=engine)

数据导入

数据导入可以通过多种方式实现，例如从 CSV 文件或 JSON 文件中读取数据。以下是一个从 CSV 文件导入数据的示例：

import csv
from sqlalchemy.orm import Session

def import_data_from_csv(csv_file_path: str):
    with open(csv_file_path, "r") as file:
        reader = csv.DictReader(file)
        session = Session(engine)
        for row in reader:
            user = User(**row)
            session.add(user)
        session.commit()
        session.close()

数据导出

数据导出可以通过将数据写入 CSV 文件或 JSON 文件来实现。以下是一个将数据导出到 CSV 文件的示例：

from sqlalchemy.orm import Session

def export_data_to_csv(csv_file_path: str):
    session = Session(engine)
    with open(csv_file_path, "w", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=["id", "name", "age"])
        writer.writeheader()
        for user in session.query(User).all():
            writer.writerow({"id": user.id, "name": user.name, "age": user.age})
    session.close()

创建 FastAPI 应用

现在，我们可以创建一个 FastAPI 应用，并使用上述函数：

from fastapi import FastAPI

app = FastAPI()

@app.post("/import/")
async def import_data(csv_file: UploadFile = File(...)):
    import_data_from_csv(csv_file.filename)
    return {"message": "Data imported successfully"}

@app.post("/export/")
async def export_data(csv_file_path: str = Depends(FilePath(...))):
    export_data_to_csv(csv_file_path)
    return {"message": "Data exported successfully"}

运行应用

使用以下命令启动 FastAPI 应用：

uvicorn main:app --reload

现在，你可以通过访问 /import/ 和 /export/ 端点来导入和导出数据。

总结

在本文中，我们学习了如何使用 FastAPI 和 SQLAlchemy 实现数据库数据的导入与导出。通过这些示例，你可以轻松地将这些技术应用到你的项目中。希望这篇文章能帮助你快速上手 FastAPI！

-- 展开阅读全文 --