Generics Issue with a Base Class Deriving from Table: AttributeError: type object 'AppModel' has no attribute '__parameters__'

[pavdwest]

Hi,

I'm really enjoying Piccolo-orm, it's just the right level of customisability and complexity for me - thanks!

I'm however running into an issue now that I'm refactoring things into a more production-like implementation, specifically with my base model class which naturally will contain common functionalities. The following example results in the error as stated in the description, and I just can't figure out how to fix it. My DTO classes are derived from Pydantic, but I don't think that has any bearing on the issue because when deriving my AppModel from a dummy base class (not Table) it all works correctly, so there must be something in the Table class that is not working nicely with generics:

import abc
from typing import Generic, TypeVar, Type
from datetime import datetime

from piccolo.table import Table
from piccolo.columns import Timestamp, BigSerial, Boolean
from piccolo.columns.defaults.timestamp import TimestampNow
from inflection import humanize

from src.dtos import CreateDTO, ReadDTO, UpdateDTO


# Generic typevars for create, read, and update DTOs
CreateDTOClassType = TypeVar('CreateDTOClassType', bound=CreateDTO)
ReadDTOClassType = TypeVar('ReadDTOClassType', bound=ReadDTO)
UpdateDTOClassType = TypeVar('UpdateDTOClassType', bound=UpdateDTO)


class AppModel[
        CreateDTOClassType: Type[CreateDTO],
        ReadDTOClassType: Type[ReadDTO],
        UpdateDTOClassType: Type[UpdateDTO],
    ](Table):

    id = BigSerial(required=True, primary_key=True)
    created_at = Timestamp(required=True, default=TimestampNow)
    updated_at = Timestamp(required=True, default=TimestampNow, auto_update=datetime.now)
    deactivated = Boolean(required=True, default=False)

    # DTOs
    CreateDTOClass: Type[CreateDTO] = None
    ReadDTOClass: Type[ReadDTO] = None
    UpdateDTOClass: Type[UpdateDTO] = None

    @classmethod
    async def create(cls, dto: CreateDTOClassType) -> ReadDTOClassType:
        item = cls(**dto.model_dump())
        await item.save()
        await item.refresh()
        return cls.CreateDTOClass(**item.to_dict())

And the specific model:

from piccolo.columns import Varchar

from src.models import AppModel
from src.modules.note.dtos import NoteCreateDTO, NoteReadDTO, NoteUpdateDTO


# Using generics results in AttributeError: type object 'AppModel' has no attribute '__parameters__'
class Note(
    AppModel[
        NoteCreateDTO,
        NoteReadDTO,
        NoteUpdateDTO
    ]
):
    # Fields
    title = Varchar(required=True)
    body = Varchar(required=True)

    # DTO info
    CreateDTOClass = NoteCreateDTO
    ReadDTOClass = NoteReadDTO
    UpdateDTOClass = NoteUpdateDTO

Any assistance would be much appreciated. Thanks!

[sinisaos]

@pavdwest I don't know exactly what the problem is in your case (no full code provided), but do you want to make a generic crud class for all tables? If so, I would make a BaseTable class with all crud methods from which all other table classes will inherit. Something like this

Example code

import asyncio

import uvicorn
from fastapi import FastAPI
from piccolo.columns import Varchar
from piccolo.table import Table
from pydantic import BaseModel

from piccolo_conf import DB


class BaseTable(Table):
    """
    Base class with crud methods.
    All other tables inherit from this parent class
    """

    @classmethod
    async def read_all(cls):
        return await cls.select().order_by(
            cls._meta.primary_key, ascending=False
        )

    @classmethod
    async def read_single(cls, pk: int):
        row = await cls.select().where(cls._meta.primary_key == pk).first()
        if not row:
            raise ValueError("Record does not exists")
        return row

    @classmethod
    async def create_single(cls, data_model: BaseModel):
        row = cls(**data_model.model_dump())
        await row.save()
        return row.to_dict()

    @classmethod
    async def update_single(cls, pk: int, data_model: BaseModel):
        row = await cls.objects().get(cls._meta.primary_key == pk)
        if not row:
            raise ValueError("Record does not exists")
        for key, value in data_model.model_dump().items():
            setattr(row, key, value)

        await row.save()
        return row.to_dict()

    @classmethod
    async def delete_single(cls, pk: int):
        row = await cls.objects().get(cls._meta.primary_key == pk)
        if not row:
            raise ValueError("Record does not exists")
        await row.remove()


# Table example
class Manager(BaseTable, db=DB):
    name = Varchar()


# Schemas
class CreateDTO(BaseModel):
    name: str


class ReadDTO(BaseModel):
    id: int
    name: str


class UpdateDTO(BaseModel):
    name: str


app = FastAPI()


# Routes example
@app.get("/managers/", response_model=list[ReadDTO])
async def managers_list():
    return await Manager.read_all()


@app.get("/managers/{pk:int}", response_model=ReadDTO)
async def manager_single(pk: int):
    return await Manager.read_single(pk=pk)


@app.post("/managers/", response_model=ReadDTO)
async def manager_create(data_model: CreateDTO):
    return await Manager.create_single(data_model=data_model)


@app.put("/managers/{pk:int}", response_model=ReadDTO)
async def manager_update(pk: int, data_model: UpdateDTO):
    return await Manager.update_single(pk=pk, data_model=data_model)


@app.delete("/managers/{pk:int}", response_model=None)
async def manager_delete(pk: int):
    return await Manager.delete_single(pk=pk)


async def main():
    # Tables creating
    await Manager.create_table(if_not_exists=True)


if __name__ == "__main__":
    asyncio.run(main())

    uvicorn.run(app, host="127.0.0.1", port=8000)

I don't know what web framework you are using but better exceptions should be made based on your usage framework. Hope this helps.

[pavdwest]

Thanks @sinisaos, yes my example doesn't contain any exceptions, I've omitted a lot of code as it has no bearing on the question regarding why generics don't seem to work when using Table as a base class. Here is a full example demonstrating the issue:

from typing import TypeVar, Type
from datetime import datetime
import asyncio

from piccolo.engine.sqlite import SQLiteEngine
from piccolo.table import Table
from piccolo.columns import BigSerial, Timestamp, Varchar
from piccolo.columns.defaults.timestamp import TimestampNow
from pydantic import BaseModel


# Base DTO Classes
class CreateDTO(BaseModel):
    ...


class ReadDTO(BaseModel):
    id: int


CreateDTOType = TypeVar('CreateDTOType', bound=CreateDTO)
ReadDTOType = TypeVar('ReadDTOType', bound=ReadDTO)


# Base Model

# class AppModel(Table):  # Without generics, uncomment this line and comment the next for working implementation without generics
class AppModel[
    CreateDTOType: Type[CreateDTO],
    ReadDTOType: Type[ReadDTO],
    ](Table):

    # Fields
    id = BigSerial(required=True, primary_key=True)
    created_at = Timestamp(required=True, default=TimestampNow)
    updated_at = Timestamp(required=True, default=TimestampNow, auto_update=datetime.now)

    # DTO Classes
    CreateDTOClass: Type[CreateDTOType] = None
    ReadDTOClass: Type[ReadDTOType] = None

    @classmethod
    async def create(cls, dto: CreateDTOType) -> ReadDTOType:
        item = cls(**dto.model_dump())
        await item.save()
        await item.refresh()
        return cls.ReadDTOClass(**item.to_dict())


# Note DTOs
class CreateNoteDTO(CreateDTO):
    title: str
    body: str


class ReadNoteDTO(ReadDTO):
    title: str
    body: str


# Note Model
class NoteModel(
    AppModel[
        CreateNoteDTO,
        ReadNoteDTO,
    ]
    # AppModel # Uncomment this line and comment the previous for working implementation without generics
):
    # Fields
    title = Varchar(required=True)
    body = Varchar(required=True)

    # DTO Classes
    CreateDTOClass = CreateNoteDTO
    ReadDTOClass = ReadNoteDTO


# Delete db if exists
DB = SQLiteEngine(path='db.sqlite')


async def main():
    # Create tables
    await NoteModel.create_table(if_not_exists=True)

    # Test
    see_type_hints_work = await NoteModel.create(
        CreateNoteDTO(
            title='Test Title',
            body='Test Body',
        )
    )
    print(see_type_hints_work)


if __name__ == '__main__':
    asyncio.run(main())

You can see that type hinting will work correctly when e.g. hovering over NoteModel.create in your IDE, however the generics setup fails at runtime.
The reason for using generics is to get nice type hinting using a standard Python feature and avoiding code duplication. E.g. I will also have a generic CRUD controller for each model.

<edit: added some comments>

[sinisaos]

@pavdwest I'm far from an expert, but I don't think it has anything to do with Piccolo. Shouldn't creation of generics be like this

from typing import Generic, TypeVar, Type, get_args

# Base Model
class AppModel(Generic[CreateDTOType, ReadDTOType], Table):
   ...

async def main():
    # check at runtime
    print(get_args(AppModel.__orig_bases__[0])) # result is (~CreateDTOType, ~ReadDTOType)
    ...

That way type hinting works correctly and you can check generics at runtime. Sorry if I'm wrong.

[pavdwest]

@sinisaos thank you very much for the feedback - I'm no expert on Python generics either and tbh I'm quite confused about the latest canonical way of doing things, however, your suggestion to use typing.Generic seems to have done the trick! I was under the impression generics should work without explicitly using typing.Generic from 3.12 onwards but I'm clearly confused, this seems to work nicely now.

Full simplified example for anyone interested, and thanks again @sinisaos, much appreciated:

from typing import Generic, TypeVar, Type
from datetime import datetime
import asyncio

from piccolo.engine.sqlite import SQLiteEngine
from piccolo.table import Table
from piccolo.columns import BigSerial, Timestamp, Varchar
from piccolo.columns.defaults.timestamp import TimestampNow
from pydantic import BaseModel


# Base DTO Classes
class CreateDTO(BaseModel):
    ...


class ReadDTO(BaseModel):
    id: int


CreateDTOType = TypeVar('CreateDTOType', bound=CreateDTO)
ReadDTOType = TypeVar('ReadDTOType', bound=ReadDTO)


# Base Model
class AppModel(
    Generic[CreateDTOType, ReadDTOType],
    Table,
):

    # Fields
    id = BigSerial(required=True, primary_key=True)
    created_at = Timestamp(required=True, default=TimestampNow)
    updated_at = Timestamp(required=True, default=TimestampNow, auto_update=datetime.now)

    # DTO Classes
    CreateDTOClass: Type[CreateDTOType] = None
    ReadDTOClass: Type[ReadDTOType] = None

    @classmethod
    async def create(cls, dto: CreateDTOType) -> ReadDTOType:
        item = cls(**dto.model_dump())
        await item.save()
        await item.refresh()
        return cls.ReadDTOClass(**item.to_dict())


# Note DTOs
class CreateNoteDTO(CreateDTO):
    title: str
    body: str


class ReadNoteDTO(ReadDTO):
    title: str
    body: str


# Note Model
class NoteModel(
    AppModel[
        CreateNoteDTO,
        ReadNoteDTO,
    ]
):
    # Fields
    title = Varchar(required=True)
    body = Varchar(required=True)

    # DTO Classes
    CreateDTOClass = CreateNoteDTO
    ReadDTOClass = ReadNoteDTO


# DB
DB = SQLiteEngine(path='db.sqlite')


async def main():
    # Create tables
    await NoteModel.create_table(if_not_exists=True)

    # Test
    see_type_hints_work = await NoteModel.create(
        CreateNoteDTO(
            title='Test Title',
            body='Test Body',
        )
    )
    print(see_type_hints_work)


if __name__ == '__main__':
    asyncio.run(main())