(Не)адекватное использование Generic в python-коде: получение T-класса и пример нафига это может понадобиться

GenericDAL_example.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Объявление структуры БД.\n",
    "\"\"\"\n",
    "from sqlalchemy import Column, Integer, String\n",
    "from sqlalchemy.orm import declarative_base\n",
    "\n",
    "\n",
    "_Base = declarative_base()\n",
    "\n",
    "\n",
    "class MyOrmModel(_Base):\n",
    "    __tablename__ = \"my_table\"\n",
    "\n",
    "    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)\n",
    "    my_data = Column(String)\n"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Объявление pydantic-моделей для использования в бизнес-логике.\n",
    "\"\"\"\n",
    "from typing import Optional\n",
    "from pydantic import BaseModel\n",
    "\n",
    "\n",
    "class MyModel(BaseModel):\n",
    "    id: int\n",
    "    my_data: Optional[str]\n",
    "\n",
    "    class Config:\n",
    "        orm_mode = True"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Подключение к БД и создание таблиц.\n",
    "\"\"\"\n",
    "from sqlalchemy.orm import sessionmaker\n",
    "from sqlalchemy import create_engine\n",
    "\n",
    "\n",
    "engine = create_engine(\"sqlite:///:memory:\", future=True)\n",
    "create_session = sessionmaker(engine)\n",
    "\n",
    "_Base.metadata.create_all(engine)"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Создание DAL+Mapper, который будет корректно тайпиться.\n",
    "\"\"\"\n",
    "import functools\n",
    "from typing import TypeVar, Generic, Type\n",
    "\n",
    "from sqlalchemy import insert\n",
    "from sqlalchemy.future import select\n",
    "from sqlalchemy.orm import Session\n",
    "\n",
    "\n",
    "_SAModel = TypeVar(\"_SAModel\")\n",
    "_BModel = TypeVar(\"_BModel\")\n",
    "\n",
    "\n",
    "class _BaseDAL(Generic[_SAModel, _BModel]):\n",
    "    \"\"\"Базовый класс для DataAccessLayer'ов.\n",
    "    Содержит стандартные методы с готовой реализацией.\n",
    "    Корректно поддерживает typing и подстановку классов через Generic-типы.\n",
    "    Предназначен для наследования.\n",
    "\n",
    "    Первый Generic-аргумент - модель ORM, второй - бизнес-модель.\n",
    "    \"\"\"\n",
    "\n",
    "    def __init__(self, session: Session):\n",
    "        self.session = session\n",
    "\n",
    "    def create(self, **kwargs) -> None:\n",
    "        \"\"\"Создаёт новую запись с заданными параметрами\"\"\"\n",
    "        q = insert(self._get_orm_model_cls()).values(kwargs)\n",
    "        _ = self.session.execute(q)\n",
    "\n",
    "    def get_all(self, *, limit: int = None, offset: int = 0) -> list[_BModel]:\n",
    "        \"\"\"Возвращает первые limit записей со смещением offset.\n",
    "        Если limit не задан, то возвращает все имеющиеся записи.\n",
    "        \"\"\"\n",
    "        # SQLAlchemy query V2\n",
    "        q = select(self._get_orm_model_cls())\n",
    "        if limit is not None:\n",
    "            q = q.limit(limit).offset(offset)\n",
    "        r = self.session.execute(q)\n",
    "        records = r.scalars().all()\n",
    "\n",
    "        business_model_cls = self._get_business_model_cls()\n",
    "        # pydantic.BaseModel.from_orm\n",
    "        return list(map(business_model_cls.from_orm, records))\n",
    "\n",
    "    # Здесь были другие методы, но для примера они не нужны\n",
    "\n",
    "    @classmethod\n",
    "    def _get_nth_generic_parameter(cls, n: int) -> Type:\n",
    "        \"\"\"Получает n-ный generic-параметр базового класса.\n",
    "\n",
    "        Warning:\n",
    "            Реализация может перестать работать при апгрейде python\n",
    "            из-за нестабильности внутреннего api модуля typing.\n",
    "            Работает на python 3.8+. Протестировано на 3.9.\n",
    "        \"\"\"\n",
    "        from typing import get_args\n",
    "        # noinspection PyUnresolvedReferences\n",
    "        return get_args(cls.__orig_bases__[0])[n]\n",
    "\n",
    "    @classmethod\n",
    "    @functools.lru_cache(maxsize=1)\n",
    "    def _get_orm_model_cls(cls) -> Type[_SAModel]:\n",
    "        \"\"\"Возвращает generic-класс orm-модели БД\"\"\"\n",
    "        orm_model_cls = cls._get_nth_generic_parameter(0)\n",
    "        return orm_model_cls\n",
    "\n",
    "    @classmethod\n",
    "    @functools.lru_cache(maxsize=1)\n",
    "    def _get_business_model_cls(cls) -> Type[_BModel]:\n",
    "        \"\"\"Возвращает generic-класс бизнес-модели по умолчанию\"\"\"\n",
    "        business_model_cls = cls._get_nth_generic_parameter(1)\n",
    "        return business_model_cls\n",
    "\n",
    "\n",
    "# Warn:\n",
    "#     При прямом наследовании линтер Pycharm'а не может просчитать типы.\n",
    "#     Однако если добавить промежуточного наследника, то всё определяется корректно.\n",
    "class BaseDAL(_BaseDAL[_SAModel, _BModel]):\n",
    "    pass"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "outputs": [],
   "source": [
    "class MyOrmModelDAL(BaseDAL[MyOrmModel, MyModel]):\n",
    "    pass"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "id=1 my_data='hello'\n",
      "id=2 my_data='some_data'\n",
      "id=3 my_data=None\n",
      "\n",
      "id=1 my_data='hello'\n",
      "id=2 my_data='some_data'\n",
      "id=3 my_data=None\n"
     ]
    }
   ],
   "source": [
    "session: Session = create_session()\n",
    "repo = MyOrmModelDAL(session)\n",
    "\n",
    "repo.create(my_data=\"hello\")\n",
    "repo.create(my_data=\"some_data\")\n",
    "repo.create()\n",
    "session.commit()\n",
    "\n",
    "print(*repo.get_all(), sep=\"\\n\", end=\"\\n\\n\")\n",
    "\n",
    "repo.create(my_data=\"hello2\")\n",
    "repo.create(my_data=\"some_data2\")\n",
    "repo.create()\n",
    "session.rollback()\n",
    "\n",
    "print(*repo.get_all(), sep=\"\\n\")"
   ],
   "metadata": {
    "collapsed": false,
    "pycharm": {
     "name": "#%%\n"
    }
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}