fastapi 框架怎么封装连接数据库,并且读取.env文件连接数据库?

FastAPI是一个用于构建Web API的高级Python框架，它轻量级且易于上手。在FastAPI中，你可以通过各种ORM（如SQLAlchemy、Peewee等）或直接使用数据库连接库（如asyncio、aiomysql等）来处理数据库操作。以下是使用FastAPI封装连接数据库，并从`.env`文件获取数据库配置的一个基本示例：

1. 首先，你需要安装所需的依赖。例如，如果你选择使用SQLAlchemy，可以运行：

   pip install fastapi uvicorn sqlalchemy aiopg async-timeout python-dotenv

2. 在`main.py`或其他启动文件中，引入必要的模块并设置环境变量读取：

   from dotenv import load_dotenv
   import os

   load_dotenv()  # 加载.env文件

   DATABASE_URL = os.environ.get("DATABASE_URL")  # 获取数据库URL

3. 使用`SQLAlchemy`创建一个数据库工厂函数，根据配置动态创建连接池：

   from sqlalchemy import create_engine
   from sqlalchemy.ext.asyncio import AsyncEngine

   def get_db():
       engine = create_engine(DATABASE_URL, echo=True) if os.getenv('DEBUG') else create_engine(DATABASE_URL)
       return AsyncEngine(engine)

   # 现在你可以将get_db()注入到需要访问数据库的地方，比如FastAPI的Depends装饰器中
   from fastapi import FastAPI, Depends
   app = FastAPI()

   @app.on_event("startup")
   async def startup():
       await get_db().connect()

4. 在模型层，定义数据表对应的ORM类：

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

   Base = declarative_base()

   class User(Base):
       __tablename__ = "users"
       id = Column(Integer, primary_key=True)
       name = Column(String)

5. 调用数据库时，使用依赖注入确保在每个请求上下文中都有数据库连接：

   async def get_user(db: AsyncConnection = Depends(get_db)):
       user = await db.execute("SELECT * FROM users WHERE id = :id", {"id": 1})
       result = await user.fetchone()
       return User(**result)

6. 在FastAPI路由中使用这个函数：

   @app.get("/user/{user_id}")
   async def read_user(user_id: int, db=Depends(get_db)):
       # ...其他业务逻辑...