First Steps

Install SQLModel

First you need to install sqlmodel and dependencies:

pip install sqlmodel psycopg2 alembic

A basic CRUD app

Consider a basic create-read-update-delete (CRUD) app where can create “Book” & “Category” instances.

import uuid
from typing import List

import uvicorn
from fastapi import FastAPI, Depends
from sqlalchemy import select, delete
from sqlmodel import Session, create_engine

from database import run_async_upgrade
from models.models import Category, CategoryPUT, Book, BookPUT

app = FastAPI(debug=True)

engine = create_engine('postgresql://postgres:123456@localhost:5432/fastapi_example', echo=True)


@app.on_event("startup")
def on_startup():
    print("Start migration")
    run_async_upgrade()
    print("DB success upgrade !")


def get_session() -> Session:
    with Session(engine) as conn:
        yield conn


# region Categories
@app.get("/references/category")
def get_list_categories(conn: Session = Depends(get_session)) -> List[Category]:
    items = conn.execute(select(Category)).scalars().all()
    return items


@app.post("/references/category")
def add_category(model: Category, conn: Session = Depends(get_session)) -> Category:
    conn.add(model)
    conn.commit()
    return model


@app.delete("/references/category/{guid}")
def delete_category(guid: str, conn: Session = Depends(get_session)) -> bool:
    conn.execute(
        delete(Category).filter(Category.guid == uuid.UUID(guid))
    )
    conn.commit()
    return True


@app.put("/references/category/{guid}")
def update_category(guid: str, model: CategoryPUT, conn: Session = Depends(get_session)) -> Category:
    model_db = conn.execute(
        select(Category).filter(Category.guid == uuid.UUID(guid))
    ).scalar()
    # Update fields
    for name, val in model.dict(exclude_unset=True).items():
        setattr(model_db, name, val)
    conn.commit()
    conn.refresh(model_db)
    return model_db


# endregion

# region Books
@app.get("/references/books")
def get_list_books(conn: Session = Depends(get_session)) -> List[Book]:
    items = conn.execute(select(Book)).scalars().all()
    return items


@app.post("/references/books")
def add_book(model: Book, conn: Session = Depends(get_session)) -> Book:
    conn.add(model)
    conn.commit()
    return model


@app.delete("/references/books/{guid}")
def delete_book(guid: str, conn: Session = Depends(get_session)) -> bool:
    conn.execute(
        delete(Book).filter(Book.guid == uuid.UUID(guid))
    )
    conn.commit()
    return True


@app.put("/references/book/{guid}")
def update_book(guid: str, model: BookPUT, conn: Session = Depends(get_session)) -> Book:
    model_db = conn.execute(
        select(Book).filter(Book.guid == uuid.UUID(guid))
    ).scalar()
    # Update fields
    for name, val in model.dict(exclude_unset=True).items():
        setattr(model_db, name, val)
    conn.commit()
    conn.refresh(model_db)
    return model_db


# endregion
if __name__ == "__main__":
    uvicorn.run('main:app', host="localhost", port=8001, reload=True, debug=True)

If you look at the highlighted lines above, you can see that construction:

conn: Session = Depends(get_session)

has been repeated 8 times and there are a lot of duplicates in the code.

The OpenAPI specification looks like this:

Class based API

By using the Routable class, we can consolidate the endpoint signatures and reduce the number of repeated dependencies.

To use the Routable class, you need to do the following:

• #0: Import the Rotable class and decorators (get, post, put, delete) for the endpoints.

• #1: Creating your class (controller) and inheriting from Routable.

• #2: Just list all your dependencies as class fields.
• #3: Add the keyword self to your endpoints. The first parameter of methods is the instance the method is called on.
• #4: Include all your controllers (endpoints) in the router.

import uuid
from typing import List

import sqlalchemy
import uvicorn
from class_based_fastapi import Routable, get, put, post, delete  # 0. Import
from fastapi import FastAPI, Depends
from sqlalchemy import select
from sqlmodel import Session, create_engine

from database import run_async_upgrade
from models.models import Category, CategoryPUT, Book, BookPUT

app = FastAPI(debug=True)

engine = create_engine('postgresql://postgres:123456@localhost:5432/fastapi_example', echo=True)


@app.on_event("startup")
def on_startup():
    print("Start migration")
    run_async_upgrade()
    print("DB success upgrade !")


def get_session() -> Session:
    with Session(engine) as conn:
        yield conn


# region Categories
class CategoryAPI(Routable):  # 1. Create class
    NAME_MODULE = Category.__name__
    conn: Session = Depends(get_session)  # 2. Add depend

    @get("")
    def get_list_categories(self) -> List[Category]:  # 3. Add `self` param and remove conn
        items = self.conn.execute(select(Category)).scalars().all()
        return items

    @post("")
    def add_category(self, model: Category) -> Category:  # 3. Add `self` param and remove conn
        self.conn.add(model)
        self.conn.commit()
        return model

    @delete("{guid}")
    def delete_category(self, guid: str) -> bool:  # 3. Add `self` param and remove conn
        self.conn.execute(
            sqlalchemy.delete(Category).filter(Category.guid == uuid.UUID(guid))
        )
        self.conn.commit()
        return True

    @put("{guid}")
    def update_category(self, guid: str, model: CategoryPUT) -> Category:  # 3. Add `self` param and remove conn
        model_db = self.conn.execute(
            select(Category).filter(Category.guid == uuid.UUID(guid))
        ).scalar()
        # Update fields
        for name, val in model.dict(exclude_unset=True).items():
            setattr(model_db, name, val)
        self.conn.commit()
        self.conn.refresh(model_db)
        return model_db


# endregion

# region Books

class BookAPI(Routable):  # 1. Create class
    NAME_MODULE = Book.__name__
    conn: Session = Depends(get_session)  # 2. Add depend

    @get("")
    def get_list_books(self, conn: Session = Depends(get_session)) -> List[Book]:  # 3. Add `self` param and remove conn
        items = conn.execute(select(Book)).scalars().all()
        return items

    @post("")
    def add_book(self, model: Book,
                 conn: Session = Depends(get_session)) -> Book:  # 3. Add `self` param and remove conn
        conn.add(model)
        conn.commit()
        return model

    @delete("{guid}")
    def delete_book(self, guid: str) -> bool:  # 3. Add `self` param and remove conn
        self.conn.execute(
            delete(Book).filter(Book.guid == uuid.UUID(guid))
        )
        self.conn.commit()
        return True

    @put("{guid}")
    def update_book(self, guid: str, model: BookPUT) -> Book:  # 3. Add `self` param and remove conn
        model_db = self.conn.execute(
            select(Book).filter(Book.guid == uuid.UUID(guid))
        ).scalar()
        # Update fields
        for name, val in model.dict(exclude_unset=True).items():
            setattr(model_db, name, val)
        self.conn.commit()
        self.conn.refresh(model_db)
        return model_db


app.include_router(CategoryAPI.routes())  # 4. Include routes
app.include_router(BookAPI.routes())  # 4. Include routes

# endregion
if __name__ == "__main__":
    uvicorn.run('main:app', host="localhost", port=8001, reload=True, debug=True)

The OpenAPI Specification looks like this:

An example of the project is available at the link: #TODO: добавить ссылку

Using Generics, you can get rid of repetitive actions, more details in the “Generics” section.