Better python code(1-7)

1.Dependency inversion 依赖反转

类似于C#的接口

image.png

• 更改代码

from abc import ABC,abstractmethod

class Switchable(ABC):
    
    def turn_on(self):
        pass

    def turn_off(self):
        pass


class LightBulb(Switchable):
    def turn_on(self):
        print("LightBulb:turned on...")

    def turn_off(self):
        print("LightBulb:turned off...")


class ElectricPowerSwitch:

    def __init__(self,c:Switchable) -> None:
        self.client = c
        self.on =False

    def press(self):
        if self.on:
            self.client.turn_off()
            self.on = False
        else:
            self.client.turn_on()
            self.on = True

l = LightBulb()
switch = ElectricPowerSwitch(l)
switch.press()
switch.press()

更改后可以看出，以前的开关和灯泡有强耦合，现在降低了耦合，在构造函数时，传进的去的是一个接口而不是一个灯泡类。

• 此时我们增加一个电风扇类继承接口即可，这样也不需要更改构造函数直接传入

class Fan(Switchable):
    def turn_on(self):
        print("Fan:turned on...")

    def turn_off(self):
        print("Fan:turned off...")
 
f = Fan()
switch = ElectricPowerSwitch(f)
switch.press()
switch.press()

2.The stractegy pattern 策略模式

还是用上面的abstractmethod

• 给一function指定参数类型和返回类型

from typing import Callable,List

def process_tickets(self,processing_strategy_fn:Callable[[List[SupportTicket]],List[SupportTicket]]):
    return []

3.Observer Pattern 观察者模式

注册一个新用户后，会给开发者发送一条消息，并且再给该新用户发送欢迎邮件，且还会再日志中写一个log

• 未使用观察者模式的代码user.py

image.png

• 使用观察者模式
  even.py

subscribers = dict()

def subscribe(envent_tpye:str,fn):
    if not envent_tpye in subscribers:
        subscribers[envent_tpye] =[]
    subscribers[envent_tpye].append(fn) #将事件和fn方法绑定,用=添加就只是一个key对应一个value，用append那么value的将会是list形式,{'add': [<function add at 0x00000285EE48ADD0>]}

def post_event(event_tpye:str,data):
    if not event_tpye in subscribers:
        return
    for fn in subscribers[event_tpye]:
        fn(data)

log_listener.py

from lib.log import log #记录日志的模块
from .event import subcribe

def handle_user_registed_event(user):
    log(f"User registed with email adress {user.email}")

def handle_user_password_forgotten_event(user):
    log(f"User forgotten with email adress {user.email}")

def setup_log_event_handlers(): # 事件订阅
    subscribe("user_registed",handle_user_registed_event)
    subscribe("user_password_forgotten",handle_user_password_forgotten_event)

user.py只需要使用post_event注册用户即可

def register_new_user(name:str,password:str,email:str):
    user = create_user(name,password,email)
    post_event("user_registered",user)

4.模板方法

类似于订单系统或者爬虫系统，都有一个稳定的流程，登录获取Cookie，爬取网页，处理爬取数据，存储数据的系统都可以使用模板方法

• 不适用设计模式的交易机器人

  
  image.png
  

  -改进的交易机器人

from typing import List
from abc import ABC,abstractmethod

class TradeBot(ABC):
    def connect(self):
        print("connecting to Crypto exchange...")

    def get_market_data(self,coin:str)->List[float]:
        return [10,12,10,14]


    @abstractmethod
    def should_buy(self,prices:List[float])-> bool:
        pass
    
    @abstractmethod
    def should_sell(self,prices:List[float])->bool:
        pass
    def check_prices(self,coin:str):
        self.connect()
        prices = self.get_market_data(coin)
        should_buy = self.should_buy(prices)
        should_sell = self.should_sell(prices)
        if should_buy:
            print(f"you should buy {coin}")
        if should_sell:
            print(f"you should sell")
        else:
            print("no action needed for coin")


class AverageTrader(TradeBot):
    def list_average(self,l:List[float])->float:
        return sum(l)/len(l)

    def should_buy(self,prices:List[float])-> bool:
        return prices[-1]<self.list_average(prices)

    def should_sell(self,prices:List[float])-> bool:
        return prices[-1]>self.list_average(prices)

class MinMaxTrader(TradeBot):
    def should_buy(self,prices:List[float])-> bool:
        return prices[-1]==min(prices)

    def should_sell(self,prices:List[float])-> bool:
        return prices[-1]==max(prices)


# application = MinMaxTrader()
application = AverageTrader() #只需要切换想要的策略即可
application.check_prices("Btc/usd")

5.桥模式

从上面的代码可以看出来，交易机器人不应该有链接网络def conncet和获取数据的功能def get_market_data，为了让交易机器人更加完善，需要引入桥模式

from typing import List
from abc import ABC,abstractmethod

class Exchange(ABC):
    @abstractmethod
    def connect(self):
        pass

    @abstractmethod
    def get_market_data(self,coin:str)->List[float]:
        pass

class BinanceExchange(Exchange):
    def connect(self):
        print("connecting to binance....")

    def get_market_data(self,coin:str)->List[float]:
        return [10,20,19,14]

class AlibabaExchange(Exchange):
    def connect(self):
        print("connecting to Alibaba....")
        
    def get_market_data(self,coin:str)->List[float]:
        return [1,1,1,1]

class TradeBot(ABC):
    def __init__(self,exchange:Exchange) -> None:
        self.exchange =exchange

    @abstractmethod
    def should_buy(self,prices:List[float])-> bool:
        pass
    
    @abstractmethod
    def should_sell(self,prices:List[float])->bool:
        pass
    def check_prices(self,coin:str):
        self.exchange.connect()
        prices = self.exchange.get_market_data(coin)
        should_buy = self.should_buy(prices)
        should_sell = self.should_sell(prices)
        if should_buy:
            print(f"you should buy {coin}")
        if should_sell:
            print(f"you should sell")
        else:
            print("no action needed for coin")


class AverageTrader(TradeBot):
    def list_average(self,l:List[float])->float:
        return sum(l)/len(l)

    def should_buy(self,prices:List[float])-> bool:
        return prices[-1]<self.list_average(prices)

    def should_sell(self,prices:List[float])-> bool:
        return prices[-1]>self.list_average(prices)

class MinMaxTrader(TradeBot):
    def should_buy(self,prices:List[float])-> bool:
        return prices[-1]==min(prices)

    def should_sell(self,prices:List[float])-> bool:
        return prices[-1]==max(prices)


application = MinMaxTrader(AlibabaExchange()) #传入不同的网站即可
# application = AverageTrader(BinanceExchange()) #只需要切换想要的策略即可
application.check_prices("Btc/usd")
>>>
connecting to Alibaba....
you should buy Btc/usd
you should sell

6.错误处理以及上下文管理器

• 自定义错误

import sqlite3

class NotFoundError(Exception):
    pass

class NotAuthorizedError(Exception):
    pass

def fecth_log(id:str):
    try:
        print("连接数据库")
        print("执行数据库语句")
        print("返回数据")
        if not "有权限":
            raise NotAuthorizedError(f"you are now allow to access")

    except sqlite3.OperationalError as e:
        print(e)
        raise NotFoundError(f"Uable to find blog with id{id}")
    finally:
        print("关闭数据库")

• 自定义上下文管理器
  用来处理数据库发生异常后，无法正确关闭链接

import sqlite3

class SQLite:
    def __init__(self,file="application.db") -> None:
        self.file = file
    
    def __enter__(self):
        self.conn = sqlite3.connect(self.file)
        return self.conn.cursor()
    def __exit__(self):
        self.conn.close() 


class NotFoundError(Exception):
    pass

class NotAuthorizedError(Exception):
    pass

def fecth_log(id:str):
    try:
        with SQLite("appliction.db") as cur:
            print("执行数据库语句")
            print("返回数据")
    except sqlite3.OperationalError as e:
        print(e)
        raise NotFoundError(f"Uable to find blog with id{id}")

• retry装饰器

import time
import math
from functools import wraps

def retry(ExceptionToCheck, tries=4, delay=3, backoff=2, logger=None):
    """Retry calling the decorated function using an exponential backoff.

    http://www.saltycrane.com/blog/2009/11/trying-out-retry-decorator-python/
    original from: http://wiki.python.org/moin/PythonDecoratorLibrary#Retry

    :param ExceptionToCheck: the exception to check. may be a tuple of
        exceptions to check
    :type ExceptionToCheck: Exception or tuple
    :param tries: number of times to try (not retry) before giving up
    :type tries: int
    :param delay: initial delay between retries in seconds
    :type delay: int
    :param backoff: backoff multiplier e.g. value of 2 will double the delay
        each retry
    :type backoff: int
    :param logger: logger to use. If None, print
    :type logger: logging.Logger instance
    """
    def deco_retry(f):

        @wraps(f)
        def f_retry(*args, **kwargs):
            mtries, mdelay = tries, delay
            while mtries > 1:
                try:
                    return f(*args, **kwargs)
                except ExceptionToCheck as e:
                    msg = "%s, Retrying in %d seconds..." % (str(e), mdelay)
                    if logger:
                        logger.warning(msg)
                    else:
                        print(msg)
                    time.sleep(mdelay)
                    mtries -= 1
                    mdelay *= backoff
            return f(*args, **kwargs)

        return f_retry  # true decorator

    return deco_retry

@retry(Exception, tries=4)
def test_fail(text):
    raise Exception("Fail")

test_fail("it works!")

• 日志装饰器

import logging 
from functools import wraps

# Example from: https://www.geeksforgeeks.org/create-an-exception-logging-decorator-in-python/

def create_logger(): 
    
    # create a logger object 
    logger = logging.getLogger('exc_logger') 
    logger.setLevel(logging.INFO) 
    
    # create a file to store all the 
    # logged exceptions 
    logfile = logging.FileHandler('exc_logger.log') 
    
    fmt = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    formatter = logging.Formatter(fmt) 
    
    logfile.setFormatter(formatter) 
    logger.addHandler(logfile) 
    
    return logger 

logger = create_logger() 

# you will find a log file 
# created in a given path 


def exception(logger):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except:
                issue = "exception in "+func.__name__+"\n"
                issue = issue+"=============\n"
                logger.exception(issue)
                raise
        return wrapper
    return decorator 


@exception(logger) 
def divideByZero(): 
    return 12/0

# Driver Code 

try:
    divideByZero() 
except:
    pass