Using FastAPI to develop projects, some references on how to plan the directory structure of the project and some handling of base class encapsulation.

When developing projects with FastAPI, a good directory structure can help you better organize your code and improve maintainability and extensibility. Similarly, encapsulation of base classes can further reduce development code, provide convenience, and reduce the chance of errors.

Below is an example of a recommended directory structure:

my_fastapi_project/
├── app/
│ ├── __init__.py
│ ├──── # entry file
│ ├── core/
│ │ ├── __init__.py
│ │ ├── # Configuration File
│ │ ├── # Security-related
│ │ └── ...            # Other core functions
│ ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│ │ │ │ ├── __init__.py
│ │ │ │ ├─── user-related interface
│ │ │ │ ├── # Other interfaces
│ │ │ │ └── ...
│ │ │ └── ...              # alternativeAPI
│ ├── models/
│ │ ├── __init__.py
│ │ ├─── user model # user model
│ │ ├── # Other models
│ │ └── ...
│ ├─── schemas/
│ │ ├── expense or outlay__init__.py
│ │ ├─── # expense or outlay户数据模型
│ │ ├── # Other data models
│ │ └── ...
│ ├─── crud/
│ │ ├── subscriberscrudmanipulate─ __init__.py
│ │ ├──── subscribersCRUDmanipulate
│ │ ├── # (sth. or sb) elseCRUDmanipulate
│ │ └── ...
│ ├─── db/
│ │ ├── __init__.py
│ │ ├─── # Database base setup
│ │ ├── # database session
│ │ └── ...
│ ├── tests/
│ │ ├── __init__.py
│ │ ├── test_main.py # Test Master File
│ │ ├── test_users.py # User-related testing
│ │ └── ...
│ └─── utils/
│ ├── __init__.py
│ ├─── # instrumented function
│ └── ...
├── .env                    # Environment variable files
├── alembic/                # Database Migration Tools Catalog
│ ├──env.py
│   ├── 
│   └── versions/
│ └── ...
├── # Alembic configuration files
├── # Project dependencies
├── Dockerfile # Docker configuration files
└── # Project description file

Catalog structure description:

app/: The main directory of the project, containing all application-related code.
- : The project's entry file to launch the FastAPI application.
- core/:: Core functions such as configuration, security, etc.
- api/: API routes and views, managed in versions.
- models/:: Database modeling.
- schemas/: Data model for request and response validation.
- crud/:: Database operations (CRUD: create, read, update, delete).
- db/:: Database-related settings and session management.
- tests/:: Test code.
- utils/:: Tool functions and utility modules.
.env: Environment variable file for storing sensitive information such as database connection strings.
alembic/: The configuration directory for the database migration tool Alembic.
: A list of project dependencies.
Dockerfile: Docker configuration file for containerized deployment.
:: Project description documents.

This structure can be adapted to meet project needs, but it is good practice to keep it clear and modular.

python project total __init__.py, does that make sense?？

In the Python project, the__init__.py The main purpose of the file is to identify the catalog as a Python package. It makes the modules in the directory available for import and use. In some cases, the__init__.py can be more than just an empty file and can contain some initialization code.

__init__.py The significance of the

Identifies the catalog as a package:
- Anyone who contains__init__.py directories are considered a package by the Python interpreter so that you can use package import syntax such asimport 。
Initialization code:

This can be done in the__init__.py contains some initialization code, such as importing submodules within a package, setting package-level variables or functions, configuring logging, and so on. Example:

# mypackage/__init__.py
from .submodule1 import func1
from .submodule2 import func2

__all__ = ["func1", "func2"]

3. Simplified import

- 　This is accomplished through the use of the__init__.py The import of submodules in a package can simplify the package import path, allowing users to import functions or classes directly from the package without having to know the specific module structure.

# mypackage/__init__.py
from .submodule import MyClass

# Now you can do
from mypackage import MyClass

For Python 3.3 and above, the__init__.py Documentation is not mandatory, even without__init__.py file, the Python interpreter can also recognize packages. However, adding the__init__.py file is still a good habit to avoid unintended behavior in some cases, and makes it clear that the directory is a package.

2. Fast API project development process

In the FastAPI project, CRUD operations are usually performed in a specializedcrud module is implemented. This module calls the SQLAlchemy model object for database operations.

1. Defining the model (models/)

from sqlalchemy import Column, Integer, String
from .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. Create a database session (db/)

from sqlalchemy import create_engine
from  import sessionmaker

DATABASE_URL = "sqlite:///./"  # Using a SQLite database as an example

engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

3. Define CRUD operations (crud/)

from  import Session
from  import User
from  import UserCreate, UserUpdate

def get_user(db: Session, user_id: int):
    return (User).filter( == user_id).first()

def get_user_by_email(db: Session, email: str):
    return (User).filter( == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 10):
    return (User).offset(skip).limit(limit).all()

def create_user(db: Session, user: UserCreate):
    db_user = User(
        email=,
        hashed_password=user.hashed_password,  # Passwords should be hashed in practical applications
        full_name=user.full_name
    )
    (db_user)
    ()
    (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 = 
        db_user.full_name = user.full_name
        ()
        (db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    db_user = get_user(db, user_id)
    if db_user:
        (db_user)
        ()
    return db_user

4. Define the data model (schemas/)

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. Using CRUD operations in API endpoints (api/v1/endpoints/)

from fastapi import APIRouter, Depends, HTTPException
from  import Session
from app import crud, models, schemas
from  import SessionLocal

router = APIRouter()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        ()

@("/users/", response_model=)
def create_user(user: , db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

@("/users/{user_id}", response_model=)
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

@("/users/{user_id}", response_model=)
def update_user(user_id: int, user: , 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

@("/users/{user_id}", response_model=)
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

@("/users/", response_model=List[])
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. Registration of routes ()

from fastapi import FastAPI
from . import users

app = FastAPI()

app.include_router(, prefix="/api/v1", tags=["users"])

if __name__ == "__main__":
    import uvicorn
    (app, host="0.0.0.0", port=8000)

7. Initialization of the database (db/)

from  import engine
from  import user

.create_all(bind=engine)

8. Running the application

In the project root directory run.

uvicorn :app --reload

This way, your CRUD layer can call the model object to perform database operations. The above code shows how you can define the model, database session, CRUD operations, data model and API endpoints and combine them together to implement a simple user management system.

3. The actual FastAPI project's encapsulation of the base class

It is possible to encapsulate regular CRUD operations by creating a generic CRUD base class and then have specific CRUD classes inherit from this base class. This reduces duplicate code and improves maintainability and reusability. The following is an example implementation.

1. Create a generic CRUD base class (crud/)

rom typing import Generic, Type, TypeVar, Optional, List
from pydantic import BaseModel
from  import Session
from .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]):
         = model

    def get(self, db: Session, id: int) -> Optional[ModelType]:
        return ().filter( == id).first()

    def get_multi(self, db: Session, skip: int = 0, limit: int = 100) -> List[ModelType]:
        return ().offset(skip).limit(limit).all()

    def create(self, db: Session, obj_in: CreateSchemaType) -> ModelType:
        obj_in_data = obj_in.dict()
        db_obj = (**obj_in_data)  # type: ignore
        (db_obj)
        ()
        (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_obj)
        return db_obj

    def remove(self, db: Session, id: int) -> ModelType:
        obj = ().get(id)
        (obj)
        ()
        return obj

2. Define user CRUD operations (crud/)

from typing import Any
from  import Session
from  import CRUDBase
from  import User
from  import UserCreate, UserUpdate

class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
    def get_by_email(self, db: Session, email: str) -> Any:
        return ().filter( == email).first()

user = CRUDUser(User)

3. Define the data model (schemas/)

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. Use CRUD operations in API endpoints (api/v1/endpoints/)

from fastapi import APIRouter, Depends, HTTPException
from  import Session
from typing import List
from app import crud, schemas
from  import SessionLocal
from  import User

router = APIRouter()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        ()

@("/users/", response_model=)
def create_user(user: , db: Session = Depends(get_db)):
    db_user = .get_by_email(db, email=)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return (db=db, obj_in=user)

@("/users/{user_id}", response_model=)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = (db, id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@("/users/{user_id}", response_model=)
def update_user(user_id: int, user: , db: Session = Depends(get_db)):
    db_user = (db=db, id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return (db=db, db_obj=db_user, obj_in=user)

@("/users/{user_id}", response_model=)
def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = (db=db, id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return (db=db, id=user_id)

@("/users/", response_model=List[])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    users = .get_multi(db, skip=skip, limit=limit)
    return users

The rest is a similar approach to the previous one.

In this way, you can encapsulate regular CRUD operations in generic CRUD base classes, while specific CRUD classes (such as theCRUDUser) only need to inherit the base class and add specific operation methods. This not only reduces duplicate code, but also improves the maintainability and reusability of the code.

If you wish you can encapsulate regular CRUD operation methods by defining a generic API base class and then inherit this base class in the specific endpoint file. This will further reduce duplicate code and improve code maintainability and reusability.

Create a generic API base class (api/)

from typing import Type, TypeVar, Generic, List
from fastapi import APIRouter, Depends, HTTPException
from  import Session
from pydantic import BaseModel
from  import CRUDBase
from  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]):
         = crud
         = APIRouter()

        ("/", response_model=ModelType)(self.create_item)
        ("/{item_id}", response_model=ModelType)(self.read_item)
        ("/{item_id}", response_model=ModelType)(self.update_item)
        ("/{item_id}", response_model=ModelType)(self.delete_item)
        ("/", response_model=List[ModelType])(self.read_items)

    def get_db(self):
        db = SessionLocal()
        try:
            yield db
        finally:
            ()

    async def create_item(self, item_in: CreateSchemaType, db: Session = Depends(self.get_db)):
        db_item = (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 = (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 = (db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return (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 = (db=db, id=item_id)
        if not db_item:
            raise HTTPException(status_code=404, detail="Item not found")
        return (db=db, id=item_id)

    async def read_items(self, skip: int = 0, limit: int = 10, db: Session = Depends(self.get_db)):
        items = .get_multi(db=db, skip=skip, limit=limit)
        return items

Define user endpoints using the generic API base class (api/v1/endpoints/)

from fastapi import APIRouter
from  import user as user_crud
from  import User, UserCreate, UserUpdate
from  import CRUDRouter

user_router = CRUDRouter[User, UserCreate, UserUpdate](user_crud)
router = user_router.router

Registering a Route ()

rom fastapi import FastAPI
from . import users

app = FastAPI()

app.include_router(, prefix="/api/v1/users", tags=["users"])

if __name__ == "__main__":
    import uvicorn
    (app, host="0.0.0.0", port=8000)

By doing this, you can make theCRUDRouter The base class encapsulates the regular CRUD operation methods, and then inherits this base class and passes the corresponding CRUD object in the specific endpoint file. This can further reduce duplication of code and improve the maintainability and reusability of the code.

4. SQLAlchemy model base class definition

.base_class Usually a file used to define the base class of a SQLAlchemy model. In this file, we will define a basic Base class which is the base class for all SQLAlchemy models. Below is an example implementation:

defineBase Class (db/base_class.py)

from  import as_declarative, declared_attr

@as_declarative()
class Base:
    id: int
    __name__: str

    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

elaborate

@as_declarative(): This is a decorator provided by SQLAlchemy that will decorate the class as a declarative base class. All subclasses inheriting from this class will automatically become declarative.
id: int: This is a type annotation indicating that each model class will have aid Attributes. Specific field definitions (e.g.Column(Integer, primary_key=True)) will be defined in each specific model class.
__name__: str: This is another type annotation indicating that each model class will have a__name__ Properties.
@declared_attr: This is a decorator provided by SQLAlchemy that allows us to define some generic properties for declarative base classes. In this example, it is used to automatically generate__tablename__ Attribute. The value of this attribute is the lowercase form of the name of the model class.

definedBase class can be used as a base class for all SQLAlchemy models, simplifying model definition.

Full sample project structure: for better understanding, here is a sample project containing theBase The complete project structure of the class definition:

.
├── app
│   ├── __init__.py
│   ├── api
│   │   ├── __init__.py
│   │   └── v1
│   │       ├── __init__.py
│   │       └── endpoints
│   │           ├── __init__.py
│   │           └── 
│   ├── crud
│   │   ├── __init__.py
│   │   ├── 
│   │   └── 
│   ├── db
│   │   ├── __init__.py
│   │   ├── 
│   │   ├── base_class.py
│   │   └── 
│   ├── models
│   │   ├── __init__.py
│   │   └── 
│   ├── schemas
│   │   ├── __init__.py
│   │   └── 
│   └──

The models/ class file is defined as follows

from sqlalchemy import Column, Integer, String
from .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)

With this structure and definition, you can create a clean, extensible FastAPI project capable of quickly defining new database models and generating the corresponding CRUD operations and API endpoints.