使用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
的意义：
将目录标识为包：
- 任何包含
__init__.py
的目录都会被Python解释器认为是一个包，这样你就可以使用包导入语法，如
import mypackage.module
。
初始化代码：
- 可以在
__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 candofrom 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 importColumn, Integer, Stringfrom app.db.base_class importBaseclassUser(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 importcreate_enginefrom sqlalchemy.orm importsessionmaker
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 importSessionfrom app.models.user importUserfrom app.schemas.user importUserCreate, UserUpdatedefget_user(db: Session, user_id: int):return db.query(User).filter(User.id ==user_id).first()defget_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):returndb.query(User).offset(skip).limit(limit).all()defcreate_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)returndb_userdefupdate_user(db: Session, user_id: int, user: UserUpdate):
db_user=get_user(db, user_id)ifdb_user:
db_user.email=user.email
db_user.full_name=user.full_name
db.commit()
db.refresh(db_user)returndb_userdefdelete_user(db: Session, user_id: int):
db_user=get_user(db, user_id)ifdb_user:
db.delete(db_user)
db.commit()return db_user
4. 定义数据模型 (
schemas/user.py
)
from pydantic importBaseModelclassUserBase(BaseModel):
email: str
full_name: str=NoneclassUserCreate(UserBase):
hashed_password: strclassUserUpdate(UserBase):pass
classUser(UserBase):
id: intclassConfig:
orm_mode= True
5. 在API端点中使用CRUD操作 (
api/v1/endpoints/users.py
)
from fastapi importAPIRouter, Depends, HTTPExceptionfrom sqlalchemy.orm importSessionfrom app importcrud, models, schemasfrom app.db.session importSessionLocal
router=APIRouter()defget_db():
db=SessionLocal()try:yielddbfinally:
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)ifdb_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 isNone:raise HTTPException(status_code=404, detail="User not found")returndb_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 isNone:raise HTTPException(status_code=404, detail="User not found")returndb_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 isNone:raise HTTPException(status_code=404, detail="User not found")returndb_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 importFastAPIfrom app.api.v1.endpoints importusers
app=FastAPI()
app.include_router(users.router, prefix="/api/v1", tags=["users"])if __name__ == "__main__":importuvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
7. 初始化数据库 (
db/base.py
)
from app.db.session importenginefrom app.models importuser
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 importGeneric, Type, TypeVar, Optional, Listfrom pydantic importBaseModelfrom sqlalchemy.orm importSessionfrom app.db.base_class importBase
ModelType= TypeVar("ModelType", bound=Base)
CreateSchemaType= TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType= TypeVar("UpdateSchemaType", bound=BaseModel)classCRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):def __init__(self, model: Type[ModelType]):
self.model=modeldef 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]:returndb.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)returndb_objdef 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 inobj_data:if field inupdate_data:
setattr(db_obj, field, update_data[field])
db.commit()
db.refresh(db_obj)returndb_objdef 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 importAnyfrom sqlalchemy.orm importSessionfrom app.crud.base importCRUDBasefrom app.models.user importUserfrom app.schemas.user importUserCreate, UserUpdateclassCRUDUser(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 importBaseModelclassUserBase(BaseModel):
email: str
full_name: str=NoneclassUserCreate(UserBase):
hashed_password: strclassUserUpdate(UserBase):pass
classUser(UserBase):
id: intclassConfig:
orm_mode= True
4、在API端点中使用CRUD操作 (
api/v1/endpoints/users.py
)
from fastapi importAPIRouter, Depends, HTTPExceptionfrom sqlalchemy.orm importSessionfrom typing importListfrom app importcrud, schemasfrom app.db.session importSessionLocalfrom app.models.user importUser
router=APIRouter()defget_db():
db=SessionLocal()try:yielddbfinally:
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)ifdb_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 isNone:raise HTTPException(status_code=404, detail="User not found")returndb_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 isNone: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 isNone: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 importType, TypeVar, Generic, Listfrom fastapi importAPIRouter, Depends, HTTPExceptionfrom sqlalchemy.orm importSessionfrom pydantic importBaseModelfrom app.crud.base importCRUDBasefrom app.db.session importSessionLocal
ModelType= TypeVar("ModelType", bound=BaseModel)
CreateSchemaType= TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType= TypeVar("UpdateSchemaType", bound=BaseModel)classCRUDRouter(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)defget_db(self):
db=SessionLocal()try:yielddbfinally:
db.close()
asyncdef create_item(self, item_in: CreateSchemaType, db: Session =Depends(self.get_db)):
db_item= self.crud.create(db=db, obj_in=item_in)returndb_item
asyncdef read_item(self, item_id: int, db: Session =Depends(self.get_db)):
db_item= self.crud.get(db=db, id=item_id)if notdb_item:raise HTTPException(status_code=404, detail="Item not found")returndb_item
asyncdef 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 notdb_item:raise HTTPException(status_code=404, detail="Item not found")return self.crud.update(db=db, db_obj=db_item, obj_in=item_in)
asyncdef delete_item(self, item_id: int, db: Session =Depends(self.get_db)):
db_item= self.crud.get(db=db, id=item_id)if notdb_item:raise HTTPException(status_code=404, detail="Item not found")return self.crud.remove(db=db, id=item_id)
asyncdef 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 importAPIRouterfrom app.crud.user importuser as user_crudfrom app.schemas.user importUser, UserCreate, UserUpdatefrom app.api.deps importCRUDRouter
user_router=CRUDRouter[User, UserCreate, UserUpdate](user_crud)
router= user_router.router
注册路由 (
main.py
)
rom fastapi importFastAPIfrom app.api.v1.endpoints importusers
app=FastAPI()
app.include_router(users.router, prefix="/api/v1/users", tags=["users"])if __name__ == "__main__":importuvicorn
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 importas_declarative, declared_attr
@as_declarative()classBase:
id: int__name__: str
@declared_attrdef __tablename__(cls) ->str:return cls.__name__.lower()
详细解释
@as_declarative()
: 这是SQLAlchemy提供的一个装饰器，它会将类装饰为一个声明性基类。所有继承自这个类的子类都会自动成为声明性类。
id: int
: 这是一个类型注释，表示每个模型类都会有一个
id
属性。具体的字段定义（例如
Column(Integer, primary_key=True)
）会在每个具体的模型类中定义。
__name__: str
: 这是另一个类型注释，表示每个模型类都会有一个
__name__
属性。
@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 importColumn, Integer, Stringfrom app.db.base_class importBaseclassUser(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端点。
