Creating an AutoCRUD in FastAPI + SQLAlchemy

▶

An AutoCRUD would be one of the most interesting and obvious operations we could use, as we have seen that everything is modular—just like we did before, in a separate file, handling our own WebSockets with FastAPI, using dependency injection and class definitions directly in the routes. Therefore, this would be the next logical step.

There is a package called FastAPI CRUD Router that promises exactly that: automatically creating routes to delete, update, get, create, etc. Once installed, you can configure it with Pydantic classes (dependencies), register it, and you would have a functional AutoCRUD.

The problem is that this package is quite old and works with Pydantic v1:

https://github.com/awtkns/fastapi-crudrouter

Which nowadays represents an inconvenience. Personally, this is what I like least about microframeworks (and FastAPI can be considered as such): they bring the minimum necessary to function and, from there, you must orchestrate everything (as we saw with clean architecture). Something similar happens in Flask.

That’s why, before installing any package that seems to solve something “automatically”, check its GitHub repository. If you see that it hasn't had updates for years, run away. At some point (if not already), it will fail you.

So, what do we do?

️ Creating a Generic AutoCRUD

Since external packages are outdated, the best thing is to create our own AutoCRUD. It is a simple process if we understand how generics work in Python.

We are going to create our own AutoCRUD, as it really is quite simple.

I’m going to do everything in a single file for simplicity. In a real project, you can structure it however you prefer. You have the complete code in the repository associated with the full course and book that you can search for in this publication.

Defining Base Models

First, we create a Pydantic BaseModel. Something simple:

• Model for Category
• Model for Task

Both inherit from BaseModel, as they simply model data. We don't need complex inheritance or additional logic.

mycrud.py

from pydantic import BaseModel, Field
class CategoryBase(BaseModel):
    name: str
class Category(CategoryBase):
    id: int = Field(..., ge=1) # Ensure id is greater than or equal to 1
    class Config:
        from_attributes = True

Using Generics for a Reusable CRUD

Here is where it gets interesting.

We are going to create an AutoCRUD class that receives:

• The schema
• The prefix
• The tags
• Other necessary arguments

# We define a generic type for our schemas
T = TypeVar("T", bound=BaseModel)
class MyCRUDRouter(Generic[T], APIRouter):
    def __init__(self, schema: Type[T], *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.schema = schema
        self.db: List[T] = []

Since we will work with different models (Category, Task, etc.), we need our class to be generic.

For that we use TypeVar, where:

• The base type will be BaseModel
• The concrete type can be any model that inherits from it

This allows us to create a robust, typed, and reusable CRUD.

Multiple Inheritance and Method Resolution Order (MRO)

Our class will inherit from:

• Generic
• APIRouter

This is where the Method Resolution Order (MRO) comes into play.

When we call super().__init__(), Python must decide which constructor to execute first. The order depends on how we declare the inheritance.

In this case:

class MyCRUDRouter(Generic[T], APIRouter):

Python will look first in Generic, then in APIRouter.

Since Generic does not have a relevant constructor, it will end up using the one from APIRouter, which is where we are actually interested in parameters such as:

• prefix
• tags

That is why we pass them using *args and **kwargs.

Why are the routes inside the init?

Here comes the "weird" part.

Functions such as:

• @self.post
• @self.get
• @self.put
• @self.delete

are defined inside the __init__.

T = TypeVar("T", bound=BaseModel)
class MyCRUDRouter(Generic[T], APIRouter):
    def __init__(self, schema: Type[T], *args, **kwargs):
        ***
        @self.post("/", response_model=self.schema, status_code=status.HTTP_201_CREATED)
        async def create(item: self.schema):
            ***
        @self.get("/", response_model=List[self.schema])
        async def get_all():
            ***
        @self.get("/{item_id}", response_model=self.schema)
        async def get_one(item_id: int):
            ***
        @self.put("/{item_id}", response_model=self.schema)
        async def update(item_id: int, updated_data: self.schema):
            ***
        # --- DELETE ---
        @self.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
           ***
    def _find_item(self, item_id: int) -> Optional[T]:
        ***

Why?

Because decorators are executed when Python reads the class, not when you create the instance. If we tried to use self.schema or self.prefix outside the constructor, at that moment they wouldn't exist yet, and the application would fail.

Therefore, the routes must be declared inside the __init__, where we already have access to:

• self.schema
• self.db
• and other configurations

FastAPI decorators (like @self.post) are executed when Python reads the class. If we put them outside the constructor, they would try to access the schema (self.schema) before it is assigned when creating the object, causing an error.

If we move it outside, everything simply explodes.

Mock Implementation (In Memory)

First, we make a simple version using a list as an in-memory database:

• create
• get_all
• get_one
• update
• delete

Everything works using the generic type T.

This allows us to quickly test the functionality without a real database.

The code looks like this:

from pydantic import BaseModel, Field
from typing import List, Type, TypeVar, Generic, Optional
from fastapi import APIRouter, HTTPException, status, Depends
from typing import Generic, TypeVar, Type, List
from pydantic import BaseModel
from sqlalchemy.orm import Session
from sqlalchemy import select
T = TypeVar("T", bound=BaseModel)
class MyCRUDRouter(Generic[T], APIRouter):
    def __init__(self, schema: Type[T], *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.schema = schema
        self.db: List[T] = []
        # --- CREATE ---
        @self.post("/", response_model=self.schema, status_code=status.HTTP_201_CREATED)
        async def create(item: self.schema):
            self.db.append(item)
            return item
        # --- READ ALL ---
        @self.get("/", response_model=List[self.schema])
        async def get_all():
            return self.db
        # --- READ ONE ---
        @self.get("/{item_id}", response_model=self.schema)
        async def get_one(item_id: int):
            item = self._find_item(item_id)
            if not item:
                raise HTTPException(status_code=404, detail="Item not found")
            return item
        # --- UPDATE ---
        @self.put("/{item_id}", response_model=self.schema)
        async def update(item_id: int, updated_data: self.schema):
            for index, item in enumerate(self.db):
                if getattr(item, 'id', None) == item_id:
                    # In Pydantic V2 we use model_dump to update
                    self.db[index] = updated_data
                    return updated_data
            raise HTTPException(status_code=404, detail="Could not update: not found")
        # --- DELETE ---
        @self.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
        async def delete(item_id: int):
            for index, item in enumerate(self.db):
                if getattr(item, 'id', None) == item_id:
                    self.db.pop(index)
                    return
            raise HTTPException(status_code=404, detail="Could not delete: not found")
    def _find_item(self, item_id: int) -> Optional[T]:
        """Helper method to find by ID."""
        return next((item for item in self.db if getattr(item, 'id', None) == item_id), None)

Database Version (SQLAlchemy)

The previous example works in memory, but for a real project, we must integrate it with SQLAlchemy. The only thing that really changes is the logic inside the route functions:

• Conversion: We transform the Pydantic model into a SQLAlchemy model.
• ID Exclusion: When creating (POST), we must exclude the id field so that the database generates it automatically as autoincremental.
• Operations: We replace list handling with db.add(), db.commit(), and db.refresh().

Then we replace the in-memory list with:

• SQLAlchemy Model
• Database Session
• Real operations with commit

Here the only thing that changes is:

• Convert the Pydantic model to a SQLAlchemy model
• Save it in the database
• Correctly handle the autoincremental ID

We had to exclude the id in the creation schema, so that the database generates it internally and not us:

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, relationship
# 1. Define the Base Class (Recommended in SQLAlchemy 2.0)
class Base(DeclarativeBase):
    pass
# 2. The Entity Model
class CategoryModel(Base):
    __tablename__ = "categories"
    # We define the columns
    id=Column(Integer, primary_key=True, index=True)
    name = Column(String(255), nullable=False, unique=True)
    # Optional: If you want a category to have many tasks
    # tasks = relationship("TaskModel", back_populates="category")
    def __repr__(self):
        return f"<Category(id={self.id}, name='{self.name}')>"        
        
class SQLCRUDRouter(Generic[T, M], APIRouter):
    def __init__(self, schema: Type[T], model: Type[M], get_db_func, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.schema = schema
        self.model = model
        self.get_db = get_db_func
        # --- CREATE ---
        @self.post("/", response_model=self.schema, status_code=status.HTTP_201_CREATED)
        def create(item: self.schema, db: Session = Depends(self.get_db)):
            # We convert Pydantic to SQLAlchemy Model
            
            # 2. We create the instance of the SQLAlchemy model with clean data
            # db_item = self.model(**item.model_dump()) # must remove the id
            db_item = self.model(**item.model_dump(exclude={'id'}, exclude_unset=True))
    
            
          
            db.add(db_item)
            db.commit()
            db.refresh(db_item)
            return db_item
        # --- READ ALL ---
        @self.get("/", response_model=List[self.schema])
        def get_all(db: Session = Depends(self.get_db)):
            # We use the new SQLAlchemy 2.0 syntax (select)
            result = db.execute(select(self.model)).scalars().all()
            return result
        # --- READ ONE ---
        @self.get("/{item_id}", response_model=self.schema)
        def get_one(item_id: int, db: Session = Depends(self.get_db)):
            db_item = db.get(self.model, item_id)
            if not db_item:
                raise HTTPException(status_code=404, detail="Not found")
            return db_item
        # --- DELETE ---
        @self.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
        def delete(item_id: int, db: Session = Depends(self.get_db)):
            db_item = db.get(self.model, item_id)
            if not db_item:
                raise HTTPException(status_code=404, detail="Not found")
            db.delete(db_item)
            db.commit()
            return
        # --- UPDATE ---
        @self.put("/{item_id}", response_model=self.schema)
        def update(item_id: int, updated_data: self.schema, db: Session = Depends(self.get_db)):
            # 1. Search for the existing record
            db_item = db.get(self.model, item_id)
            if not db_item:
                raise HTTPException(status_code=404, detail="Could not update: not found")
            # 2. Extract Pydantic data
            # We exclude 'id' to prevent them from trying to change the row's PK
            # exclude_unset=True allows partial updates if the schema supports it
            update_dict = updated_data.model_dump(exclude={'id'}, exclude_unset=True)
            # 3. Update model attributes dynamically
            for key, value in update_dict.items():
                setattr(db_item, key, value)
            # 4. Persist
            db.commit()
            db.refresh(db_item)
            return db_item

Using the AutoCRUD

Once the class is created, we simply do:

# app.include_router(
#     MyCRUDRouter(schema=Category, prefix="/categories", tags=["Categories"])
# )
app.include_router(
    MyCRUDRouter(schema=Task, prefix="/tasks", tags=["Tasks"])
)
app.include_router(
    SQLCRUDRouter(schema=Category, model=CategoryModel,get_db_func=get_database_session, prefix="/categories", tags=["Categories"])
)

And automatically we have:

• POST
• GET
• GET by ID
• PUT
• DELETE

Fully functional.

Conclusion

Creating an AutoCRUD in FastAPI is much simpler than it seems. We don't need outdated packages or to depend on third parties.

With this, you will automatically obtain:

• Swagger Documentation: Routes generated for each entity with their respective models.
• Validation: Strong typing thanks to Python generics.
• Functionality: A complete CRUD (Create, Read, Update, Delete) working with just a couple of lines of code per entity.

With:

• Generics
• Multiple Inheritance
• APIRouter
• Pydantic
• SQLAlchemy

We can build a robust, typed, and reusable solution and most importantly: really understanding how it works on the inside.