python基础--接口与归一化设计封装异常网络编程
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python基础--接口与归一化设计封装异常网络编程相关的知识，希望对你有一定的参考价值。
1 接口与归一化设计
1.1 归一化概念：
归一化的好处：
1.归一化让使用者无需关心对象的类是什么，只需要知道这些对象都具备某些功能就可以了，这极大降低了使用者的使用难度。
2.归一化使得高层的外部使用者可以不加区分的处理所有接口兼容的对象集合
继承的两种用途
一：继承基类的方法，并且做出自己改变或者扩展（代码重用）：实践中，继承的这种用途意义并不很大，甚至常常是有害的。因为它使得子类与基类出现强耦合。
二：声明某个子类兼容于某基类，定义一个接口类（模仿java的Interface），接口类中定义了一些接口名（就是函数名）且并未实现接口的功能，子类继承接口类，并且实现接口中的功能
class Interface: ‘‘‘ 定义接口Interface类来模仿接口的概念， python中压根就没有interface关键字来定义一个接口。 ‘‘‘ def read(self): # 定接口函数read pass def write(self): # 定义接口函数write pass class Txt(Interface): # 文本，具体实现read和write def read(self): print(‘文本数据的读取方法‘) def write(self): print(‘文本数据的读取方法‘) class Sata(Interface): #磁盘，具体实现read和write def read(self): print(‘硬盘数据的读取方法‘) def write(self): print(‘硬盘数据的读取方法‘) class Process(Interface): def read(self): print(‘进程数据的读取方法‘) def write(self): print(‘进程数据的读取方法‘) t = Txt() s = Sata() p = Process() t.read() # 运行结果：文本数据的读取方法 s.read() # 运行结果：硬盘数据的读取方法 p.read() # 运行结果：进程数据的读取方法
1.2 抽象类
了解知识点
与java一样，python也有抽象类的概念但是同样需要借助模块实现，抽象类是一个特殊的类，它的特殊之处在于只能被继承，不能被实例化
借助abc模块，进行模拟抽象类
import abc class Interface(metaclass=abc.ABCMeta): ‘‘‘ 定义接口Interface类来模仿接口的概念， python中压根就没有interface关键字来定义一个接口。 ‘‘‘ @abc.abstractmethod def read(self): # 定接口函数read pass @abc.abstractmethod def write(self): # 定义接口函数write pass class Txt(Interface): # 文本，具体实现read和write def read(self): pass def write(self): pass t = Txt()
1.3 多态
面向对象的多态、多态性
多态：指的是同一种事物的多种形态
动物有多种形态：人、狗、猪
# 多态：同一种事物的多种形态 class Animal: # 同一类事物:动物 def talk(self): pass class People(Animal): # 动物的形态之一:人 def talk(self): print(‘say hello‘) class Dog(Animal): # 动物的形态之二:狗 def talk(self): print(‘say wangwang‘) p=People() d=Dog() p.talk() d.talk()
1.4 多态性
多态性：可以在不考虑实例类型的前提下使用实例
# 多态：同一种事物的多种形态 class Animal: # 同一类事物:动物 def talk(self): pass class People(Animal): # 动物的形态之一:人 def talk(self): print(‘say hello‘) class Dog(Animal): # 动物的形态之二:狗 def talk(self): print(‘say wangwang‘) # 多态性：可以在不考虑实例类型的前提下使用实例 p=People() d=Dog() def Talk(animal): animal.talk() Talk(p) Talk(d)
1.5 多态性的好处
多态性的好处：
1.增加了程序的灵活性
2.增加了程序可扩展性
1.6 鸭子类型
Python崇尚鸭子类型，即‘如果看起来像、叫声像而且走起路来像鸭子，那么它就是鸭子’
python程序员通常根据这种行为来编写程序。例如，如果想编写现有对象的自定义版本，可以继承该对象
也可以创建一个外观和行为像，但与它无任何关系的全新对象，后者通常用于保存程序组件的松耦合度。
利用标准库中定义的各种‘与文件类似’的对象，尽管这些对象的工作方式像文件，但他们没有继承内置文件对象的方法
# 二者都像鸭子,二者看起来都像文件,因而就可以当文件一样去用 class TxtFile: def read(self): pass def write(self): pass class DiskFile: def read(self): pass def write(self): pass
2 封装
2.1 隐藏
在python中用双下划线开头的方式将属性隐藏起来
class Foo: __N = 1111 # _Foo__N def __init__(self,name): self.__Name=name # self._Foo__Name=name def __f1(self): # _Foo__f1 print(‘f1‘) def f2(self): self.__f1() # self._Foo__f1() f = Foo(‘egon‘) # print(f.__N) # 把数据类型隐藏起来了 # 这种隐藏需要注意的问题： # 1：这种隐藏只是一种语法上变形操作，并不会将属性真正隐藏起来 print(Foo.__dict__) print(f.__dict__) # 运行结果：{‘_Foo__Name‘: ‘egon‘} print(f._Foo__Name) # 运行结果：egon print(f._Foo__N) # 运行结果：1111 # 2:这种语法级别的变形，是在类定义阶段发生的，并且只在类定义阶段发生 Foo.__x = 123123 print(Foo.__dict__) # 运行结果：...‘__doc__‘: None, ‘__x‘: 123123} print(Foo.__x) # 运行结果：123123 f.__x = 123123123 print(f.__dict__) # 运行结果：{‘_Foo__Name‘: ‘egon‘, ‘__x‘: 123123123} print(f.__x) # 运行结果：123123123 # 3:在子类定义的__x不会覆盖在父类定义的__x，因为子类中变形成了：_子类名__x, # 而父类中变形成了：_父类名__x，即双下滑线开头的属性在继承给子类时， # 子类是无法覆盖的。 class Foo: def __f1(self): # _Foo__f1 print(‘Foo.f1‘) def f2(self): self.__f1() # self._Foo_f1 class Bar(Foo): def __f1(self): # _Bar__f1 print(‘Bar.f1‘) b = Bar() b.f2() # 运行结果：Foo.f1
2.2 封装数据属性
# 封装不是单纯意义的隐藏 # 1：封装数据属性：将属性隐藏起来，然后对外提供访问属性的接口， # 关键是我们在接口内定制一些控制逻辑从而严格控制使用对数据属性的使用 class People: def __init__(self,name,age): if not isinstance(name,str): raise TypeError(‘%s must be str‘ %name) if not isinstance(age,int): raise TypeError(‘%s must be int‘ %age) self.__Name=name self.__Age=age def tell_info(self): print(‘<名字：%s 年龄：%s>‘ %(self.__Name,self.__Age)) def set_info(self,x,y): if not isinstance(x,str): raise TypeError(‘%s must be str‘ %x) if not isinstance(y,int): raise TypeError(‘%s must be int‘ %y) self.__Name=x self.__Age=y p=People(‘egon‘,18) p.tell_info() # 运行结果：<名字：egon 年龄：18> p.set_info(‘Egon‘,19) p.tell_info() # 运行结果：<名字：egon 年龄：19>
2.3 封装函数属性
封装函数属性的目的：隔离复杂度
# 2：封装函数属性：为了隔离复杂度 # 取款是功能,而这个功能有很多功能组成:插卡、密码认证、输入金额、打印账单、取钱 # 对使用者来说,只需要知道取款这个功能即可,其余功能我们都可以隐藏起来,很明显这么做 # 隔离了复杂度,同时也提升了安全性 class ATM: def __card(self): print(‘插卡‘) def __auth(self): print(‘用户认证‘) def __input(self): print(‘输入取款金额‘) def __print_bill(self): print(‘打印账单‘) def __take_money(self): print(‘取款‘) def withdraw(self): self.__card() self.__auth() self.__input() self.__print_bill() self.__take_money() a = ATM() a.withdraw() # 运行结果： # 插卡 # 用户认证 # 输入取款金额 # 打印账单 # 取款
2.4 静态属性
class Foo: @property def f1(self): print(‘Foo.f1‘) f = Foo() f.f1 # 运行结果：Foo.f1
示例：计算BMI指数
‘‘‘ 例：BMI指数（bmi是计算而来的，但很明显它听起来像是一个属性而非方法， 如果我们将其做成一个属性，更便于理解） 成人的BMI数值： 过轻：低于18.5 正常：18.5-23.9 过重：24-27 肥胖：28-32 非常肥胖, 高于32 体质指数（BMI）=体重（kg）÷身高^2（m） EX：70kg÷（1.75×1.75）=22.86 ‘‘‘ class People: def __init__(self,name,weight,height): self.name=name self.weight=weight self.height=height @property def bmi(self): return self.weight / (self.height**2) p=People(‘jack‘,75,1.80) p.height=1.86 print(p.bmi)
3 面向对象高级
3.1 反射
通过字符串，反射到真实的属性上，找到真实的属性
class Foo: x=1 def __init__(self,name): self.name = name def f1(self): print(‘from f1‘) print(Foo.x) # Foo.__dict__[‘x‘] f = Foo(‘egon‘) print(f.__dict__) print(f.name) print(f.__dict__[‘name‘]) # hasattr print(hasattr(f,‘name‘)) # f.name print(hasattr(f,‘f1‘)) # f.f1 print(hasattr(f,‘x‘)) # f.x # setattr setattr(f,‘age‘,18) # f.age=18 # getattr print(getattr(f,‘name‘)) # f.name print(getattr(f,‘abc‘,None)) # f.abc print(getattr(f,‘name‘,None)) # f.abc func = getattr(f,‘f1‘) # f.f1 func() # delattr delattr(f,‘name‘) # del f.name print(f.__dict__)
3.2 item系列
Item系类，主要包括：__getitem__、__setitem__、 __delitem__，通过这几个item，可以像操作字典一样，操作对象的属性。
class Foo: def __getitem__(self, item): print(‘=====>get‘) return self.__dict__[item] def __setitem__(self, key, value): self.__dict__[key]=value # setattr(self,key,value) def __delitem__(self, key): self.__dict__.pop(key) f = Foo() f[‘name‘] = "jack" # 设置 print(f["name"]) # 取出属性 del f["name"] # 删除 print(f.__dict__)
3.3 __str__
通过__str__，打印对象信息
class People: def __init__(self,name,age,sex): self.name=name self.age=age self.sex=sex def __str__(self): # 在对象被打印时触发执行 return ‘<name:%s age:%s sex:%s>‘ %(self.name,self.age,self.sex) p=People(‘alex‘,38,‘male‘) print(p) # 运行结果：<name:alex age:38 sex:male>
3.4 析构方法
用法：清理一些资源，清理一些python解析器不能清理的资源，例如：清理操作系统的资源，打开文件，向操作系统发起调用，关闭文件句柄资源
class Foo: def __init__(self, name): self.name = name def __del__(self): # 在对象资源被释放时触发 print(‘-----del------‘) f = Foo("mary") del f print(‘对象被释放‘)
4 异常
# 逻辑错误 # TypeError for i in 3: pass # NameError aaaaa # ValueError int(‘asdfsadf‘) # IndexError l=[1,2] l[1000] #KeyError d = {‘a‘:1} d[‘b‘] # AttributeError class Foo:pass Foo.x
4.1 异常处理
如果错误发生的条件可预知的，我们需要用if进行处理，在错误发生之前进行预防
如果错误发生的条件时不可预知的，则需要用到try…except，在错误发生之后进行处理
#基本语法为 try: # 被检测的代码块 except 异常类型： # try中一旦检测到异常，就执行这个位置的逻辑 # 示例 try: f=open(‘a.txt‘) g=(line.strip() for line in f) print(next(g)) print(next(g)) except StopIteration: f.close()
4.2 多分支异常
try: aaaa print(‘====>1‘) l = [] l[3] print(‘====>2‘) d = {} d[‘x‘] print(‘====>3‘) except NameError as e: print(e) except IndexError as e: print(e) except KeyError as e: print(e)
4.3 万能异常
try: # aaaa print(‘====>1‘) l = [] l[3] print(‘====>2‘) except Exception as e: print(e)
4.4 基本结构
try: aaaa print(‘====>1‘) l=[] l[3] print(‘====>2‘) d={} d[‘x‘] print(‘====>3‘) except NameError as e: print(e) except IndexError as e: print(e) except KeyError as e: print(e) except Exception as e: print(e) else: print(‘在没有错误的时候执行‘) finally: print(‘无论有无错误，都会执行‘)
4.5 自定义异常
class EgonException(BaseException): def __init__(self, msg): self.msg=msg def __str__(self): return ‘<%s>‘ %self.msg raise EgonException(‘jack 的异常‘)
5 Socket编程
IP + 端口，标识唯一一个软件、应用
tcp是基于链接的，必须先启动服务端，然后再启动客户端去链接服务端
5.1 简单套接字
5.1.1 服务端
import socket # 买手机 phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 插卡 phone.bind(("127.0.0.1", 8080)) # 开机 phone.listen(5) # 等待电话 print("server start...") conn,client_addr = phone.accept() # tcp链接，client_addr print("链接", conn) print(client_addr) # 基于建立的链接，收发消息 client_data = conn.recv(1024) print("客户端的消息", client_data) conn.send(client_data.upper()) # 挂电话链接 conn.close() # 关机 phone.close()
5.1.2 客户端
import socket phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.connect(("127.0.0.1", 8080)) phone.send("hello".encode("utf-8")) server_data = phone.recv(1024) print("服务端回应的消息", server_data) phone.close()
5.2 加上通信循环
5.2.1 服务端
import socket phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) phone.bind(("127.0.0.1", 8080)) phone.listen(5) print("server start...") conn,client_addr = phone.accept() while True: # 通讯循环 client_data = conn.recv(1024) print("客户端的消息", client_data) conn.send(client_data.upper()) conn.close() phone.close()
5.2.2 客户端
import socket phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.connect(("127.0.0.1", 8080)) while True: msg = input(">>").strip() if not msg:continue phone.send(msg.encode("utf-8")) server_data = phone.recv(1024) print("server back:", server_data.decode("utf-8")) phone.close()
5.3 加上链接循环
5.3.1 服务端
import socket phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) phone.bind(("127.0.0.1", 8080)) phone.listen(5) print("server start...") while True: # 链接循环 conn,client_addr = phone.accept() while True: # 通讯循环 try: client_data = conn.recv(1024) if not client_data:break # 针对linux系统 conn.send(client_data.upper()) except Exception: # 针对windows break conn.close() phone.close()
5.3.2 客户端
import socket phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM) phone.connect((‘127.0.0.1‘,8080)) while True: msg=input(‘>>: ‘).strip() if not msg:continue phone.send(msg.encode(‘utf-8‘)) server_data = phone.recv(1024) print(server_data.decode(‘utf-8‘)) phone.close()
5.4 模拟ssh远程执行命令
5.4.1 服务端
import socket import subprocess phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) phone.bind(("127.0.0.1", 8080)) phone.listen(5) print("server start...") while True: # 链接循环 conn,client_addr = phone.accept() while True: # 通讯循环 try: cmd = conn.recv(1024) if not cmd:break # 针对linux系统 # 执行命令，拿到结果 res = subprocess.Popen(cmd.decode("utf-8"), shell = True, stdout = subprocess.PIPE, stderr = subprocess.PIPE) stdout = res.stdout.read() stderr = res.stderr.read() conn.send(stdout+stderr) except Exception: # 针对windows break conn.close() phone.close()
5.4.2 客户端
import socket phone=socket.socket(socket.AF_INET,socket.SOCK_STREAM) phone.connect((‘127.0.0.1‘,8080)) while True: cmd = input(‘>>‘).strip() if not cmd:continue # 发命令 phone.send(cmd.encode(‘utf-8‘)) # 收命令的执行结果 cmd_res = phone.recv(1024) # 打印结果 print(cmd_res.decode(‘gbk‘)) phone.close()
5.5 粘包现象
Tcp出现粘包现象，udp不会出现粘包；主要是tcp基于流式的，像水流一样，没有开头，没有结尾，源源不断，会发生粘包；udp，是数据报的，不会出现粘包。
5.5.1 服务端
from socket import * s = socket(AF_INET, SOCK_STREAM) s.bind(("127.0.0.1", 8080)) s.listen(5) conn,addr = s.accept() # 收发消息 data1 = conn.recv(1024) print("data1:", data1) data2 = conn.recv(1024) print("data2:", data2) conn.close() s.close()
5.5.2 客户端
from socket import * c = socket(AF_INET, SOCK_STREAM) c.connect(("127.0.0.1", 8080)) c.send("hello".encode("utf-8")) c.send("world".encode("utf-8")) c.close()
5.6 ssh远程执行命令+定制报头
5.6.1 服务端
import socket import struct import subprocess phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) phone.bind(("127.0.0.1", 8080)) phone.listen(5) print("server start...") while True: # 链接循环 conn,client_addr = phone.accept() print(conn, client_addr) while True: # 通讯循环 try: cmd = conn.recv(1024) if not cmd:break # 执行命令，拿到结果 res = subprocess.Popen(cmd.decode("utf-8"), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout = res.stdout.read() stderr = res.stderr.read() # 制作报头 header = struct.pack("i", len(stdout)+len(stderr)) # 先发报头（固定长度） conn.send(header) # 再发真实数据 conn.send(stdout) conn.send(stderr) except Exception: # 针对windows break conn.close() phone.close()
5.6.2 客户端
import socket import struct phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.connect(("127.0.0.1", 8080)) while True: cmd = input(">>").strip() if not cmd:continue # 发命令 phone.send(cmd.encode("utf-8")) # 先收报头 header = phone.recv(4) total_size = struct.unpack("i", header)[0] # 再收命令的执行结果 recv_size = 0 data = b"" while recv_size < total_size: recv_data = phone.recv(1024) recv_size += len(recv_data) data += recv_data # 打印结果 print(data.decode("gbk")) phone.close()
5.7 定制报头的正确方式
5.7.1 服务端
import socket import struct import subprocess import json phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) phone.bind(("127.0.0.1", 8080)) phone.listen(5) print("server start...") while True: # 链接循环 conn,client_addr = phone.accept() print(conn, client_addr) while True: # 通讯循环 try: cmd = conn.recv(1024) if not cmd:break # 执行命令，拿到结果 res = subprocess.Popen(cmd.decode("utf-8"), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout = res.stdout.read() stderr = res.stderr.read() # 制作报头 header_dic = {"total_size":len(stdout)+len(stderr), "md5":None} header_json = json.dumps(header_dic) header_bytes = header_json.encode("utf-8") # 1.先发报头的长度（固定4个bytes） conn.send(struct.pack("i", len(header_bytes))) # 2.再发报头 conn.send(header_bytes) # 3.最后发真实数据 conn.send(stdout) conn.send(stderr) except Exception: # 针对windows break conn.close() phone.close()
5.7.2 客户端
import socket import struct import json phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.connect(("127.0.0.1", 8080)) while True: cmd = input(">>").strip() if not cmd:continue # 发命令 phone.send(cmd.encode("utf-8")) # 先收报头的长度 struct_res = phone.recv(4) header_size = struct.unpack("i", struct_res)[0] # 再收报头 header_bytes = phone.recv(header_size) print(header_bytes) head_json = header_bytes.decode("utf-8") head_dic = json.loads(head_json) total_size = head_dic["total_size"] # 再收命令的执行结果 recv_size = 0 data = b"" while recv_size < total_size: recv_data = phone.recv(1024) recv_size += len(recv_data) data += recv_data # 打印结果 print(data.decode("gbk")) phone.close()
5.8 服务端实现并发
5.8.1 服务端
import socketserver class MyTcphandler(socketserver.BaseRequestHandler): def handle(self): while True: # 通信循环 print(self.request) data = self.request.recv(1024) self.request.send(data.upper()) if __name__ == ‘__main__‘: # 取代链接循环 server = socketserver.ThreadingTCPServer(("127.0.0.1",8080), MyTcphandler) server.serve_forever()
5.8.2 客户端
import socket phone = socket.socket(socket.AF_INET, socket.SOCK_STREAM) phone.connect(("127.0.0.1", 8080)) while True: msg = input(">>").strip() if not msg:continue phone.send(msg.encode("utf-8")) server_data = phone.recv(1024) print(server_data.decode("utf-8")) phone.close()
以上是关于python基础--接口与归一化设计封装异常网络编程的主要内容，如果未能解决你的问题，请参考以下文章