573 lines
14 KiB
Plaintext
573 lines
14 KiB
Plaintext
{
|
|
"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
|
|
}
|