使用FastAPI来开发项目，项目的目录结构如何规划的一些参考和基类封装的一些处理

使用FastAPI开发项目时，良好的目录结构可以帮助你更好地组织代码，提高可维护性和扩展性。同样，对基类的封装，也可以进一步减少开发代码，提供便利，并减少出错的几率。

下面是一个推荐的目录结构示例：

my_fastapi_project/ ├── app/ │   ├── __init__.py │   ├── main.py            # 入口文件 │   ├── core/ │   │   ├── __init__.py │   │   ├── config.py      # 配置文件 │   │   ├── security.py    # 安全相关 │   │   └── ...            # 其他核心功能 │   ├── api/ │   │   ├── __init__.py │   │   ├── v1/ │   │   │   ├── __init__.py │   │   │   ├── endpoints/ │   │   │   │   ├── __init__.py │   │   │   │   ├── users.py     # 用户相关接口 │   │   │   │   ├── items.py     # 其他接口 │   │   │   │   └── ... │   │   │   └── ...              # 其他版本的API │   ├── models/ │   │   ├── __init__.py │   │   ├── user.py         # 用户模型 │   │   ├── item.py         # 其他模型 │   │   └── ... │   ├── schemas/ │   │   ├── __init__.py │   │   ├── user.py         # 用户数据模型 │   │   ├── item.py         # 其他数据模型 │   │   └── ... │   ├── crud/ │   │   ├── __init__.py │   │   ├── user.py         # 用户CRUD操作 │   │   ├── item.py         # 其他CRUD操作 │   │   └── ... │   ├── db/ │   │   ├── __init__.py │   │   ├── base.py         # 数据库基础设置 │   │   ├── session.py      # 数据库会话 │   │   └── ... │   ├── tests/ │   │   ├── __init__.py │   │   ├── test_main.py    # 测试主文件 │   │   ├── test_users.py   # 用户相关测试 │   │   └── ... │   └── utils/ │       ├── __init__.py │       ├── utils.py        # 工具函数 │       └── ... ├── .env                    # 环境变量文件 ├── alembic/                # 数据库迁移工具目录 │   ├── env.py │   ├── script.py.mako │   └── versions/ │       └── ... ├── alembic.ini             # Alembic 配置文件 ├── requirements.txt        # 项目依赖 ├── Dockerfile              # Docker 配置文件 └── README.md               # 项目说明文件

目录结构说明：

• app/: 项目的主目录，包含所有应用相关代码。
  • main.py: 项目的入口文件，启动FastAPI应用。
  • core/: 核心功能，如配置、安全等。
  • api/: API路由和视图，分版本管理。
  • models/: 数据库模型。
  • schemas/: 数据模型，用于请求和响应的验证。
  • crud/: 数据库操作（CRUD：创建、读取、更新、删除）。
  • db/: 数据库相关设置和会话管理。
  • tests/: 测试代码。
  • utils/: 工具函数和公用模块。
• .env: 环境变量文件，用于存储敏感信息，如数据库连接字符串。
• alembic/: 数据库迁移工具Alembic的配置目录。
• requirements.txt: 项目依赖列表。
• Dockerfile: Docker配置文件，用于容器化部署。
• README.md: 项目说明文件。

这个结构可以根据项目需求进行调整，但保持清晰和模块化是良好的实践。

 

python项目总的__init__.py，有意义吗？

在Python项目中，__init__.py 文件的主要作用是将目录标识为一个Python包。它使得目录中的模块可以被导入和使用。在一些情况下，__init__.py 可以不仅仅是一个空文件，还可以包含一些初始化代码。

__init__.py 的意义：

1. 将目录标识为包：

   • 任何包含 __init__.py 的目录都会被Python解释器认为是一个包，这样你就可以使用包导入语法，如 import mypackage.module。

2. 初始化代码：

• 可以在 __init__.py 中包含一些初始化代码，如导入包内的子模块、设置包级别的变量或函数、配置日志记录等。例如：

# mypackage/__init__.py from .submodule1 import func1 from .submodule2 import func2  __all__ = ["func1", "func2"]

        3.简化导入

• • 　通过在 __init__.py 中导入子模块，可以简化包的导入路径，使得用户可以直接从包中导入函数或类，而不必知道具体的模块结构。

# mypackage/__init__.py from .submodule import MyClass  # Now you can do from mypackage import MyClass

对于Python 3.3及以上版本，__init__.py 文件不是强制性的，即使没有 __init__.py 文件，Python解释器也可以识别包。然而，添加 __init__.py 文件仍然是一个良好的习惯，可以避免某些情况下的意外行为，并且明确表示该目录是一个包。

 

2、Fast API项目的开发处理过程

在FastAPI项目中，CRUD操作通常在一个专门的 crud 模块中实现。这个模块会调用SQLAlchemy模型对象来进行数据库操作。

1. 定义模型 (models/user.py)

from sqlalchemy import Column, Integer, String from app.db.base_class import Base  class User(Base):     __tablename__ = "users"          id = Column(Integer, primary_key=True, index=True)     email = Column(String, unique=True, index=True, nullable=False)     hashed_password = Column(String, nullable=False)     full_name = Column(String, index=True)

2. 创建数据库会话 (db/session.py)

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker  DATABASE_URL = "sqlite:///./test.db"  # 使用SQLite数据库作为示例  engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

3. 定义CRUD操作 (crud/user.py)

from sqlalchemy.orm import Session from app.models.user import User from app.schemas.user import UserCreate, UserUpdate  def get_user(db: Session, user_id: int):     return db.query(User).filter(User.id == user_id).first()  def get_user_by_email(db: Session, email: str):     return db.query(User).filter(User.email == email).first()  def get_users(db: Session, skip: int = 0, limit: int = 10):     return db.query(User).offset(skip).limit(limit).all()  def create_user(db: Session, user: UserCreate):     db_user = User(         email=user.email,         hashed_password=user.hashed_password,  # 在实际应用中应该对密码进行哈希处理         full_name=user.full_name     )     db.add(db_user)     db.commit()     db.refresh(db_user)     return db_user  def update_user(db: Session, user_id: int, user: UserUpdate):     db_user = get_user(db, user_id)     if db_user:         db_user.email = user.email         db_user.full_name = user.full_name         db.commit()         db.refresh(db_user)     return db_user  def delete_user(db: Session, user_id: int):     db_user = get_user(db, user_id)     if db_user:         db.delete(db_user)         db.commit()     return db_user

4. 定义数据模型 (schemas/user.py)

from pydantic import BaseModel  class UserBase(BaseModel):     email: str     full_name: str = None  class UserCreate(UserBase):     hashed_password: str  class UserUpdate(UserBase):     pass  class User(UserBase):     id: int      class Config:         orm_mode = True

5. 在API端点中使用CRUD操作 (api/v1/endpoints/users.py)

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from app import crud, models, schemas from app.db.session import SessionLocal  router = APIRouter()  def get_db():     db = SessionLocal()     try:         yield db     finally:         db.close()  @router.post("/users/", response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):     db_user = crud.get_user_by_email(db, email=user.email)     if db_user:         raise HTTPException(status_code=400, detail="Email already registered")     return crud.create_user(db=db, user=user)  @router.get("/users/{user_id}", response_model=schemas.User) def read_user(user_id: int, db: Session = Depends(get_db)):     db_user = crud.get_user(db, user_id=user_id)     if db_user is None:         raise HTTPException(status_code=404, detail="User not found")     return db_user  @router.put("/users/{user_id}", response_model=schemas.User) def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):     db_user = crud.update_user(db=db, user_id=user_id, user=user)     if db_user is None:         raise HTTPException(status_code=404, detail="User not found")     return db_user  @router.delete("/users/{user_id}", response_model=schemas.User) def delete_user(user_id: int, db: Session = Depends(get_db)):     db_user = crud.delete_user(db=db, user_id=user_id)     if db_user is None:         raise HTTPException(status_code=404, detail="User not found")     return db_user  @router.get("/users/", response_model=List[schemas.User]) def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):     users = crud.get_users(db, skip=skip, limit=limit)     return users

6. 注册路由 (main.py)

from fastapi import FastAPI from app.api.v1.endpoints import users  app = FastAPI()  app.include_router(users.router, prefix="/api/v1", tags=["users"])  if __name__ == "__main__":     import uvicorn     uvicorn.run(app, host="0.0.0.0", port=8000)

7. 初始化数据库 (db/base.py)

from app.db.session import engine from app.models import user  user.Base.metadata.create_all(bind=engine)

8. 运行应用

在项目根目录下运行:

uvicorn app.main:app --reload

这样，你的CRUD层就可以调用模型对象来进行数据库操作了。上述代码展示了如何定义模型、数据库会话、CRUD操作、数据模型和API端点，并将它们结合在一起，实现一个简单的用户管理系统。

 

3、实际FastAPI项目对基类的封装

可以通过创建一个通用的CRUD基类来封装常规的CRUD操作，然后让特定的CRUD类继承这个基类。这样可以减少重复代码，提高代码的可维护性和可复用性。下面是一个实现示例。

1、创建通用CRUD基类 (crud/base.py)

rom typing import Generic, Type, TypeVar, Optional, List from pydantic import BaseModel from sqlalchemy.orm import Session from app.db.base_class import Base  ModelType = TypeVar("ModelType", bound=Base) CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)  class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):     def __init__(self, model: Type[ModelType]):         self.model = model      def get(self, db: Session, id: int) -> Optional[ModelType]:         return db.query(self.model).filter(self.model.id == id).first()      def get_multi(self, db: Session, skip: int = 0, limit: int = 100) -> List[ModelType]:         return db.query(self.model).offset(skip).limit(limit).all()      def create(self, db: Session, obj_in: CreateSchemaType) -> ModelType:         obj_in_data = obj_in.dict()         db_obj = self.model(**obj_in_data)  # type: ignore         db.add(db_obj)         db.commit()         db.refresh(db_obj)         return db_obj      def update(self, db: Session, db_obj: ModelType, obj_in: UpdateSchemaType) -> ModelType:         obj_data = db_obj.dict()         update_data = obj_in.dict(skip_defaults=True)         for field in obj_data:             if field in update_data:                 setattr(db_obj, field, update_data[field])         db.commit()         db.refresh(db_obj)         return db_obj      def remove(self, db: Session, id: int) -> ModelType:         obj = db.query(self.model).get(id)         db.delete(obj)         db.commit()         return obj

2、定义用户CRUD操作 (crud/user.py)

from typing import Any from sqlalchemy.orm import Session from app.crud.base import CRUDBase from app.models.user import User from app.schemas.user import UserCreate, UserUpdate  class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):     def get_by_email(self, db: Session, email: str) -> Any:         return db.query(self.model).filter(self.model.email == email).first()  user = CRUDUser(User)

3、定义数据模型 (schemas/user.py)

from pydantic import BaseModel  class UserBase(BaseModel):     email: str     full_name: str = None  class UserCreate(UserBase):     hashed_password: str  class UserUpdate(UserBase):     pass  class User(UserBase):     id: int      class Config:         orm_mode = True

4、在API端点中使用CRUD操作 (api/v1/endpoints/users.py)

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from typing import List from app import crud, schemas from app.db.session import SessionLocal from app.models.user import User  router = APIRouter()  def get_db():     db = SessionLocal()     try:         yield db     finally:         db.close()  @router.post("/users/", response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):     db_user = crud.user.get_by_email(db, email=user.email)     if db_user:         raise HTTPException(status_code=400, detail="Email already registered")     return crud.user.create(db=db, obj_in=user)  @router.get("/users/{user_id}", response_model=schemas.User) def read_user(user_id: int, db: Session = Depends(get_db)):     db_user = crud.user.get(db, id=user_id)     if db_user is None:         raise HTTPException(status_code=404, detail="User not found")     return db_user  @router.put("/users/{user_id}", response_model=schemas.User) def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):     db_user = crud.user.get(db=db, id=user_id)     if db_user is None:         raise HTTPException(status_code=404, detail="User not found")     return crud.user.update(db=db, db_obj=db_user, obj_in=user)  @router.delete("/users/{user_id}", response_model=schemas.User) def delete_user(user_id: int, db: Session = Depends(get_db)):     db_user = crud.user.get(db=db, id=user_id)     if db_user is None:         raise HTTPException(status_code=404, detail="User not found")     return crud.user.remove(db=db, id=user_id)  @router.get("/users/", response_model=List[schemas.User]) def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):     users = crud.user.get_multi(db, skip=skip, limit=limit)     return users

其他的就是类似前面的做法了。

通过这种方式，你可以在通用的CRUD基类中封装常规的CRUD操作，而特定的CRUD类（如 CRUDUser）只需要继承这个基类并添加特定的操作方法。这样不仅减少了重复代码，也提高了代码的可维护性和可复用性。

如果你希望可以通过定义一个通用的API基类来封装常规的CRUD操作方法，然后在具体的端点文件中继承这个基类。这样可以进一步减少重复代码，提高代码的可维护性和可复用性。

 创建通用API基类 (api/deps.py)

from typing import Type, TypeVar, Generic, List from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from pydantic import BaseModel from app.crud.base import CRUDBase from app.db.session import SessionLocal  ModelType = TypeVar("ModelType", bound=BaseModel) CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)  class CRUDRouter(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):     def __init__(self, crud: CRUDBase[ModelType, CreateSchemaType, UpdateSchemaType]):         self.crud = crud         self.router = APIRouter()          self.router.post("/", response_model=ModelType)(self.create_item)         self.router.get("/{item_id}", response_model=ModelType)(self.read_item)         self.router.put("/{item_id}", response_model=ModelType)(self.update_item)         self.router.delete("/{item_id}", response_model=ModelType)(self.delete_item)         self.router.get("/", response_model=List[ModelType])(self.read_items)      def get_db(self):         db = SessionLocal()         try:             yield db         finally:             db.close()      async def create_item(self, item_in: CreateSchemaType, db: Session = Depends(self.get_db)):         db_item = self.crud.create(db=db, obj_in=item_in)         return db_item      async def read_item(self, item_id: int, db: Session = Depends(self.get_db)):         db_item = self.crud.get(db=db, id=item_id)         if not db_item:             raise HTTPException(status_code=404, detail="Item not found")         return db_item      async def update_item(self, item_id: int, item_in: UpdateSchemaType, db: Session = Depends(self.get_db)):         db_item = self.crud.get(db=db, id=item_id)         if not db_item:             raise HTTPException(status_code=404, detail="Item not found")         return self.crud.update(db=db, db_obj=db_item, obj_in=item_in)      async def delete_item(self, item_id: int, db: Session = Depends(self.get_db)):         db_item = self.crud.get(db=db, id=item_id)         if not db_item:             raise HTTPException(status_code=404, detail="Item not found")         return self.crud.remove(db=db, id=item_id)      async def read_items(self, skip: int = 0, limit: int = 10, db: Session = Depends(self.get_db)):         items = self.crud.get_multi(db=db, skip=skip, limit=limit)         return items

使用通用API基类定义用户端点(api/v1/endpoints/users.py)

from fastapi import APIRouter from app.crud.user import user as user_crud from app.schemas.user import User, UserCreate, UserUpdate from app.api.deps import CRUDRouter  user_router = CRUDRouter[User, UserCreate, UserUpdate](user_crud) router = user_router.router

注册路由 (main.py)

rom fastapi import FastAPI from app.api.v1.endpoints import users  app = FastAPI()  app.include_router(users.router, prefix="/api/v1/users", tags=["users"])  if __name__ == "__main__":     import uvicorn     uvicorn.run(app, host="0.0.0.0", port=8000)

通过这种方式，你可以在 CRUDRouter 基类中封装常规的CRUD操作方法，然后在具体的端点文件中继承这个基类并传递相应的CRUD对象。这样可以进一步减少重复代码，提高代码的可维护性和可复用性。

 

4、SQLAlchemy模型的基类定义

app.db.base_class 通常是用于定义SQLAlchemy模型基类的文件。在这个文件中，我们会定义一个基本的Base类，这个类是所有SQLAlchemy模型的基类。下面是一个实现示例：

定义 Base 类 (db/base_class.py)

from sqlalchemy.ext.declarative import as_declarative, declared_attr  @as_declarative() class Base:     id: int     __name__: str      @declared_attr     def __tablename__(cls) -> str:         return cls.__name__.lower()

详细解释

1. @as_declarative(): 这是SQLAlchemy提供的一个装饰器，它会将类装饰为一个声明性基类。所有继承自这个类的子类都会自动成为声明性类。

2. id: int: 这是一个类型注释，表示每个模型类都会有一个 id 属性。具体的字段定义（例如 Column(Integer, primary_key=True)）会在每个具体的模型类中定义。

3. __name__: str: 这是另一个类型注释，表示每个模型类都会有一个 __name__ 属性。

4. @declared_attr: 这是SQLAlchemy提供的一个装饰器，允许我们为声明性基类定义一些通用的属性。在这个例子中，它用于自动生成 __tablename__ 属性。这个属性的值是模型类的名称的小写形式。

这样定义的 Base 类可以作为所有SQLAlchemy模型的基类，简化模型的定义。

完整示例项目结构：为了更好地理解，这里展示一个包含 Base 类定义的完整项目结构：

. ├── app │   ├── __init__.py │   ├── api │   │   ├── __init__.py │   │   └── v1 │   │       ├── __init__.py │   │       └── endpoints │   │           ├── __init__.py │   │           └── users.py │   ├── crud │   │   ├── __init__.py │   │   ├── base.py │   │   └── user.py │   ├── db │   │   ├── __init__.py │   │   ├── base.py │   │   ├── base_class.py │   │   └── session.py │   ├── models │   │   ├── __init__.py │   │   └── user.py │   ├── schemas │   │   ├── __init__.py │   │   └── user.py │   └── main.py

models/user.py 类文件如下定义

from sqlalchemy import Column, Integer, String from app.db.base_class import Base  class User(Base):     __tablename__ = "users"      id = Column(Integer, primary_key=True, index=True)     email = Column(String, unique=True, index=True, nullable=False)     hashed_password = Column(String, nullable=False)     full_name = Column(String, index=True)

通过这种结构和定义，您可以创建一个简洁、可扩展的FastAPI项目，能够快速定义新的数据库模型并生成相应的CRUD操作和API端点。