{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Decorator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "from urllib.parse import parse_qs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Property"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Connector:\n",
    "    def __init__(self, dialect: str, dsn: str) -> None:\n",
    "        self.dialect = dialect\n",
    "        self.dsn = dsn\n",
    "\n",
    "    def __repr__(self) -> str:\n",
    "        return f'{self.__class__.__name__}<dialect: {self.dialect}, dsn: {self.dsn}>'\n",
    "\n",
    "    def __enter__(self):\n",
    "        print(f\"Connecting {self.dialect} database...\")\n",
    "        return self.connect()\n",
    "\n",
    "    def __exit__(self, *exc):\n",
    "        self.close()\n",
    "\n",
    "    def connect(self):\n",
    "        return self\n",
    "\n",
    "    def close(self):\n",
    "        print(f\"Closing database connection...\")\n",
    "\n",
    "    def query(self, sql: str):\n",
    "        print(f\"Executing sql: {sql}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "class Database:\n",
    "\n",
    "    DSNPattern = re.compile(\n",
    "        \"\"\"\n",
    "        (?:\n",
    "            (?P<dialect>.*?):// # dialect: mysql, sqlite, postgresql, etc.\n",
    "            (?P<username>.*?):  # username\n",
    "            (?P<password>.*?)@  # password\n",
    "            (?P<host>.*?):      # host\n",
    "            (?P<port>\\d+)       # port\n",
    "        )\n",
    "        \"\"\",\n",
    "        re.S | re.X,\n",
    "    )\n",
    "\n",
    "    def __init__(self, dsn):\n",
    "        self.dsn = dsn\n",
    "        self._conn = None\n",
    "\n",
    "    @property\n",
    "    def conn(self):\n",
    "        dialect = self._parse_dsn(self.dsn).get(\"dialect\")\n",
    "        self._conn = Connector(dialect=dialect, dsn=self.dsn)\n",
    "        return self._conn\n",
    "\n",
    "    @conn.setter\n",
    "    def conn(self, connector):\n",
    "        if not isinstance(connector, Connector):\n",
    "            raise TypeError(\n",
    "                f\"Can't set an invalid connector to Database `conn` property.\"\n",
    "            )\n",
    "        self._conn = connector\n",
    "\n",
    "    def _parse_dsn(self, dsn: str):\n",
    "        return self.DSNPattern.search(dsn).groupdict()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connector<dialect: mysql, dsn: mysql://user:password@127.0.0.1:3306>\n"
     ]
    }
   ],
   "source": [
    "dsn = \"mysql://user:password@127.0.0.1:3306\"\n",
    "database = Database(dsn=dsn)\n",
    "print(database.conn)\n",
    "\n",
    "# # raise error when value isn't Connector object\n",
    "# database.conn = \"foo\" \n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Classmethod"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Dialect:\n",
    "    @classmethod\n",
    "    def parse_options(cls, qs: str):\n",
    "        raise NotImplementedError"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "class PostgreSQL(Dialect):\n",
    "    @classmethod\n",
    "    def parse_options(cls, qs: str):\n",
    "        features = [\"client_encoding\"]\n",
    "        params = parse_qs(qs)\n",
    "        options = {}\n",
    "        for key in params.keys():\n",
    "            if key in features:\n",
    "                options[key] = params[key]\n",
    "\n",
    "        return options"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "class MySQL(Dialect):\n",
    "    @classmethod\n",
    "    def parse_options(cls, qs: str):\n",
    "        features = [\"charset\", \"timezone\"]\n",
    "        params = parse_qs(qs)\n",
    "        options = {}\n",
    "        for key in params.keys():\n",
    "            if key in features:\n",
    "                options[key] = params[key]\n",
    "\n",
    "        return options"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "class Connector:\n",
    "    def __init__(self, dialect: str, dsn: str, options=None) -> None:\n",
    "        self.dialect = dialect\n",
    "        self.dsn = dsn\n",
    "        self.options = options or None\n",
    "\n",
    "    def __repr__(self) -> str:\n",
    "        if not self.options:\n",
    "            return (\n",
    "                f'{self.__class__.__name__}<dialect: {self.dialect}, dsn: {self.dsn}>'\n",
    "            )\n",
    "        return f'{self.__class__.__name__}<dialect: {self.dialect}, dsn: {self.dsn}, options: {self.options}>'\n",
    "\n",
    "    def __enter__(self):\n",
    "        print(f\"Connecting {self.dialect} database...\")\n",
    "        return self.connect()\n",
    "\n",
    "    def __exit__(self, *exc):\n",
    "        self.close()\n",
    "\n",
    "    def connect(self):\n",
    "        return self\n",
    "\n",
    "    def close(self):\n",
    "        print(f\"Closing database connection...\")\n",
    "\n",
    "    def query(self, sql: str):\n",
    "        print(f\"Executing sql: {sql}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "class Database:\n",
    "\n",
    "    DSNPattern = re.compile(\n",
    "        \"\"\"\n",
    "        (?:\n",
    "            (?P<dialect>.*?)        # dialect: mysql, sqlite, postgresql, etc.\n",
    "            ://(?P<username>.*?)    # username\n",
    "            :(?P<password>.*?)      # password\n",
    "            @(?P<host>.*?)          # host\n",
    "            :(?P<port>\\d+)          # port\n",
    "            [?]*(?P<options>.*)     # database options\n",
    "        )\n",
    "        \"\"\",\n",
    "        re.S | re.X,\n",
    "    )\n",
    "\n",
    "    def __init__(self, dsn: str, dialect: Dialect = None):\n",
    "        self.dsn = dsn\n",
    "        self.dialect = dialect\n",
    "\n",
    "    @property\n",
    "    def conn(self):\n",
    "        parts = self._parse_dsn(self.dsn)\n",
    "        dialect = parts.get(\"dialect\")\n",
    "        qs = parts.get(\"options\")\n",
    "        options = None\n",
    "\n",
    "        if qs and self.dialect:\n",
    "            options = self.dialect.parse_options(qs)\n",
    "\n",
    "        self._conn = Connector(dialect=dialect, dsn=self.dsn, options=options)\n",
    "        return self._conn\n",
    "\n",
    "    @conn.setter\n",
    "    def conn(self, connector):\n",
    "        if not isinstance(connector, Connector):\n",
    "            raise TypeError(\n",
    "                f\"Can't set an invalid connector to Database `conn` property.\"\n",
    "            )\n",
    "        self._conn = connector\n",
    "\n",
    "    def _parse_dsn(cls, dsn: str):\n",
    "        return cls.DSNPattern.search(dsn).groupdict()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connector<dialect: mysql, dsn: mysql://user:password@127.0.0.1:3306>\n"
     ]
    }
   ],
   "source": [
    "dsn = \"mysql://user:password@127.0.0.1:3306\"\n",
    "database = Database(dsn=dsn, dialect=PostgreSQL)\n",
    "print(database.conn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connector<dialect: mysql, dsn: mysql://user:password@127.0.0.1:3306?charset=utf8&timezone=Asia/Shanghai, options: {'charset': ['utf8'], 'timezone': ['Asia/Shanghai']}>\n"
     ]
    }
   ],
   "source": [
    "mysqldsn = (\n",
    "    \"mysql://user:password@127.0.0.1:3306?charset=utf8&timezone=Asia/Shanghai\"\n",
    ")\n",
    "database = Database(dsn=mysqldsn, dialect=MySQL)\n",
    "print(database.conn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connector<dialect: postgresql, dsn: postgresql://user:password@127.0.0.1:5432?client_encoding=utf8, options: {'client_encoding': ['utf8']}>\n"
     ]
    }
   ],
   "source": [
    "pgdsn = \"postgresql://user:password@127.0.0.1:5432?client_encoding=utf8\"\n",
    "database = Database(dsn=pgdsn, dialect=PostgreSQL)\n",
    "print(database.conn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Mixin"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Tesla<name=Roadster, price=$200,000>\n"
     ]
    }
   ],
   "source": [
    "class Vehicle:\n",
    "    def __init__(self, klass) -> None:\n",
    "        self.klass = klass\n",
    "\n",
    "    def move(self):\n",
    "        raise NotImplementedError\n",
    "\n",
    "\n",
    "class Car(Vehicle):\n",
    "    def __init__(self, name) -> None:\n",
    "        super().__init__(klass=\"car\")\n",
    "        self.name = name\n",
    "        self.wheel = 4\n",
    "\n",
    "    def move(self):\n",
    "        print(f\"Car {self.name} is moving...\")\n",
    "\n",
    "\n",
    "class Tesla(Car):\n",
    "    def __init__(self, name, price) -> None:\n",
    "        super().__init__(name)\n",
    "        self.price = price\n",
    "    def __repr__(self) -> str:\n",
    "        return f\"Tesla<name={self.name}, price={self.price}>\"\n",
    "\n",
    "\n",
    "roadster = Tesla(\"Roadster\", \"$200,000\")\n",
    "print(roadster)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Car Roadster is moving...\n"
     ]
    }
   ],
   "source": [
    "roadster.move()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Vehicle:\n",
    "    def __init__(self, klass) -> None:\n",
    "        self.klass = klass\n",
    "\n",
    "    def move(self):\n",
    "        raise NotImplementedError\n",
    "\n",
    "\n",
    "class Car(Vehicle):\n",
    "    def __init__(self, name) -> None:\n",
    "        super().__init__(klass=\"car\")\n",
    "        self.name = name\n",
    "        self.wheel = 4\n",
    "\n",
    "    def move(self):\n",
    "        print(f\"Car {self.name} is moving...\")\n",
    "\n",
    "\n",
    "class PowerChargeMixin:\n",
    "    def charge(self):\n",
    "        import time\n",
    "\n",
    "        print(f\"[INFO] {self.name} is charging...\")\n",
    "        time.sleep(2)\n",
    "        print(\"[INFO] charing over.\")\n",
    "\n",
    "\n",
    "class SentryModeMixin:\n",
    "    def watch(self):\n",
    "        print(\"[INFO] Turning on sentry mode:\")\n",
    "        for _ in range(3):\n",
    "            print(\"\\tguarding...\")\n",
    "\n",
    "\n",
    "class FastAccelerationMixin:\n",
    "    def accelerate(self, speedup):\n",
    "        print(f\"[INFO] {self.name} is speeding up to {speedup}...\")\n",
    "\n",
    "\n",
    "class OTAMixin:\n",
    "    def upgrade(self):\n",
    "        print(f\"[INFO] {self.name}'s OS version is updating...\")\n",
    "\n",
    "\n",
    "class Tesla(\n",
    "    PowerChargeMixin,\n",
    "    SentryModeMixin,\n",
    "    FastAccelerationMixin,\n",
    "    OTAMixin,\n",
    "    Car,\n",
    "):\n",
    "    def __init__(self, name, price) -> None:\n",
    "        super().__init__(name)\n",
    "        self.price = price\n",
    "\n",
    "    def __repr__(self) -> str:\n",
    "        return f\"Tesla<name={self.name}, price={self.price}>\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "roadster = Tesla(\"Roadster\", price=\"$200,000\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] Roadster is charging...\n",
      "[INFO] charing over.\n"
     ]
    }
   ],
   "source": [
    "roadster.charge()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] Roadster is speeding up to 100m/s...\n"
     ]
    }
   ],
   "source": [
    "roadster.accelerate(\"100m/s\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] Roadster's OS version is updating...\n"
     ]
    }
   ],
   "source": [
    "roadster.upgrade()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] Turning on sentry mode:\n",
      "\tguarding...\n",
      "\tguarding...\n",
      "\tguarding...\n"
     ]
    }
   ],
   "source": [
    "roadster.watch()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.9.0 ('pandas-startup')",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.0"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "13977d4cc82dee5f9d9535ceb495bd0ab12a43c33c664e5f0d53c24cf634b67f"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}