Generics¶

In previous section¶

In the previous section, we implemented our application using classes. Got rid of the repetition of dependencies.

A simplified example of the work done:

from typing import List

from fastapi import FastAPI, Depends
from sqlmodel import Session, create_engine

from class_based_fastapi import Routable, get, put, post, delete  # 0. Import
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
        ...

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

    @delete("{guid}")
    def delete_category(self, guid: str) -> bool:  # 3. Add `self` param and remove conn
        ...

    @put("{guid}")
    def update_category(self, guid: str, model: CategoryPUT) -> Category:  # 3. Add `self` param and remove conn
        ...
# 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
        ...

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

    @delete("{guid}")
    def delete_book(self, guid: str) -> bool:  # 3. Add `self` param and remove conn
        ...

    @put("{guid}")
    def update_book(self, guid: str, model: BookPUT) -> Book:  # 3. Add `self` param and remove conn
        ...
# endregion

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

The presented code uses classes, but does not have many advantages over the standard FastAPI. It only allows you to get rid of duplicate dependencies.

The real magic is ahead…

Generics and inheritance¶

Let’s get rid of duplications of the same type of database operations and use Generic.

To use the Generic, you need to do the following:

#0: Import the Generic and TypeVar class.
#1: Create generic types to create generic endpoints.
#2: Create generic base API controller (class).
#3: Change the endpoint according to the example.
#4: Inherit controllers from generic controller.
#5: Include all your controllers (endpoints) in the router.

import uuid
from typing import List, Generic, TypeVar  # 0. Import

import sqlalchemy
import uvicorn
from class_based_fastapi import Routable, get, put, post, delete
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


T = TypeVar('T')  # 1. Create generic type
TPut = TypeVar('TPut')  # 1. Create generic type


class BaseAPI(Routable, Generic[T, TPut]):  # 2. Create generic base API controller
    conn: Session = Depends(get_session)

    def __init__(self):
        self._type_db_model = self._get_type_generic(T)

    def _get_type_generic(self, tvar: TypeVar):
        return next(filter(lambda x: x['name'] == tvar.__name__, self.__class__.__generic_attribute__))['type']

    @get("")
    def get_list_categories(self) -> List[T]:  # 3. Specifying  generic types
        items = self.conn.execute(select(self._type_db_model)).scalars().all()
        return items

    @post("")
    def add_category(self, model: T) -> T:
        self.conn.add(model)
        self.conn.commit()
        return model

    @delete("{guid}")
    def delete_category(self, guid: str) -> bool:
        self.conn.execute(
            sqlalchemy.delete(self._type_db_model).filter(self._type_db_model.guid == uuid.UUID(guid))
        )
        self.conn.commit()
        return True

    @put("{guid}")
    def update_category(self, guid: str, model: TPut) -> T:  # 3. Specifying  generic types
        model_db = self.conn.execute(
            select(self._type_db_model).filter(self._type_db_model.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


# Categories
class CategoryAPI(BaseAPI[Category, CategoryPUT]):  # 4. Inheriting the base controller
    NAME_MODULE = Category.__name__


# Books
class BookAPI(BaseAPI[Book, BookPUT]):  # 4. Inheriting the base controller
    NAME_MODULE = Book.__name__


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

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: добавить ссылку