feat: 新增技能扩展N5一章相关示例源码

This commit is contained in:
100gle
2022-11-10 09:48:19 +08:00
parent cb8a449048
commit 901f7c10d8
4 changed files with 684 additions and 0 deletions

View File

@@ -0,0 +1,572 @@
{
"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
}

View File

@@ -0,0 +1,48 @@
class InvalidTypeError(Exception):
def __init__(self, value, code, detail, *args) -> None:
super().__init__(value, *args)
self.code = code
self.detail = detail
def __repr__(self) -> str:
return self.detail
def check(v, klass):
klass_names = [k.__name__ for k in klass]
if not isinstance(v, tuple(klass)):
raise InvalidTypeError(
f"{v.__class__.__name__} class is unknown, use {klass_names}.",
code=500,
detail={"value": v, "type": v.__class__.__name__},
)
return v
def add(a, b):
klass = [int, float]
try:
a = check(a, klass)
b = check(b, klass)
except InvalidTypeError as e:
error = {
"name": InvalidTypeError.__name__,
"code": e.code,
"detail": e.detail,
}
raise TypeError(error) from e
return a + b
def main():
try:
print(add(1, "2"))
except TypeError as e:
print(e.args)
if __name__ == '__main__':
main()

View File

@@ -0,0 +1,22 @@
def make_cards(values):
container = values or []
for value in container:
yield value
def main():
from collections.abc import Generator
cards = make_cards(list("JQK"))
print(cards)
print(f"Is the cards variable a generator? {isinstance(cards, Generator)}")
for card in cards:
print(card)
results = list(cards)
print(results)
if __name__ == '__main__':
main()

View File

@@ -0,0 +1,42 @@
class CardGroup:
def __init__(self, container) -> None:
self.container = container
self.index = 0
def __iter__(self):
return self
def __next__(self):
if self.index < len(self.container):
value = self.container[self.index]
self.index += 1
else:
raise StopIteration
return value
class Cards:
def __init__(self, values=None) -> None:
self._container = values or []
def __iter__(self):
return CardGroup(self._container)
def __getitem__(self, index):
return self._container[index]
def main():
from collections.abc import Iterable
cards = Cards(list("JQK"))
print(f"Is the cards variable an iterale? {isinstance(cards, Iterable)}")
for card in cards:
print(card)
print(cards[1:])
if __name__ == '__main__':
main()