Как управлять классом универсального шаблона, наследующим от другого универсального класса в Python ⇐ Python

[Anonymous]

Я пытаюсь создать общий репозиторий модели Pydantic Schema/SQlAlchemy, используя шаблон/универсальные классы, но с двумя этапами наследования
Первый этап — иметь GenericSchemaRepository с универсальной схемой Pydantic
Этот GenericSchemaRepository может используется с репозиторием модели SqlAlchemy, но с репозиторием другого типа "не sql" (CSV, XML, ...)
Поэтому я хочу, чтобы класс второго этапа, который наследуется от GenericSchemaRepository[Schema] и добавляет явную модель SqlAlchemy, чтобы наконец иметь Репозиторий со схемой и моделью, имеющими общие шаблонные типы
Надеюсь, я достаточно ясно объясняю
У меня это не получается. У меня ошибка
TypeError: Some type variables (~SCHEMA) are not listed in Generic[~MODEL]
Я думаю, что это проблема синтаксиса, поскольку я не совсем знаком с синтаксисом Generic/TypeVar
Заранее спасибо за ваш ответ и ваше время< /p>

from typing import TypeVar,Type,Generic, Any, Optional,List
from abc import ABC,abstractmethod
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.sql.expression import ScalarSelect, and_, or_
from sqlalchemy.orm import Session
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import Mapped
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
Base= declarative_base()

SCHEMA= TypeVar("SCHEMA", bound=BaseModel)
MODEL= TypeVar("MODEL", bound=Any)

SQLALCHEMY_DATABASE_URL = 'sqlite:///./database.db'
engine = create_engine(SQLALCHEMY_DATABASE_URL)

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

class GenericRepository(Generic[SCHEMA],ABC):
@abstractmethod
def get_by_id(self, id: int) -> Optional[SCHEMA]:
raise NotImplementedError()

@abstractmethod
def list(self, **filters) -> List[SCHEMA]:
raise NotImplementedError()

@abstractmethod
def add(self, record: SCHEMA) -> SCHEMA:
raise NotImplementedError()

@abstractmethod
def update(self, record: SCHEMA) -> SCHEMA:
raise NotImplementedError()

@abstractmethod
def delete(self, id: int) -> None:
raise NotImplementedError()

class GenericSqlRepository(GenericRepository[SCHEMA], Generic[MODEL], ABC):

def __init__(self, session: Session, model_cls=Type[MODEL]) -> None:
self._session = session
self._model_cls = model_cls

def _construct_get_stmt(self, id: int) -> ScalarSelect:
stmt = select(self._model_cls).where(self._model_cls.id == id)
return stmt

def get_by_id(self, id: int) -> Optional[SCHEMA]:
stmt = self._construct_get_stmt(id)
# mapper MODEL->SCHEMA missing for record (I know)
return self._session.exec(stmt).first()

def _construct_list_stmt(self, **filters) -> ScalarSelect:
stmt = select(self._model_cls)
where_clauses = []
for c, v in filters.items():
if not hasattr(self._model_cls, c):
raise ValueError(f"Invalid column name {c}")
where_clauses.append(getattr(self._model_cls, c) == v)

if len(where_clauses) == 1:
stmt = stmt.where(where_clauses[0])
elif len(where_clauses) > 1:
stmt = stmt.where(and_(*where_clauses))
return stmt

def list(self, **filters) -> List[SCHEMA]:
stmt = self._construct_list_stmt(**filters)
# mapper MODEL->SCHEMA missing for record (I know)
return self._session.exec(stmt).all()

def add(self, record: SCHEMA) -> SCHEMA:
# mapper SCHEMA->MODEL missing for record (I know)
self._session.add(record)
self._session.flush()
self._session.refresh(record)
return record

def update(self, record: SCHEMA) -> SCHEMA:
# mapper SCHEMA->MODEL missing for record (I know)
self._session.add(record)
self._session.flush()
self._session.refresh(record)
return record

def delete(self, id: int) -> None:
record = self.get_by_id(id)
if record is not None:
self._session.delete(record)
self._session.flush()

class MySchema(BaseModel):
pass

class MyModel(Base):
__tablename__="mytable"
id : Mapped[int] = mapped_column('id', primary_key=True, index=True)

class MyRepository(GenericSqlRepository[MySchema,MyModel]):
pass

myrepo = MyRepository(SessionLocal)


Подробнее здесь: https://stackoverflow.com/questions/791 ... lass-in-py