From 901f7c10d811fcc9bcfe4732d6e6481b4d092387 Mon Sep 17 00:00:00 2001 From: 100gle <569590461@qq.com> Date: Thu, 10 Nov 2022 09:48:19 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E6=8A=80=E8=83=BD?= =?UTF-8?q?=E6=89=A9=E5=B1=95N5=E4=B8=80=E7=AB=A0=E7=9B=B8=E5=85=B3?= =?UTF-8?q?=E7=A4=BA=E4=BE=8B=E6=BA=90=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- code/newsletter/N5/class-and-methods.ipynb | 572 +++++++++++++++++++++ code/newsletter/N5/exceptions.py | 48 ++ code/newsletter/N5/generator.py | 22 + code/newsletter/N5/iter.py | 42 ++ 4 files changed, 684 insertions(+) create mode 100644 code/newsletter/N5/class-and-methods.ipynb create mode 100644 code/newsletter/N5/exceptions.py create mode 100644 code/newsletter/N5/generator.py create mode 100644 code/newsletter/N5/iter.py diff --git a/code/newsletter/N5/class-and-methods.ipynb b/code/newsletter/N5/class-and-methods.ipynb new file mode 100644 index 0000000..a90b9b7 --- /dev/null +++ b/code/newsletter/N5/class-and-methods.ipynb @@ -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__}'\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: mysql, sqlite, postgresql, etc.\n", + " (?P.*?): # username\n", + " (?P.*?)@ # password\n", + " (?P.*?): # host\n", + " (?P\\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\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__}'\n", + " )\n", + " return f'{self.__class__.__name__}'\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: mysql, sqlite, postgresql, etc.\n", + " ://(?P.*?) # username\n", + " :(?P.*?) # password\n", + " @(?P.*?) # host\n", + " :(?P\\d+) # port\n", + " [?]*(?P.*) # 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\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\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\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\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\"\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\"\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 +} diff --git a/code/newsletter/N5/exceptions.py b/code/newsletter/N5/exceptions.py new file mode 100644 index 0000000..72c3583 --- /dev/null +++ b/code/newsletter/N5/exceptions.py @@ -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() diff --git a/code/newsletter/N5/generator.py b/code/newsletter/N5/generator.py new file mode 100644 index 0000000..a7d192b --- /dev/null +++ b/code/newsletter/N5/generator.py @@ -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() diff --git a/code/newsletter/N5/iter.py b/code/newsletter/N5/iter.py new file mode 100644 index 0000000..ccb0738 --- /dev/null +++ b/code/newsletter/N5/iter.py @@ -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()