使用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 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()
详细解释
-
@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 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端点。
标签:__,封装,FastAPI,py,db,user,基类,import,id From: https://www.cnblogs.com/wuhuacong/p/18380808