Python 进阶：网络、多线程、异步、数据库、WEB

关于 Python 系列其他文章的传送门

一、些零散的函数¶

glob¶

glob 模块可根据 Unix 终端所用规则找出所有匹配特定模式的路径名。支持使用 ** 的递归。

1 2	import glob glob.glob(pathname, *, recursive=False)

glob 会 return 一个列表，包含匹配到的文件(str)。

当 recursive=True 时，则模式 ** 将匹配目录中的任何文件以及零个或多个目录、子目录和符号链接。

分割字符串 partition 和 split 区别¶

str.partition(str)：根据指定的分隔符将字符串进行分割 -> 如果字符串包含指定的分隔符，则返回一个 3 元的元组，第一个为分隔符左边的子串，第二个为分隔符本身，第三个为分隔符右边的子串；

只分割一次，一分为二

str.split(str="", num=string.count(str))：指定分隔符对字符串进行切片，如果参数 num 有指定值，则分隔 num+1 个子字符串 -> 返回分割后的字符串列表；

分割成 N 个

检查对象是否为可调用 callable()¶

用于检查一个对象是否是可调用的。如果返回 True，object 仍然可能调用失败；但如果返回 False，调用对象 object 绝对不会成功。

对于函数、方法、lambda、类以及实现了 __call__ 方法的类实例对象, 它都返回 True。

也就是说，实例的父类如果没有实现 __call__，则会返回 False。

`@inspect.isXXXXXX`：用于检查数据类型¶

inspect.ismodule(object)：是否为模块；
inspect.isclass(object)：是否为类；
inspect.ismethod(object)：是否为方法。

>>> import inspect

>>> def func():
    pass

>>> class A:
    def method(self):
        pass

>>> print(inspect.isfunction(func),
          inspect.isclass(A),
          inspect.ismethod(A().method))
True True True

`@inspect.signature`：用于获取函数形参的详细信息¶

inspect.signature(fn)：返回函数的形参；
inspect.signature(fn).parameters：返回一个有序字典，以形参为 key，value 为默认值，默认值未定义则为空，是 Parameter 类型；
Parameter_Object.default：提取默认值；
Parameter_Object.annotation：提取注释信息。

import inspect

def add(x: str, y: int = 100):
    return x, y

>>> print(inspect.signature(add))
(x: str, y: int = 100)


>>> print(inspect.signature(add).parameters)
OrderedDict([('x', <Parameter "x: str">), ('y', <Parameter "y: int = 100">)])


>>> dict = inspect.signature(add).parameters

>>> print(dict['x'], dict['y'], sep='  |  ')
x: str  |  y: int = 100


>>> print(dict['x'].default, dict['y'].default, sep='  |  ')
<class 'inspect._empty'>  |  100

>>> print(dict['y'].annotation)
<class 'int'>

`@functools.partial` 偏函数¶

给它传入你的 func，和你想固定下来的参数(*args, **kwargs)，它会返回一个偏函数对象，这个对象和 func 一样，可以被调用，同时在调用的时候如果有更多的位置参数，它们会被附加到 *args 中。如果有额外的关键字参数提供，它们将会合并并覆盖原有的 **kwargs。示例：

from functools import partial

def add(*args):
    return sum(args)

add_100 = partial(add, 100)
print(add_100(1, 2, 3))  # 106

add_101 = partial(add, 101)
print(add_101(1, 2, 3))  # 107

partial 的实现大致等同于如下代码：

def partial(func, *args, **keywords):
    def newfunc(*fargs, **fkeywords):
        newkeywords = keywords.copy()
        newkeywords.update(fkeywords)
        return func(*args, *fargs, **newkeywords)
    newfunc.func = func
    newfunc.args = args
    newfunc.keywords = keywords
    return newfunc

`@functools.total_ordering` 补全缺失的比较运算符的装饰器¶

自动补全：魔术方法中缺失的比较运算符

from functools import total_ordering

"""
可以通过这个装饰器，自动补全其他比较的魔术方法
缺点：效率低！！！！！不适合放在需要频繁比较的核心位置
"""
@total_ordering
class A:
    def __init__(self, age):
        self.age = age

    def __eq__(self, other):    # 等于必须写
        return self.age == other.age

    def __lt__(self, other):    # （lt = 小于）其他的比较方法任选其一
        return self.age < other.age


print(A(4) == A(5))
print(A(4) < A(5))
print(A(4) <= A(5))
print(A(4) > A(5))
print(A(4) >= A(5))

Hashable 测试是否可 hash 值¶

测试是否可 hash，除了调用 hash() 函数，还可以：

1 2	# 判断 a 是否可以 hash isinstance(a, Hashable)

`startswith` 判断字符串前缀的方法¶

startswith()：用于检查字符串是否是以指定子字符串开头，如果是则返回 True，否则返回 False。如果参数 beg 和 end 指定值，则在指定范围内检查。

语法：

1	str.startswith(str, beg=0,end=len(string))

`replace` 字符串替换¶

replace()：把字符串中的 old（旧字符串）替换成 new (新字符串)，如果指定第三个参数 max，则替换不超过 max 次。

1	str.replace(old, new[, max])

脚本模块 `subprocess` 和临时文件模块 `tempfile`¶

执行脚本，而脚本执行的是命令，而命令就是写好的程序，这些程序执行就是一个个进程。

python 有很多运行进程的方式，不过都太旧过时了.
建议使用标准库 subprocess 模块，会启动一个子进程。

class subprocess.Popen(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=True, shell=False, cwd=None, env=None, universal_newlines=False, startupinfo=None, creationfiags=0, restore_signals=True, start_new_session=False, pass_fds=())

shell 为 True，则使用 shell 来执行 args，建议 args 是一个字符串。

from subprocess import Popen, PIPE

p = Popen('echo hello', shell=True, stdout=PIPE)
code = p.wait()  # 阻塞等，code 为 0 是正确执行
text = p.stdout.read()  # bytes
print(code, text)

tempfile.TemporaryFile：
脚本执行的输出可能非常大，使用 PIPE 不太合适，使用临时文件模块 tempfile.TemporaryFile：

这个模块会自动匹配当前操作系统，在临时环境目录中创建一个临时文件，比如 Linux 下的 /tmp 目录中。

from subprocess import Popen, PIPE
from tempfile import TemporaryFile

tf = TemporaryFile()

script = 'echo1 12345\n ping baidu.com -c 2'

texts = []
codes = 0
for line in script.splitlines():  # 按行分割
    with TemporaryFile('w+') as tf:  # 打开临时文件
        p = Popen(line, shell=True, stdout=tf)
        code = p.wait()  # 阻塞等，code 为 0 是正确执行
        tf.flush()  # flush 文件到磁盘
        tf.seek(0)  # 游标回到开头
        text = tf.read()  # 读取执行结果
        texts.append(text)
        codes = codes | code  # 总状态码，有一次运行失败即为失败

print(codes)
print(texts)

IP 地址的约束 ipaddress 库¶

Python 3.3 提供了 ipaddress 库。

1	ipaddress.ip_address(address)

address 可以是 int 或者 str，返回一个 IPAddress 对象。使用 str 则会返回一个 IP 地址。

还可以判断 ip 的类型：

# ipaddress 地址验证

import ipaddress

ips = ['127.0.0.1', '192.168.0.1', '169.254.123.1', '0.0.0.0', '239.168.0.255', '224.0.0.1', '8.8.8.8']
for ip in ips:
    print(ip)
    ip = ipaddress.ip_address(ip)
    print(f"Win {ip.is_link_local}")  # 169.254 地址
    print(f"回环 {ip.is_loopback}")  # 回环
    print(f"多播 {ip.is_multicast}")  # 多播
    print(f"公网 {ip.is_global}")  # 全球范围地址,公网的
    print(f"私有 {ip.is_private}")  # 私有
    print(f"保留 {ip.is_reserved}")  # 保留地址
    print(f"版本 {ip.version}")  # 4
    print('---------')

netifaces – 获取主机网络接口、ip 列表¶

1	pip install netifaces

netifaces.interfaces()：返回接口列表，比如 ['en1', 'en2', 'bridge0']；
netifaces.ifaddresses(interface)：取指定接口上的 IP 地址，返回的是一个字典:
- {18: [{'addr': 'a4:83:e7:e1:44:a2'}], 30: [{'addr': 'fe80::14a9:f7eb:4ebc:7b9a%en0', 'netmask': 'ffff:ffff:ffff:ffff::/64', 'flags': 1024}], 2: [{'addr': '192.168.50.30', 'netmask': '255.255.255.0', 'broadcast': '192.168.50.255'}]}：key 为 2 的就是 IPv4 的地址。

import netifaces

ifaces = netifaces.interfaces()  # 所有的接口列表

for interface in ifaces:
    ips = netifaces.ifaddresses(interface)
    if 2 in ips:  # 如果有 2 这个 key，说明有 ipv4 地址，其他的不需要
        for ip in ips[2]:  # 每个接口上的 ip
            print(ip['addr'])

# 运行结果
127.0.0.1
192.168.50.30
10.211.55.2
10.37.129.2

获取主机名方式¶

使用 hostname 的话，在 windows 和 Linux 取主机名方式不一样。

使用 socket.gethostname() 可以在所有平台使用获取主机名。

二、小技巧¶

`slots` 在字典上节省内存¶

在 Python 中使用了大量的字典来保存类、函数、实例等的属性，而字典为了提升查询效率，必须使用空间换取时间。

一般来说一个对象，属性多一点，都储存在字典中便于查询，问题不大。
但是如果是数百万个对象，那么字典占的就有点大了。这个时候，可以使用 __slots__ 把属性字典给省了。

一旦类提供了 __slots__，就会阻止实例产生保存着属性的 __dict__。

class A:
    w = 5   # 不会影响类的属性

    __slots__ = ('x',)  # 一旦定义了，实例化后就只提供这里定义的属性，__dict__ 就为空了

    def __init__(self):
        self.x = 10
        self.y = 20     # 因为 slots 内只定义了 x,那么当实例化时这句会报错

>>> A().__dict__    # 这句也会报错，因为没有 dict 了
>>> A().__solts__   # 想查询属性，应该这样使用

__slots__ 不影响子类实例，不会继承下去。

应用场景：使用在需要构建数百万以上对象，且内存容量较为紧张，实例的属性简单、固定且不用动态增加的场景。

未实现和未实现异常的区别¶

NotImplemented 是个值，单值，是 NotImplemented 类的实例。
NotImplementedError 是类型，是异常，返回 type。

运算符重载中的反向方法 `radd`¶

__radd__ 的本质是什么，什么时候触发，如何自定义？

测试 a + 1:

class A:
    def __init__(self, x):
        self.x = x

    def __add__(self, other):
        print('i am add')
        return self.x + other.x

    def __radd__(self, other):
        print('i am radd')
        return self.x + other.x


a = A(4)
a + 1   # 等价 a.__add__(1)

# 运行结果会报错
i am add
AttributeError: 'int' object has no attribute 'x'

出现了 AttributeError，因为 1 是 int 类型，没有 x 这个属性，还是 __add__ 被执行了。

再测试 1 + a, 还是报错:

1 2	i am radd AttributeError: 'int' object has no attribute 'x'

但是这次执行的是 a 实例的 __radd__ 方法。

1 + a 等价于 1.__add__(a)，而 int 类型自己实现了 __add__ 方法，为什么却不抛出异常，而是执行了实例 a 的 __radd__ 方法呢？

因为 int 的这个 __add__ 方法发现自己不认识后边的 a，于是返回值就是 NotImplemented，解释器发现是这个值，就会发起对第二操作对象的 __radd__ 方法的调用，也就变成了 a.__radd__(1)

那 1 + a 能解决吗？

class A:
    def __init__(self, x):
        self.x = x

    def __add__(self, other):
        print('i am add')
        try:
            x = other.x
            return self.x + other.x
        except AttributeError:
            try:
                x = int(other)
            except:
                x = 0
            return self.x + x

    def __radd__(self, other):
        print('i am radd')
        return self + other  # 不用单独写一个处理方法，直接抛给自己的 add 处理


a = A(4)
print(a + 1)
print('=============')
print(1 + a)


# 运行结果
i am add
5
=============
i am radd
i am add
5

单例（单实例）¶

# 单实例化

# 第一种实现，通过装饰器

# 设计为只能创建一个实例

def singleton(cls):
    def _wrapper(*args, **kwargs):
        if hasattr(cls, 'count'):
            raise ImportError('只允许有一个实例')
        setattr(cls, 'count', True)

        return cls(*args, **kwargs)

    return _wrapper


@singleton
class A:
    def __init__(self):
        pass


a = A()
b = A()
print(id(a), id(b))


# 第二种实现，通过 __new__

# 设计为可以创建多个实例，单都为同一个

class B:
    def __new__(cls, *args, **kwargs):
        if hasattr(cls, '_instance'):
            return cls._instance
        cls._instance = super().__new__(cls)

        cls._instance.content = args
        return cls._instance

    def __init__(self, some_args):
        self.content = some_args


a = B(1)
b = B(2)
print(id(a), id(b))

三、设计原则¶

关于 return¶

设计一个函数时，应该 return 什么样的数据？比如设计 + 的运算符重载，是返回计算后的值，还是实例对象的本身？

class Items:
    def __init__(self):
        self.x

    # 这样是返回原本的对象，后续可以实现连续加 self + 1 + 2 + 3 ...
    def __add__(self, other):
        self.x = self.x + other.x
        return self
        """
        >>> a = Items(1), b = Items(2), c = Items(3),
        >>> a + b + c
        """

    # 而这样是直接返回一个计算结果，无法实现类似连续加，看需求选择
    def __add__(self, other):
        return self.x + other.x

关于数据容器类型的选择，`list` 还是 `dict`¶

list 适用于查多改少的环境，方便检索。因为 insert 和 remove 的代价比较大，不适合用 list。

注意！！关于 class 中的 `init` 要不要写 `return`¶

在定义 class 时，不要在 __init__ 中写任何 return 语句，__init__ 会自动 return 这个类的实例.

如果一定一定要写，只能写 return None，不接受其他任何参数！！一律报错！！

比如在做装饰器时：

class Tem:
    def __init__(self, fn):
        self.fn
        # 这里不要写任何 return 语句，会自动返回这个类的实例 Tem(add)


@Tem
def add(x, y):
    return x + y

__init__ 会自动返回这个类的实例！！不要写 return ！！！
__init__ 会自动返回这个类的实例！！不要写 return ！！！
__init__ 会自动返回这个类的实例！！不要写 return ！！！

四、文件¶

1. 文件 IO 常用操作¶

command	功能
`open`	打开
`read`	读取
`write`	写入
`close`	关闭
`readline`	行读取
`readlines`	多行读取
`seek`	文件指针操作
`tell`	指针位置

f=open("test") # 返回 file 对象

print(f.read()) # 读取文件

f close() # 关闭文件

open¶

1	open(file, mode='r', buffering=-1, encoding=None, errors=None, newline=None, closefd=True, opener=None)

file¶

打开或者要创建的文件名。

如果不指定路径，默认是当前路径。

mode 模式¶

open 默认用只读模式 rt 打开已经存在的文件。

r

只读打开文件，如果使用 write 方法，会抛异常。
如果文件不存在，抛出 FileNotFoundError 异常。

W

表示只写方式打开，如果读取则抛出异常。
如果文件不存在，则直接创建文件。
如果文件存在，则清空文件内容。

x

文件不存在，创建文件，并只写方式打开。
文件存在，抛出 FileExistsError 异常。

a

文件存在，只写打开，追加内容。
文件不存在，则创建后，只写打开，追加内容。

r 是只读，而 wxa 都是只写

wxa 都可以产生新文件：
- w 不管文件存在与否，都会生成全新内容的文件；
- a 不管文件是否存在，都能在打开的文件尾部追加；
- x 必须要求文件事先不存在，自己造一个新文件。

t 文本模式

字符流，将文件的字节按照某种字符编码理解，按照字符操作。
open 的默认 mode 就是 rt。

b 二进制模式

字节流，将文件就按照字节理解，与字符编码无关。
二进制模式操作时，字节操作使用 bytes 类型。

+

为 r、w、a、x 提供缺失的读写功能，但是，获取文件对象依旧按照 r、w、a、x 自己的特征。
+ 不能单独使用可以认为它是为前面的模式字符做增强功能的。

文件指针¶

文件指针，指向当前字节位置。

mode=r：指针起始在 0；
mode=a：指针起始在 EOF。

显示指针当前位置；

tell()

移动文件指针位置：

seek(offset[, whencel)
- offest：偏移多少字节；
- whence：从哪里开始。
文本模式下，支持从开头向后偏移的方式：
- whence 为 0 缺省值，表示从头开始，offest 只能正整数；
- whence 为 1 表示从当前位置开始偏移，offest 只接受 0，相当于原地不动，所以没什么用；
- whence 为 2 表示从 EOF 开始，offest 只接受 0，相当于移动文件指针到 EOF。
二进制模式下：
- whence 为 0 缺省值，表示从头开始，offest 只能正整数；
- whence 为 1 表示从当前位置开始偏移，offest 可正可负；
- whence 为 2 表示从 EOF 开始，offest 可正可负。
- 二进制模式支持任意起点的偏移，从头、从尾、从中间位置开始。
- 向后 seek 可以超界，但是向前 seek 的时候，不能超界，否则抛异常。

buffering 缓冲区¶

buffer 缓冲区：

缓冲区一个内存空间，一般来说是一个 FIFO 列，到缓冲区满了或者达到阈值，数据才会 flush 到磁盘。
flush()：将缓冲区数据写入磁盘；
close() 关闭前会调用 flush()；
i0.DEFAULT_BUFFER_SIZE：缺省缓冲区大小，单位字节。

如果是二进制模式，使用 io.DEFAULT_BUFFER_SIZE 值，默认是 4096 或者 8192。

如果是文本模式，如果是终端设备，是行缓存方式，如果不是，则使用二进制模式的策略。

-1 表示使用缺省 buffer 大小的；
0 只在二进制模式使用，表示关 buffer；
- 这是一种特殊的二进制模式，不需要内存的 buffer，可以看做是一个 FIFO 的文件；
1 只在文本模式使用，表示使用行缓冲。意思就是见到换行符就 flush；
大于 1 用于指定 buffer 的大小。

buffering	说明
`buffering = -1`	`t` 和 `b`，都是 `io.DEFAULT_BUFFER_SIZE`
`buffering = 0`	`b` 关闭缓冲区，`t` 不支持
`buffering = 1`	`b` 就 1 个字节，`t` 行缓冲，遇到换行符才 flush
`buffering > 1`	`b` 模式表示行缓冲大小。缓冲区的值可以超过 `io.DEFAULT_BUFFER_SIZE`，直到设定的值超出后才把缓冲区 flush。`t` 模式，是 `io.DEFAULT_BUFFER_SIZE`，flush 完后把当前字符串也写入磁盘

总结，似乎看起来很麻烦，一般来说，只需要记得:

文本模式，一般都用默认缓冲区大小；
二进制模式，是一个个字节的操作，可以指定 buffer 的大小；
一般来说，默认缓冲区大小是个比较好的选择，除非明确知道，否则不调整它；
一般编程中，明确知道需要写磁盘了，都会手动调用一次 flush，而不是等到自动 flush 或者 close 的时候。

newline 换行符号¶

文本模式中，转换换行符。

read¶

read(size=-1)：

size 表示读取的多少个字符或字节；
负数或者 None表示读取到 EOF。

readline¶

一行行读取文件内容，size设置一次能读取行内几个字符或字节：

readline(size=-1)

读取所有行的列表，指定 hint 则返回指定的行数：

readlines(hint=-1)

write¶

把字符串 s 写入到文件中并返回字符的个数：

write(s)

将字符串列表写入文件：

writelines(lines)

close¶

flush 并关闭文件对象。

如文件已经关闭，再次关闭没有任何效果。

判断¶

seekable()：是否可 seek；
readable()：是否可读；
writable()：是否可写；
closed()：是否已经关闭。

2. StringIO 和 BytesIO¶

StringIO¶

io 模块中的类，from io import StringIO；
在内存中，开辟的一个文本模式的 buffer，可以像文件对象一样操作它；
当 close 方法被调用的时候，这个 buffer 会被释放。

使用 getvalue()：获取全部内容。跟文件指针没有关系。

from io import StringIO

# 内存中构建
sio = StringIO()
print(sio.readable(), sio.writable(), sio.seekable())

# 像文件对象一样操作
sio.write("monster\nPython")
sio.seek(0)
print(sio.readline())

print(sio.getvalue())  # 无视指针，输出全部内容
sio.close()

优点：

一般来说，磁盘的操作比内存的操作要慢得多，内存足够的情况下，一般的优化思路是少落地，减少磁盘 IO 的过程，可以大大提高程序的运行效率。

BytesIO¶

io 模块中的类，from io import BytesIO；
内存中，开辟的一个二进制模式的 buffer，可以像文件对象一样操作它；
当 close 方法被调用的时候，这个 buffer 会被释放。

from io import BytesIO

# 内存中构建
bio = BytesIO()
print(bio.readable(), bio.writable(), bio.seekable())

bio.write(b"monster\nPython")
bio.seek(0)
print(bio.readline())
print(bio.getvalue())  # 无视指针，输出全部内容

bio.close()

file-like 对象¶

类文件对象，可以像文件对象一样操作。

socket 对象、输入输出对象（stdin、stdout）都是类文件对象。

from sys import stdout

f = stdout
print(type(f))
f.write('magedu.com')

3. 路径操作¶

路径操作模块：

3.4 版本之前，os path 模块；
3.4 版本开始，建议使用 pathlib 模块，提供 Path 对象来操作，包括目录和文件。
- from pathlib import Path

路径拼接和分解，操作符：/

Path 对象 / path 对象；
Path 对象 / 字符串 或者 字符串 / Path 对象。

分解：

parts 属性：可以返回路径中的每一个部分；
joinpath：连接多个字符串到 Path 对象中，joinpath(other)。

获取路径：

str()：获取路径字符串；
bytes()：获取路径字符串的 bytes。

父目录

parent：目录的逻辑父目录；
parents：父目录序列，索引 0 是直接的父；
name：目录的最后一个部分；
stem：目录最后一个部分，没有后缀；
suffix：目录中最后一个部分的扩展名；
suffixes：返回多个扩展名列表；
with_suffix(suffix)：补充扩展名到路径尾部，返回新的路径，扩展名存在则无效；
with_name(name)：替换目录最后一个部分并返回一个新的路径；
home()：返回当前工作目录；
cwd()：返回当前家目录；
resolve() 返回一个新的路径，这个新路径就是当前 Path 对象的绝对路径，如果是软链接则直接被解析；
absolute() 也可以获取绝对路径，但是推荐使用 resolve()；
exists() 目录或文件是否存在；

创建删除文件、文件夹：

touch(mode0o666, existokTrue) 创建一个文件
rmdir() 删除空目录。没有提供判断目录为空的方法
as_uri 将路径返回成 uRl，例如 file:///etc/passwd。
mkdir(mode=00777, parents=False, exist_ok=False)：创建文件夹：
- parents 是否创建父目录：
  - True 等同于 mkdir -p；
  - False 时，父目录不存在，则抛出 FileNotFoundError。
- exist_ok 参数，在 3.5 版本加入：
  - False 时，路径存在，抛出 FileExistsError；
  - True 时，FileExistsError 被忽略。

迭代当前目录：

iterdir()

判断：

is_dir()：是否是目录；
is_file()：是否是普通文件；
is_symlink()：是否是软链接；
is_socket()：是否是 socket 文件；
is_blockdevice()：是否是块设备；
is_chardevice()：是否是字符设备；
is_absolute()；是否是绝对路径。

通配符：

glob(pattern)：通配给定的模式；
rglob(pattern)：通配给定的模式，递归目录；
返回一个生成器。

匹配：

match(pattern)：模式匹配，成功返回 True。

查看元数据：

stat()：相当于 stat 命令；
istat()：同 stat()，但如果是符号链接，则显示符号链接本身的文件信息。

4. 文件操作¶

使用方法，类似 open，返回一个文件对象。

read_bytes()：以 rb 读取路径对应文件，并返回二进制流，3.5 增加的新函数；
read_text(encoding=None, errors=None) 以 rt 方式读取路径对应文件，返回文本；
Path. write_bytes(data)：以 wb 方式写入数据到路径对应文件；
write_text(data, encoding=None, errors=None)：以 wt 方式写入字符串到路径对应文件。

5. os 模块¶

os 也有 open、read、 write 等方法，但是太低级，建议使用内建函数 open、read、 write，使用方法相似。

检查操作系统平台：

属性或方法	结果
`os.name`	windows 是 `nt`, linux 是 `posix`
`os.uname()`	*nix 支持
`sys.platform`	windows 显示 `win32`, linux 显示 `linux`

os.listdir(/tem)：返回目录内容列表；
os.stat(path, *, dir_fd=None, follow_symlinks=True：本质上调用 Linux 系统的 stat
- path：路径的 string 或者 bytes，或者 fd 文件描述符；
- follow_symlinks：为 True 则返回文件本身信息，False 且如果是软链接则显示软链接本身；
os.chmod(path, mode, *, dir_fd=None, follow_symlinks=True)：改权限
- os.chmod('test', 0o777)：权限使用 hex 表示；
os.chown(path, uid, gid)：改变文件的属主、属组，但需要足够的权限。

6. shutil 模块¶

到目前为止，文件拷贝需要打开 2 个文件对象，源文件读取内容，再写入目标文件中来完成拷贝过程。

但是这样会丢失 stat 数据信息（权限等），因为根本没有复制这些信息过去。

而且目录复制怎么办呢？

Python 供了一个方便的库 shutil (高级文件操作)

copy 复制¶

copyfileobj(fsrc, fdst[, length])：文件对象的复制（无法复制内容）
- fsrc 和 fdst 是 open 打开的文件对象。fdst 要求可写；
- length 指定了表示 buffer 的大小。

import shutil

with open('/tmp/test1', 'w+') as f1:
    f1.write('abcd\n1234')
    f1.flush()
    with open('/tmp/test2', 'w+') as f2:
        shutil.copyfileobj(f1, f2)

# 怎么不能复制内容？为什么？怎么改？

copyfile(src, dst, *, follow_symlinks=True)：复制文件内容，不含元数据
- src 和 dst 为文件的路径字符串。本质上调用的就是 copyfileobj，所以不带元数据二进制内容复制；
copymode(src, dst, *, follow_symlinks=True)：仅复制权限；
copystat(src, dst, *, follow_symlinks=True)：复制元数据、stat、包含权限；
copy(src, dst, *, follow_symlinks= True)：复制文件内容、权限和部分元数据，不包括创建时间和修改时间
- 本质上调用的是 copyfile 和 copymode；
copy2 比 copy 多了复制全部元数据，但需要平台支持。

递归复制目录，默认使用 copy2，也就是带更多的元数据复制。

1	shutil.copytree(src, dst, symlinks=False, ignore=None, copy_function=copy2, ignore_dangling_symlinks=False)

其中：

src、dst 必须是目录，src 必须存在，dst 必须不存在；
ignore=func：提供一个函数 callable(src,names) -> ignored_names，它会被调用：
- src 是源目录，names 是 os.listdir(src) 的结果，就是列出 src 中的文件名，返回值是要被过滤的文件名的 set 类型数据。

递归复制目录示例：

import shutil


def ignore(src, names):
    ig = filter(lambda x: x.startswith('a'), names)  # 忽略 a
    return set(ig)

    # /tmp 下所有文件，复制到 /tmp/new 下


shutil.copytree('/tmp', '/tmp/new', ignore=ignore)

rm 删除¶

递归删除，同 rm -rf 一样危险，慎用。

它不是原子操作，有可能删除错误，就会中断，已经删除的就删除了。

shutil.rmtree(path, ignore_errors=False, onerror=None)

ignore_errors 为 True，忽略错误。当为 False 或者 omitted 时 onerror 生效。
onerror 为 callable 时，接受函数 function、path 和 execinfo。

1	shutil.rmtree(/tmp/testdir) # 类似 rm -rf

move 移动¶

move(src, dst, copy_function=copy2)

递归移动文件、目录到目标, 返回目标。

本身使用的是 os.rename 方法。

如果不支持 rename，如果是目录则先 copytree 再删除源目录。

默认使用 copy2 方法。

1 2	os.rename('/tmp/t.txt', '/tmp/newdir/t') os.rename('test3', '/tmp/py/test300')

shutil 还有打包功能，生成 tar 并压缩，支持 zip、gz、bz、x。

7. csv、ini 和 configparser 模块¶

05csv、ini.pdf

略

8. 序列化与反序列化（pickle 库、Json 和 MessagePack 库）¶

06序列化和反序列化.pdf

略

五、正则表达式¶

01正则表达式.pdf

略

六、面向对象¶

1. 多继承¶

03多继承.pdf

略

2. 查看继承的特殊属性和方法有¶

特殊属性和方法	含义	示例
`__base__`	类的基类
`__bases__`	类的基类元组
`__mro__`	显示方法查找顺序，基类的元组
`mro()` 方法	同上	`int.mro()`
`__subclasses__()`	类的子类列表	`int.__subclasses_()`

3. 特殊属性¶

属性	含义
`__name__`	`class`, `func`, `mothod` 的名字，注意：`instance` 没有
`__module__`	`class` 所在的模块名
`__bases__`	类的基类
`__doc__`	`class`, `func` 的文档字符串
`__mro__`	就是 `mro()`，返回继承的顺序
`__dict__`	显示 `class` 或 `instance` 自己的属性，是个可写的字典
`__dir__`	显示 `class` 或 `instance` 的中包括继承在内的所有属性，`dir()` 函数就是调用这个

4. 魔术方法¶

hash¶

方法	意义
`__hash__`	调用内键函数 `hash()` 时的返回值，返回一个整数。
`__eq__`	对应 `==` 操作符，判断 2 个对象是否相等，返回 `bool` 值。去重时，`__hash__` 值可能会冲突，需要再用 `__eq__` 来判断 2 对象是否相等

bool¶

方法	意义
`__bool__`	内键函数 `bool()`，调用这个函数返回布尔值，如没有定义，就找 `__len__()` 返回长度，非 0 为真，如果 `__len__()` 也没有定义，那么所有实例都返回真

可视化¶

方法	意义
`__repr__`	如果没定义，就返回 `object` 的内存地址
`__str__`	运行内键函数 `str()` `format()` `print()` 时调用。如果没有，就找 `__repr__`

__str__ 和 __repr__ 的区别：

在使用 print 或 format 时，如果是 直接作用 在对象上就会找 __str__；
其他 不是直接作用 在对象上的就找 __repr__。而且如果 __str__ 缺失，也会找 __repr__；
如果两个都没定义，就会去 object 中找。

1 2	print(a) # 直接作用 > __str__ print([a]) # 非直接作用 > __repr__

运算符重载¶

方法	意义
`<` `<=` `==` `>` `>=` `!=`	`__lt__` `__le__` `__eq__` `__gt__` `__ge__` `__ne__`
`+` `-` `` `/` `%` `//` `*` `divmod`	`__add__` `__sub__` `__mul__` `__truediv__` `__mod__` `__floordiv__` `__pow__` `__divmod__`
`+=` `-=` `=` `/=` `%=` `//=` `*=`	`__iadd__` `__isub__` `__imul__` `__itruediv__` `__imod__` `__ifloordiv__` `__ipow__`

容器相关¶

方法	意义
`__len__`	内键函数 `len()`,返回对象的长度。
`__iter__`	迭代容器时，返回一个新的可迭代对象
`__contains__`	`in` 成员运算符，如果没有定义，就调用 `__iter__` 方法遍历
`__getitem__`	在 `self[key]` 访问时触发
`__setitem__`	设置值的方法，在 `self[key] = xxx` 时触发
`__missing__`	当对字典或其子类使用 `self[key]`，但是 `key` 不存在时调用

可调用对象¶

方法	意义
`__call__`	当对实例化的对象调用时，触发，比如 `A()()`

上下文管理¶

方法	意义
`__enter__`	进入此对象相关的上下文，如果存在该方法，`with` 语句会把该方法的返回值绑定到 `as` 子句中指定的变量上
`__exit__`	退出与此对象相关的上下文

5. 反射(reflaction)相关¶

反射就是在运行时！运行时！运行时！获取类型信息，在 JAVA 中称为自省。

这种方式优点是可以在运行时，动态对属性进行操作，而装饰器、Mixin 都是定义时就决定了。

能够通过一个实例或对象，找出其 type、class、attribute 或 method，主要围绕实例或类的 __dict__

具有反射能力的函数包括：type()、isinstance()、callable、dir()、getattr() 等

反射的内键函数¶

内键函数	意义
`getattr(object, name, [, default])`	通过 `name` 返回 `object` 的属性值。如不存在，返回 `default`，如没有 `default` 则报错
`setattr(object, name, value)`	`object` 的 `name` 属性存在即覆盖，不存在则新增
`hasattr(object, name)`	判断 `object` 是否有 `name` 这个属性

class A:
    def __init__(self):
        self.x = 10

>>> a = A()
>>> print(a.__dict__)
{'x': 10}

>>> setattr(a, 'x', 20)
>>> print(a.__dict__)
{'x': 20}

>>> setattr(A, 'm', 30)
>>> print(A.__dict__['m'])
30

反射的魔术方法¶

这是提前在类中静态定义的，实例化后，当对象 self 进行某些操作时拦截并触发，无法像使用内键函数那样在运行时修改属性，

魔术方法触发的时机，示例：

class A:
    def __init__(self):                 # 魔术方法触发的时机，统统和实例 self 有关
        self.x = 10                     #   ↓↓↓↓↓↓↓

    def __setattr__(self, key, value):  # A().x = x 赋值时，调用
        pass

    def __getattribute__(self, item):   # A().x 获取时，调用
        pass

    def __getattr__(self, item):        # A().x 都找不到时，调用
        pass

    def __delattr__(self, item):        # del A().x 时，调用
        pass

    # ...

>>> a = A()
>>> a.x

魔术方法	意义
`__getattr__()`	先按继承关系找，都找不到，就调用此方法。如果没有这个，就报错。
`__setattr__()`	增加、修改实例属性时调用，比如 `self.x = x` 时会调用到此方法。
`__delattr__()`	通过实例删除属性时掉用
`__getattribute__()`	实例所有的属性访问、调用，第一个就是到这个方法，它阻止了属性的查找

属性查找顺序:

实例调用 .__getattribute__()
instance.__dict__
instance.__class__.__dict__
继承的祖先类（直到 object )的 __dict__
如都找不到，调用 __getattr__()

使用反射，动态创建属性的小技巧¶

class BaseType:
    def __init__(self, option):  # 实例化时注入参数 arg1 和 arg2
        self.option = option

    def __getattr__(self, item):    # 找不到就返回 None，或者 raise 错误都可以
        ret = self.option.get(item)
        return ret


class A(BaseType):
    pass


a = A({'arg1': 1, 'arg2': 2})

print(a.arg1)   # 通过 . 访问属性时，如果找不到，会触发反射的魔术方法 __getarrt__
print(a.arg2)

6. 描述器(Descriptors)相关¶

一个类如果实现了 __get__()、__set__()、__delete__() 这三个魔术方法的任何一个方法，就是描述器。

如果一个类的类属性设置为描述器（即属性指向一个描述器 a=A() ），那么它被称为 owner 属主。

object.__get__(self, instance, owner)
__set__(self, instance, value)
__delete__(self, instance)
- self：指当前实例，调用者；
- instance：是 owner 的实例；
- owner：是属性的所属的类。

如果仅实现了 __get__()，就是 非数据描述符(non-data descriptor)；
规律：仅在调用类的属性时拦截，可以后期覆盖掉，不影响实例的属性。

>>> class A:
        # 只有一个 get
        def __get__(self, instance, owner):
            print('=get=', self, instance, owner)


>>> class B:
        x = A()

        def __init__(self):
            self.x = 100


>>> print(B.x)   # 调用类属性时，就会被 __get__ 拦截
=get= <__main__.A object at 0x7fcef5099eb0> None <class '__main__.B'>
None

>>> print(B().x) # 但是并不影响调用实例属性
100

如果同时实现了 __get__()、__set__() 就是 数据描述符(data descriptor)。

规律：增加了，在为实例的属性赋值时拦截，并且无法覆盖。

>>> class A:
        def __get__(self, instance, owner):
            print('=get=', self, instance, owner)

        # 多了一个 set，其他一样
        def __set__(self, instance, value):
            print('-set-', self, instance, value)


>>> class B:
        x = A()

        def __init__(self):
            self.x = 100


>>> print(B.x)    # 和以前一样，调用类属性时，会被 __get__ 拦截
=get= <__main__.A object at 0x7ffd26799eb0> None <class '__main__.B'>
None

>>> print(B().x)  # 但不同的是，调用实例属性时，也会被 __set__ 拦截
-set- <__main__.A object at 0x7fc2f4199eb0> <__main__.B object at 0x7fc2f42428e0> 100
=get= <__main__.A object at 0x7fc2f4199eb0> <__main__.B object at 0x7fc2f42428e0> <class '__main__.B'>
None

>>> b = B()
>>> b.x = 200   # 被拦截了！无法覆盖！
-set- <__main__.A object at 0x7ff6c7099eb0> <__main__.B object at 0x7ff6c72473a0> 200

官方说属性查找的优先级会变成：数据描述器 > 实例的 __dict__ > 非数据描述器。

但其实查看实例的 __dict__ 可以看出，在定义了数据描述器时，如果为实例添加同名的属性，不会加到字典里，而是被描述器中的 __set__ 魔术方法拦截，仅此而已。

所以属性访问的顺序从来就没有变过。

七、多线程¶

`threading` 的属性和方法¶

名称	含义
`current_thread()`	返回当前的线程对象
`main_thread()`	返回主线程对象
`active_count()`	当前处于 `active` 状态的线程个数（包括主线程）
`enumerate()`	返回所有活着的线程的列表，不包括已经终止的线程和未开始的线程（包括主线程）
`get_ident()`	返回当前线程的 `ID`，是个非 `0` 整数
`(object).name()`	只是一个名字，可以重名
`(object).ident()`	返回线程是否还活着
`(object).is_alive()`	返回线程是否还活着

`start` 和 `run`¶

1
2
3

a = threading.Thread(target=function, name='test', args='xxx', kwargs='xxx', daemon=None)
a.start()
a.run()

run：就是个函数调用，只是把指定的函数在当前线程调用成一个普通的函数而已；
start：才是真的启动多线程；

线程安全¶

print 函数是 线程不安全 的。
logging 是 线程安全 的。

`logging` 日志输出¶

logging 输出的内容是线程安全的，而 print 是不安全的。

FORMAT 中的格式化信息，自行搜索

Logger 对象的默认级别是 WARING

构造：指定 name，返回一个名称为 name 的 Logger 实例。
未指定 name，返回根 Logger 实例

Handler：

可以单独设置 level；
可以单独设置 format；
可以设置过滤器。

Handler 类继承：

Handler
- StreamHandler：不指定则使用 sys.stderr
  - FileHandler：输出到文件
  - _StderrHandler：标准输出
- HullHandler：什么都不做

继承：

level：如未定义则继承父级；
format：如未定义则使用默认的 %(message)s

"""
# 消息传递流程
产生消息流 s2.error('i am a2 Error')
  ↓
S2 - ERROR    > 判断是否符合 s2 的 level > 交给 s2 的所有 handler   > 符合 handler 的 level > 输出到文件   2020-08-03 23:35:10,623 i am a2 Error
  ↓ 如有父级，并开启了传播开关
s1 - CRITICAL > s1 不再做 level 判断了   > 交给 s1 的所有 handler   > 符合 handler 的 level > 输出到控制台 i am a2 Error
  ↓ 如有父级，并开启了传播开关
root - ERROR  > root 不再做 level 判断了 > 交给 root 的所有 handler > 符合 handler 的 level > 输出到控制台 2020-08-03-23:35:10 4561145280 MainThread i am a2 Error

也就是说，Logging 只对自己这级产生的消息，做一次 Logging 级别的 level 判断，而从子级收到的消息就不再做 Logging 级别的判断了，直接交给自己的 handler 处理

每一级的 handler 都可以自定义 level 和消息格式
  ↓ 未自定义
level: 逐级继承     消息格式: 使用缺省值 %(message)s

注意！！！！format 只能定义到 handler 上
"""

FORMAT = '%(asctime)s %(thread)d %(threadName)s %(message)s'
logging.basicConfig(level=logging.ERROR, format=FORMAT, datefmt='%Y-%m-%d-%H:%M:%S')  # root 级

s1 = logging.getLogger('s1')  # root.s1 级
s1.setLevel(logging.CRITICAL)     # s1 级 为 CRITICAL
hd = logging.StreamHandler()               # 创建一个标准错误输出的 handler
hd_fmt = logging.Formatter('%(message)s')  # 定义消息格式
hd.setFormatter(hd_fmt)
hd.setLevel(logging.INFO)     # 定义 handler 级别
s1.addHandler(hd)


s2 = logging.getLogger('s1.s2')  # root.s1.s2 级
s2.setLevel(logging.WARNING)     # s2 级 为 WARNING
hd = logging.FileHandler('logging.txt')    # 创建一个输出到文件的 handler，注意：错误会输出到指定文件！！！！！
hd_fmt = logging.Formatter('%(asctime)s %(message)s')  # 定义消息格式，有意与 s1 设置的不一样
hd.setFormatter(hd_fmt)
hd.setLevel(logging.WARNING)  # 定义 handler 的级别
s2.addHandler(hd)

s2.error('i am a2 Error')

`daemon` 和 `non-daemon`¶

1	a = threading.Thread(target=function, daemon=false)

主线程执行完了就结束，不会等 daemon 线程，但是会等 non-daemon 线程，如果有 non-daemon 线程，会等他们全部运行结束后，主线程会 kill 掉所有的 daemon 线程并结束运行，依然不管 daemon 线程。当设置为 True 则为 daemon 线程，默认是 False；

应用场景：

后台任务：如发送心跳包、监控；
主线程工作时才有用的线程。如主线程中维护的公共资源，当主线程已经清理了，准备退出了，工作线程使用这些资源也没有意义了，一起退出最合适；
随时可以被终止的线程。

`join`¶

1	join(timeout=None)

一个线程中调用另一个线程的 join 方法，调用者将被阻塞，直到被调用线程终止；

一个线程可以被 join 多次；

调用谁的 join 方法，就是 join 谁，就要等谁，并且不管是不是 daemon 线程，都等；

`threading.local` 类¶

1	global_data = threading.local()

threading.local 类构建了一个大字典，以每一个线程实例的内存地址为 key，如：{ id(Thread) : (ref(Thread), thread-local dict) }；

通过这个类创建的实例，就可以在不同的线程中，安全的使用线程独有的数据，做到了线程间数据隔离，如图本地变量一样安全.

`Timer` 定时器/延迟执行¶

threading.Timer 是 Thread 的子类，具有线程的能力和特性；

通过这个类，可以延迟执行这个线程；

不是延迟执行函数，而是这个线程被延迟执行；

在真正执行前，对可以通过 cancel 取消执行；

`Event`¶

名称	含义
`set()`	标记为 `True`
`clear()`	标记为 `False`
`is_set()`	查看标记是否为 `True`
`wait(timeout=None)`	设置等待标记为 `True` 的时长，`None` 为无限等待，直到返回 `True`，超时未等到返回 `False`

使用同一个 Event 对象的标记 flag；
谁在 wait 就是等到 flag 变为 True，或等到超时返回 False，等待的个数不限制；

Event 的 wait 优于 time.sleep，它会更快的切换到其他线程，提高并发效率。

Event 使用：

from threading import Event, Thread
import logging

logging.basicConfig(level=logging.INFO)


def do(event: Event, interval: int):
    while not event.wait(interval):  # 在判断条件中使用，会返回 True 或 False
        logging.info('do')


e = Event()
Thread(target=do, args=(e, 3)).start()  # 启动一个线程

e.wait(10)  # 等待 10 秒后
e.set()  # event 置为 True 时，上面的 while 循环才会结束

print('END')

应用场景：

生产杯子

`Lock` 排他锁/互斥锁¶

名称	含义
`acquire(blocking=True, timeout=1)`	上锁，默认阻塞，成功获取锁，返回 `True`，否则 `False`，非阻塞时禁止设置超时时间。
`release()`	释放已上锁的锁，可以从任何线程调用释放。在未上锁的对象上调用，则抛 `RuntimeRrror` 异常。

加锁、解锁：
一般来说，上锁就需要解锁，一旦出现异常，锁无法释放，就产生了死锁，为了避免死锁，可以使用：

使用 try...finally 语句保证锁的释放；
使用 with 上下文管理保证解锁，Lock 对象支持使用 with 上下文管理。

示例：

计数器类，可以加、可以减。
当 self._val += 1 或 self._val -=1 在线程中执行的时候，有可能被打断，要加锁。

import threading
from threading import Thread, Lock
import time


class Counter:
    def __init__(self):
        self._val = 0
        self._lock = Lock()

    @property
    def value(self):
        with self._lock:
            return self._val

    def inc(self):
        try:
            self._lock.acquire()
            self._val += 1
        finally:
            self._lock.release()

    def dec(self):
        with self._lock:
            self._val -= 1


def run(c: Counter, count=100):
    for _ in range(count):
        for i in range(-50, 50):
            if i < 0:
                c.dec()
            else:
                c.inc()


t = 10  # 线程数
c = Counter()
c2 = 1000
for i in range(t):
    Thread(target=run, args=(c, c2)).start()

# print(c.value) #这一句合适吗？这一句在主线程中,很早就执行了。
# 改成如下，退出条件是,只剩下主线程的时候。
while True:
    time.sleep(1)
    if threading.active_count() == 1:
        print(threading.enumerate())
        print(c.value)      # 使用锁的情况下，10 个线程都计算多次结果也为 0
        break
    else:
        print(threading.enumerate())

锁的应用场景：

锁适用于访问和修改同一个共享资源的时候，即读写同一个资源的时候。
如果只是读取，并不修改，是不需要加锁的

使用锁的注意事项：

少用锁：使用了锁，多线程访问被锁的资源时，就成了串行，要么排队、要么争抢；
加锁时间越短越好，不需要就立即释放；
一定要避免死锁。

`Rlock` 可重入锁¶

可重入锁，是 线程相关 的锁，

线程 A 获得可重复锁，并可以多次成功获取，不会阻塞。最后要在线程 A 中做和 acquire（获得锁）次数相同的 release（释放锁）。

import threading
import time

lock = threading.RLock()
print(lock.acquire())
print(lock.acquire(blocking=False))  # 指定是否阻塞
print(lock.acquire())
print(lock.acquire(timeout=3.55))  # 默认阻塞，可以定义超时时间
print(lock.acquire(blocking=False))
# print(lock.acquire(blocking=False, timeout=19))  # 不能为非阻塞调用指定超时时间，会异常
""" 上面获取了 5 个锁 """
print('------------')

lock.release()
lock.release()
lock.release()
lock.release()
print(f'main thread {threading.current_thread().ident}')
print(f"lock in main thread ({lock})")  # 注意观察 lock 对象的信息，有 count=1
lock.release()
""" 到这里 5 个已经全部释放了"""
# lock.release() # 如果多释放一次，会报错
print('===========================')

多线程测试：可重入锁，是 线程相关 的锁，可在一个线程中获取锁，并可继续在同一线程中不阻塞的获取锁。当锁未释放完，其他线程获取锁就会阻塞，直到当前持有锁的线程释放完锁。

import threading
import time

""" 多线程测试 """
lock = threading.RLock()


def sub(lk):
    print(f'{threading.current_thread()}: {lk.acquire()}')  # 主线程锁未释放时，就阻塞在这里了
    print(f'{threading.current_thread()}: {lk.acquire(False)}')  # 非阻塞
    print(f'lock in sub thread {lock}')
    lk.release()
    print('sub 1')
    lk.release()
    print('sub 2')
    # lk.release()  # 多了一次


threading.Timer(2, sub, args=(lock,)).start()  # 延迟 2 秒启动线程，传入同一个 lock 对象

print('++++++++++++++++++++++')
print(lock.acquire())
print(lock.acquire())  # 主线程获取了 2 个锁，在释放前，其他线程的锁就会被阻塞

time.sleep(5)
print('释放 2 个主线程锁')
lock.release()
lock.release()

上了几个锁，就要释放对应数量的锁。

`Condition` 生产者、消费者¶

构造方法：conditin(lock=None)，可以传入一个 Lock 对象或 Rlock 对象，如不传默认是 Rlock。

名称	含义
`acquire(*args)`	获取锁
`wait(self, timeout=None)`	等待或超时
`notify(n=1)`	唤醒最多指定数目个数的等待的线程，没有等待的线程就没有任何操作
`notify_all()`	唤醒所有等待的线程

应用场景：

Condition 主要用于生产者、消费者模型，为了解决两者之间速度匹配的问题。
采用了通知机制，非常有效率。

使用方式：

因为 Condition 内部使用了锁（默认为 Rlock 锁），使用前必须先 acquire，用完了要 release，最好的方法是使用 with 上下文；
消费者 wait，等待通知；
生产者生产好消息，对消费者发通知，可以使用 notify 或 notify_all 方法。

使用示例：

消费者消费速度大于生产者生产速度，如果生产者准备好数据了会通知消费者来消费，避免消费者反复来查看数据是否就绪。
1个生产者，多个消费者。

from threading import Thread, Event, Condition
import logging
import random

FORMAT = '%(asctime)s %(threadName)s %(thread)d %(message)s'
logging.basicConfig(format=FORMAT, level=logging.INFO)


# 此例只是为了演示,不考虑线程安全问题

class Dispatcher:
    def __init__(self):
        self.data = None
        self.event = Event()  # event只是为了使用方便,与逻辑无关
        self.cond = Condition()

    # 生产者
    def produce(self, total):
        for _ in range(total):
            data = random.randint(0, 100)
            with self.cond:
                logging.info(data)
                self.data = data
                self.cond.notify_all()  # 通知所有等待的线程
            self.event.wait(1)  # 模拟产生数据速度
        self.event.set()

    # 消费者
    def consume(self):
        while not self.event.is_set():
            with self.cond:
                self.cond.wait()  # 阻塞等通知
                logging.info("received {}".format(self.data))
            self.event.wait(0.5)  # 模拟消费的速度


d = Dispatcher()

# 启动 5 个消费者
for i in range(5):
    c = Thread(target=d.consume, name='consumer-()'.format(i))
    c.start()

# 启动生产者
p = Thread(target=d.produce, args=(10,), name='producer ')
p.start()

注：上例中，程序本身不是线程安全的，程序逻辑有很多瑕疵，但是可以很好的帮助理解 Condition 的使用，和生产者消费者模型。

self.cond.notify-all() 发通知
如果修改为
self.cond.notify(n=2)
代表只通知 2 个消费者

这个例子，可以看到实现了消息的一对多，这其实就是广播模式。

`Barrier` 屏障/栅栏¶

在 3.2 引入的新功能。

名称	含义
`Barrier(parties, action=None, timeout=None)`	构建 `Barrier` 对象，指定参与方数目。`timeout` 是 `wait` 方法未指定超时的默认值
`n_waiting`	当前在屏障中等待的线程数
`parties`	需要多少个等待
`wait(timeout=None)`	执行到此句，等待通过屏障。返回 `0` 到线程数 `-1` 的整数，每个线程返回不同。如果 `wait` 方法设置了超时，当超时后，屏障将处于 broken 状态，直到 `reset`

所有线程冲到了 barrier.wait() 前等待，直到达到 parties 的数目，屏障打开，所有线程停止等待，继续执行。

再有线程来到 barrier.wait()，继续等到，直到到达指定个数。

就像赛马比赛，所有马匹陆续就位，等全员到齐，开闸，下一批马再陆续来到闸门前等待到齐。

名称	含义
`broken`	如果屏障处于打破的状态，返回 `True`
`abort()`	将屏障置于 `broken` 状态，等待中的线程或者调用等待方法的线程中都会抛出 `BrokenBarrierError` 异常，直到 `reset` 方法来恢复异常
`reset()`	恢复屏障，重新开始拦截

try:
    barrier.wait()  # 在屏障等待，直到全员到齐
    print('OK')
except threading.BrokenBarrierError:    # 使用 abort() 打破屏障后会被这个错误拦截
    print('error')

下面的例子中：屏障中等待了 2 个，屏障就被 break 了， waiting 的线程抛了 BrokenBarrierError 异常，新 wait 的线程也抛异常，直到屏障恢复，才继续按照 parties 数目要求继续拦截线程。

import threading
import logging

# 输出格式定义
FORMAT = '%(asctime)-15s\t [%(threadName)s, %(thread)8d] %(message)s'
logging.basicConfig(level=logging.INFO, format=FORMAT)


def worker(barrier: threading.Barrier):
    logging.info(f'waiting for {barrier.n_waiting} threads.')
    try:
        barrier_id = barrier.wait()
        logging.info(f'after barrier {barrier_id}')
    except threading.BrokenBarrierError:
        logging.info('Broken Barrier. run.')


barrier = threading.Barrier(3)
for x in range(0, 9):
    if x == 2:      # 在 x=2 时屏障被打破，等待中的、之后进来的线程都直接抛出 BrokenBarrierError 错误
        barrier.abort()
    elif x == 6:    # 当 x=6 时，屏障恢复，之后线程都在屏障后等待，直到 3 个线程到齐后，一起运行
        barrier.reset()
    threading.Event().wait(1)
    threading.Thread(target=worker, name=f'worker-{x}', args=(barrier,)).start()

应用场景：

并发初始化：
- 所有线程都必须初始化完成后，才能继续工作，例如运行前加载数据、检查，如果这些工作没完成，就开始运行，将不能工作；
- 10 个线程做 10 种准备，每个线程负责一种工作，只有个 10 个线程都完成后，才能继续工作，先完成的要等待后完成的线程；
- 例如：启动一个程序，需要先加载磁盘文件、缓存预热、初始化连接池等工作，这些工作可以齐头并进，但是只有都满足了，程序才能继续向后执行。假设数据库连接失败，则初始化工作失败，就要使用 abort 把 barrier 的状态置为 broken，所有线程收到异常退出。
工作量：
- 有 10 个计算任务，完成 6 个，就算工作完成。

`Semaphore` 信号量¶

和 Lock 很像，信号量对象内部维护一个到计数器，每一次 acquire 都会 -1，当 acquire 方法计数为 0 就阻塞请求的线程，直到其他线程对信号量 release 后，计数大于 0，才恢复阻塞的线程。

名称	含义
`Semaphore(value=1)`	构造方法。`value` 小于 0，抛出 `ValueError` 异常
`acquire(blocking=True, timeout=None)`	获取信号量，计数器 `-1`，获取成功返回 `True`
`release()`	释放信号量，计数器 `+1`

计数器永远不会低于 0，因为 acquire 的时候，发现是 0，就会被阻塞。

应用场景：

连接池：因为资源有限，且开启一个连接的成本高，所以使用连接池。

示例：

一个简单的连接池，连接池应该有容量（总数），有一个工厂方法可以供其他调用者获取连接、返回不用的连接。
下例中，使用信号量解决资源有限的问题。
1. 如果池中有资源，请求者获取资源时先把信号量减 1，再拿走资源。这样当请求超过资源数时，其他请求者只能等待。
2. 当使用者用完归还资源后信号量加 1，等待中的线程就可以被唤醒从而拿走资源。

注意：这个例子不能用到生成环境，只是为了说明信号量使用的例子，还有很多未完成功能。

import threading
import logging
import random

FORMAT = "%(asctime)s %(thread)d %(threadName)s %(message)s"
logging.basicConfig(level=logging.INFO, format=FORMAT)

class Conn:
    def __init__(self, name):
        self.name = name

    def __repr__(self):
        return self.name

class Pool:
    def __init__(self, count: int):
        self.count = count
        # 池中是连接对象的列表
        self.pool = [self._connect(f"conn-{x}") for x in range(count)]
        self.semaphore = threading.Semaphore(count)  # threading.Semaphore()

    def _connect(self, conn_name):
        # 返回一个名称
        return Conn(conn_name)

    def get_conn(self):
        # 从池中拿走一个连接
        logging.info('---------')
        self.semaphore.acquire()    # 使用 semaphore 防止当只有 1 个链接时，其他线程对链接池余量误判的问题
        logging.info('==============')
        conn = self.pool.pop()
        return conn

    def return_conn(self, conn: Conn):
        # 向池中添加一个连接
        self.pool.append(conn)
        self.semaphore.release()

# 连接池初始化
pool = Pool(3)

def worker(pool: Pool):
    conn = pool.get_conn()  # 当拿走 3 个锁达到上限后，之后的就会阻塞在这里，直到前边的进程归还了锁
    logging.info(conn)
    # 模拟使用了一段时间
    threading.Event().wait(random.randint(1, 4))
    pool.return_conn(conn)

for i in range(6):
    threading.Thread(target=worker, name=f"worker-{i}", args=(pool,)).start()

假设还没有使用信号量，就 release，反而会使计数器超过定义的上限，因为这个类对计数器的上限并没有实现什么硬限制，为了解决这个问题，应该使用 BoundedSemaphore 类。

`BoundedSemaphore` 受限信号量¶

有界的信号量，BoundedSemaphore 不允许使用 release 超出初始值范围，否则抛出 ValueError 异常。

用有界信号量，保证如果多 return_conn 就会抛异常。保证了多归还连接抛出异常。

如果归还了同一个连接多次怎么办？使用去重很容易判断出来。

没有获得信号量的线程都在阻塞着，没有线程和归还的线程争抢，当 append 后才 release，这时候等待的线程才能被唤醒，才能 pop，也就是没有获取信号量就不能 pop，这是安全的。
经过上面的分析，信号量比计算列表长度好，线程安全。

信号量和锁¶

锁，只允许同一个时间一个线程独占资源。它是特殊的信号量，即信号量计数器初值为 1。
信号量，可以多个线程访问共享资源，但这个共享资源数量有限。
锁，可以看做特殊的信号量。

concurrent 包¶

concurrent.futures 模块

1	from concurrent import futures

3.2 版本引入的模块。

异步并行任务编程模块，提供一个高级的异步可执行的便利接口。

提供了 2 个池执行器：

ThreadPoolExecutor 异步调用的 线程池 的 Executor；
ProcessPoolExecutor 异步调用的 进程池 的 Executor。

ThreadPoolExecutor 对象¶

首先需要定义一个池的执行器对象，Executor 类子类对象。

方法	含义
`ThreadPoolExecutor(max_workers=1)`	池中至多创建 `max_workers` 个线程的池来同时异步执行，返回 Executor 实例
`submit(fn, args, *kwargs)`	提交执行的函数及其参数，返回 Future 实例
`shutdown(wait=True)`	清理池

ProcessPoolExecutor 对象¶

同上面线程池的方法一样。就是使用多进程完成 ProcessPoolExecutor(max_workers=1)。

Future 类¶

使用 submit 提交执行的函数及其参数后，返回 Future 实例

方法	含义
`done()`	如果调用被成功的取消或者执行完成，返回 True
`cancelled()`	如果调用被成功的取消，返回 True
`running()`	如果正在运行且不能被取消，返回 True
`cancel()`	尝试取消调用。如果已经执行且不能取消返回 False，否则返回 True
`result(timeout=None)`	取返回的结果，如果 `timeout` 为 None，一直等待返回；`timeout` 设置到期，抛出 `concurrent.futures.TimeoutError` 异常
`exception(timeout=None)`	取返回的异常，如果 `timeout` 为 None，一直等待返回；`timeout` 设置到期，抛出 `concurrent.futures.TimeoutError` 异常

示例：

线程池和进程池使用方式
区别仅在创建 executor 实例时语句不一样

# ThreadPoolExecutor例子
import threading
from concurrent import futures
import logging
import time

# 输出格式定义
FORMAT = '%(asctime)-15s\t [%(processName)s:%(threadName)s, %(process)d:%(thread)8d] %(message)s'
logging.basicConfig(level=logging.INFO, format=FORMAT)


def worker(n):
    """ 工作线程，模拟运行 5 秒"""
    logging.info(f'begin to work({n})')
    time.sleep(5)
    logging.info(f'finished({n})')
    return n + 100  # 返回个测试结果


if __name__ == '__main__':
    # 创建进程池，池的容量为 3
    # （进程池代码必须在 main 内）
    # executor = futures.ProcessPoolExecutor(max_workers=3)

    # 创建线程池，池的容量为 3
    # （进程池代码在不在 main 都行）
    executor = futures.ThreadPoolExecutor(max_workers=3)

    fs = []

    with executor:  # 当使用上下文时这样用

        for i in range(3):  # 创建前 3 个工作线程
            future = executor.submit(worker, i)
            fs.append(future)

        for i in range(3, 6):  # 再创建后 3 个
            future = executor.submit(worker, i)
            fs.append(future)

        while True:
            time.sleep(2)  # 每隔两秒检查所有线程是否执行完毕
            logging.info(threading.enumerate())

            flag = True
            for f in fs:  # 判断是否还有未完成的任务
                logging.info(f.done())
                flag = flag and f.done()
                if f.done():  # 如果做完了,看看结果
                    logging.info(f"result = {f.result()}")
                # if not flag:  # 注释 if 看的清楚
                #     break
            logging.info('-' * 30)

            if flag:
                # 如果不用上下文，退出时就不会自动清理资源，下面这一步就需要自己做了
                # executor.shutdown()  # 如果全部运行完毕，就清理池，池中线程全部杀掉
                break

    logging.info("===end===")
    logging.info(threading.enumerate())

# 线程池一旦创建了线程，就不需要频繁清除

支持上下文管理¶

concurrent.futures.ProcessPoolExecutor 继承自 concurrent.futures.base.Executor，而父类有 __enter__、__exit__ 方法，支持上下文管理，可以使用 with 语句。
__exit__ 方法本质其实还是调用的 shutdown(wait=True)，就是一直阻塞到所有运行的任务完成

使用方法：

1
2
3

with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(pow, 323, 1235)
    print (future. result())

使用示例已经融合在上面的示例中。

as_completed 异步处理¶

上面的方法需要每隔一段时间去查看运行结果，而使用 as_completed 则可以帮你盯着运行的 Future 任务，在运行完后会通知你，这是高效的异步处理方法，使用异步/阻塞模型。

concurrent.futures.as_completed(fs, timeout=None)

fs 需要是个装了 Future 实例的迭代器，as_completed(fs) 会把这个 fs 给 fs = set(fs) 一下，之后会帮你盯着这个新 fs 中的 Future 实例，当 fs 中某个 Future 执行完后，会把这个 Future 抛出给你。

迭代器可以用 set、list、dict：

使用 set 或 list 时，内部放的必须是 Future 对象；
使用 dict 时，key 必须是 Future 对象。这样在 fs = set(fs) 时获得的才是 Future 对象。

比如：

from concurrent.futures import ThreadPoolExecutor, as_completed
import time


def fn():
    time.sleep(3)
    return True


params = ['a', 'b', 'c']
# key 必须是个 future 对象

with ThreadPoolExecutor(max_workers=1) as executor:
    future_dict = {executor.submit(fn): p for p in params}

    for future in as_completed(future_dict):
        p = future_dict[future]
        print(p)
        try:
            # 虽然已经执行完了，但可能是非正常退出的，要用 try 拦截一下
            data = future.result()
            print(data)
        except Exception as e:
            print(e)

异步处理示例：

from concurrent.futures import ThreadPoolExecutor, as_completed
import random
import threading
import time


def test_func(s, key):
    print(f'enter~~{threading.current_thread()} {s}s key={key}')
    threading.Event().wait(s)
    if key == 3:
        raise Exception(f"{key} failed~~~")
    return f'ok {threading.current_thread()}'

futures = {}

def run(fs):
    print('~~~~~~~~~~~~~~')
    while True:
        time.sleep(1)
        print('-' * 30)
        print('fs : ', fs)
        # 只要有一个任务没完成就会阻塞在这，
        # 当任意一个任务完成，就抛出一次
        # 如果内部有异常，result() 会将这个异常抛出
        # 有异常也算执行完了 complete
        # fs 内的任务都执行完了就不会阻塞了
        # fs 为空也不阻塞
        for future in as_completed(fs):
            id = fs[future]
            try:
                print(id, future.result())
            except Exception as e:
                print(e)
                print(id, 'failed')


threading.Thread(target=run, args=(futures,)).start()

time.sleep(5)  # 等 5 秒后开始提交任务

# executor = ThreadPoolExecutor(max_workers=3)  # 没使用上下文
# 使用上下文，在退出时清理线程
with ThreadPoolExecutor(max_workers=3) as executor:  # 池最大运行 3 个线程
    for i in range(7):  # 准备提交 7 个任务到池
        futures[executor.submit(test_func, random.randint(1, 8), i)] = i  # 生成任务

总结¶

该库统一了线程池、进程池调用，简化了编程。
是 Python 简单的思想哲学的体现。
唯一的缺点：无法设置线程名称。但这都不值一提。

八、网络¶

Socket¶

协议：AF 表示 Affress Family，用于 socket() 第一个参数

名称	含义
AF_INET	IPV4
AF_INET6	IPV6
AF_UNIX	Unix Domain Socket，windows 没有

Socket 类型：

名称	含义
SOCK_STREAM	TCP，面向连接的流套接字，默认值
SOCK_DGRAM	UDP，无连接的数据报文套接字

TCP 编程¶

TCP Socket 编程步骤：

创建 socket 对象 socket_obj = socket.socket()
绑定在指定的地址端口上 socket_obj.bind(address)
开始监听 socket_obj.listen()
阻塞等待链接 socket_obj.accept()
客户端连接后会返回一个新的 socket 对象和客户端地址的元组：(new_socket_obj, (clientaddr, port))

使用缓冲区接收数据：new_socket_obj.recv(bufsize)
发送数据：new_socket_obj.send(bytes)

TCP Socket 常用方法：

名称	含义
`socket.recv(bufsize)`	获取数据，默认是阻塞的方式
`socket.recvfrom(bufsize)`	获取数据，返回一个二元组`(bytes, address)`
`socket.send(bytes)`	TCP 发送数据
`socket.sendall(bytes)`	TCP 发送全部数据，成功会返回 `None`。其实就是调用 `send`
`socket.sendfile(file, offset=0, count=None)`	3.5 新增。发送一个文件直到 `EOF`，使用系统级高性能的 `os.sendfile` 机制，返回发送的字节数。`offset` 指定起始位置，如果系统不支持（比如win）或不是一个普通文件，就会使用 `send()` 发送文件。

MakeFile¶

1	socket.makefile(self, mode="r", buffering=None, *, encoding=None, errors=None, newline=None):

把该套接字包装成一个文件对象，将 recv 看做读方法，send 看做写方法。

UDP 编程¶

UDP Socket 编程步骤：

创建 socket 对象 socket_obj = socket.socket(type=socket.SOCK_DGRAM)

接收数据：socket_obj.recvfrom(bufsize)，返回一个二元组 (string, address)；
发送数据：socket_obj.sendto(bytes, address)，发给指定地址信息

UDP 是无连接协议，可以只有任意一端。

UDP 的 Socket 对象创建后，是没有占用本地地址和端口的。

UDP Socket 常用方法：

名称	含义
`socket.bind(address, port)`	立即占用指定的地址和端口
`socket.connect(raddr, port)`	立即占用随机的地址和端口，并指定远端地址和端口
`socket.send(bytes)`	使用已经建立的本地端口，把数据发往远端，需配合 `connect`
`socket.sendto(bytes, (raddr, port))`	立即占用随机的地址和端口，并把数据发往远端。可以独立使用
`socket.recv(bytes)`	返回接收的数据，必须要在占用了本地端口后
`socket.recvfrom(bytes, raddr)`	返回接收的数据和对端地址的二元组，必须要在占用了本地端口后

SocketServer¶

socket 编程过于底层，对其进行封装后就是 socketserver 模块，它是网络服务编程框架，便于快速开发。

socketserver 模块提供的不同的类，但是编程接口是一样的，即使是多进程、多线程的类也是一样，大大减少了编程的难度。

类的继承关系：

+------------+
| BaseServer |
+------------+
        |
        v
+-----------+        +------------------+
| TCPServer |------->| UnixStreamServer |
+-----------+        +------------------+
        |
        v
+-----------+        +--------------------+
| UDPServer |------->| UnixDatagramServer |
+-----------+        +--------------------+

SocketServer 简化了网络服务器的编写：

4 个同步类：
- TCPServer
- UDPServer
- UnixStreamServer
- UnixDatagramServer
2 个 Mixin 类: ForkingMixIn 和 ThreadingMixIn 类，通过 MixIN 的方式用来支持异步：
- fork 是创建多进程：
  - class ForkingUDPServer(ForkingMixIn, UDPServer)
  - class ForkingTCPServer(ForkingMixIn, TCPServer)
- thread 是创建多线程:
  - class ThreadingUDPServer(ThreadingMixIn, UDPServer)
  - class ThreadingTCPServer(ThreadingMixIn, TCPServer)

它为每一个连接提供 RequestHandlerClass 类实例，依次调用 setup、handle、finish 方法，且使用了 try...finally 结构保证 finish 方法一定能被调用。这些方法依次执行完成，如果想维持这个连接和客户端通信，就需要在 handle 函数中使用循环。

class BaseRequestHandler:
    def __init__(self, request, client_address, server):
        self.request = request
        self.client_address = client_address
        self.server = server
        self.setup()
        try:
            self.handle()
        finally:
            self.finish()

编程接口：

1、需要提供服务器绑定的地址信息，和用于处理请求的 RequestHandlerClass 类：

1	myserver = socketserver.BaseServer(server_address, RequestHandlerClass)

2、创建一个用于处理请求的 RequestHandlerClass，要求必须要继承自 BaseRequestHandler 的子类，并覆盖其内部的三种方法来创建请求处理程序类，此方法将处理传入的请求：

class RequestHandlerClass(BaseRequestHandler):
    def setup(self): # 每一个连接初始化要做什么
        pass

    def handle(self): # 每一次请求如何处理
        pass

    def finish(self): # 每一个连接清理
        pass

以后就可以在 BaseRequestHandler：类的实例上使用以下属性：

self.request：是和客户端的连接的 socket 对象；
self.server：是 TCPServer 本身；
self.client_address：是客户端地址。

3、启动 socketserver：

1	server.serve_forever() # 永久

zerorpc 库 – RPC 通信模块¶

原生的 Socke 编程太过底层，少使用。任何一门语言都要避免直接使用 socket 库开发，太过底层，难写难维护。

zerorpc 是一个非常轻巧的、跨语言的 RPC 通信模块。

它基于

ZeroMQ：消息传输协议
MessagePack：一种高效压缩的二进制序列化格式。官网：https://msgpack.org

官网：http://www.zerorpc.io/

1	pip install zerorpc

注意：不要把 zerorpc 的方法随便的放到线程中，会抛异常。

Server 代码：

import zerorpc

class MyRPC:
    def hello(self, text):  # RPC 对外的接口，其中处理客户端传来的数据
        return f"send back {text}"

s = zerorpc.Server(MyRPC())
s.bind("tcp://0.0.0.0:9000")  # 绑定
s.run()  # 持久运行监听

Client 循环发送代码:

import zerorpc
import threading

c = zerorpc.Client()
c.connect('tcp://127.0.0.1:9000')
e = threading.Event()

while not e.wait(3):
    print(c.hello('test client'))
    print('~~~~~~~~~~~~')

九、异步编程¶

同步/异步、阻塞/非阻塞¶

同步和异步的区别，强调的是结果，调用者是否得到了想要的最终结果：

同步：一直等到拿到最终结果，并且只要最终结果：
异步：就算暂时没拿到结果，也可以接受一些别的信息，但是最后还是要拿到结果

阻塞和非阻塞的区别，强调的是时间，是否等待，调用者是否还能干其他事：

阻塞：调用者就只能干等；
非阻塞：调用者可以先去忙会别的，不用一直等。

异步 IO¶

进程发起异步 IO 请求，立即返回。内核完成 IO 的两个阶段，内核给进程发一个信号。
举例：来打饭，跟大师傅说饭好了叫你，饭菜准备好了，窗口服务员把饭盛好了打电话叫你。两阶段都是异步的。在整个过程中，进程都可以忙别的，等好了才过来。
举例：今天不想出去到饭店吃饭了，点外卖，饭菜在饭店做好了(第一阶段)，快递员从饭店送到你家门口(第二阶段)。

Linux 的 aio 的系统调用，内核从版本 2.6 开始支持

同步 IO¶

同步 IO 模型包括：

阻塞 IO；
非阻塞 IO；
IO 多路复用。

IO 多路复用¶

所谓 IO 多路复用，就是同时监控多个 IO，有一个准备好了，就不需要等了开始处理，提高了同时处理 IO 的能力。他也应该属于同步模型，因为强调只要结果，不接受中间值。

实现：
- select：几乎所有操作系统平台都支持；
- poll：是对的 select 的升级；
- epoll：对 select 和 poll 的增强，在监视的基础上，增加回调机制。Linux 系统内核 2.5+ 开始支持，BSD、Mac 平台的实现是叫 kqueue，Windows 的实现是 iocp。
对比区别举例：食堂供应很多菜，有很多的取菜窗口(众多的IO)，你需要吃某三个菜，大师傅(操作系统)说要现做，需要等，你只好等待。
- select：其中一样菜好了，大师傅叫你过来说你点的菜有好的了，你得自己挨个去取菜窗口找找看哪一样好了，请服务员把做好的菜打给你。
- epoll：有菜准备好了，大师傅直接告诉你应该去几号窗口打菜，不用自己找菜了。

一般情况下，select 最多能监听 1024 个 fd(文件描述符，可以修改，但不建议改)，但是由于 select 采用轮询的方式，当管理的 IO 多了，每次都要遍历全部 fd，效率低下。

而 epoll 没有管理的 fd 的上限，且是回调机制，不需遍历，效率很高。

Python 中 IO 多路复用¶

IO多路复用：
- 大多数操作系统都支持 select 和 poll；
- Linux 2.5+ 支持 epoll；
- BSD、Mac 支持 kqueue；
- Windows 的 IOCP。

selectors 库¶

select 库属于比较底层的 IO 多路复用模块。实现了 select、poll 系统调用，这个基本上操作系统都支持。部分实现了epoll，因为太底层，使用起来不好用。

selectors 库，Python 3.4 版本开始提供，这是个高级的 IO 多路复用库。推荐使用

selectors 类的层次结构：
BaseSelector
±- SelectSelector 实现 select
±- PollSelector 实现 poll
±- EpollSelector 实现 epoll
±- DevpollSelector 实现 devpoll
±- KqueueSelector 实现 kqueue

性能排序：epoll|kqueue|devpoll > poll > select

开发中的选择
1. 完全跨平台，使用 select、poll，但是性能较差。
2. 针对不同操作系统自行选择支持的技术，这样做会提高 IO 处理的性能

调用方法：

My = selectors.DefaultSelector 会返回当前平台最有效、性能最高的实现。但是，由于没有实现 Windows 下的 IOCP，所以无法实现的只能退化为 select。
My.register(fileobj, events, data=None) 为 selector 注册一个文件对象，监视它的 IO 事件。
- fileobj 指定被监视文件对象，例如 socket 对象；
- events 指定事件是读还是写，触发的条件；
- data 可以指定一个类、函数、方法、实例等等。在 event 事件触发后，selector 会把这里指定的对象给你返回来。

Event 常量	含义
`EVENT_READ`	可读 `0b01`，内核已经准备好输入输出设备，可以开始读了
`EVENT_WRITE`	可写 `0b10`，内核准备好了，可以往里写了

aiohttp 和协程¶

07asyncio

初识异步IO框架：asyncio

理解Python协程的本质

十、WEB 开发¶

1. WSGI 服务器 wsgiref¶

WSGI 主要规定了服务器端和应用程序间的接口。

而 wsgiref 这个一个 WSGI 参考实现库，练习用的，不能用于生产！

wsgiref.simple_server 模块实现一个简单的 WSGI HTTP 服务器；
wsgiref.simple_server.make_server(host, port, app, server_class=WSGIServer, handler_class=WSGIRequestHandler)
启动一个 WSGI 服务器；
wsgiref.simple_server.demo_app(environ, start_response) 一个函数，小巧完整的 WSGI 的应用程序的实现

返回一个文本的例子：

from wsgiref.simple_server import make_server, demo_app

ip = '127.0.0.1'
port = 9999
server = make_server(ip, port, demo_app) # demo_app 一个内置的小巧完整的 WSGI 的应用程序，可调用
server.serve_forever() # 永久运行
# server.handle_request() 是执行一次

WSGI 服务器作用¶

监听 HTTP 服务端口(TCPServer，默认端口 80)；
接收浏览器端的 HTTP 请求并解析封装成 environ 环境数据；
负责调用应用程序，将 environ 和 start_response 方法传入应用程序；
将应用程序响应的正文封装成 HTTP 响应报文返回浏览器端。

WSGI APP 应用程序端¶

应用程序应该是一个可调用对象 Python 中应该是函数、类、实现了 __call__ 方法的类的实例；
这个可调用对象应该接收两个参数，environ 和 start_response。
以上的可调用对象实现，都必须返回一个可迭代对象
environ 和 start_response 这两个参数名可以是任何合法名，但是一般默认都是这 2 个名字。

res_str = b'monster.com\n'

# 函数实现
def application(environ, start_response):
    return [res_str]

# 类实现
class Application:
    def __init__(self, environ, start_response):
        pass
    def __iter__(self): # 实现此方法，对象即可迭代
        yield res_str

# 类的实例实现
class Application:
    def __call__(self, environ, start_response):
        return [res_str]

environ¶

environ 是包含了 HTTP 请求中各种信息的 dict 对象。包括请求方法、URL 中路径部分、查询字符串、IP、Port、协议、UserAgent 信息等

start_response¶

它是一个可调用对象。有 3 个参数，定义如下:
start_response(status, response_headers, exc_info=None)

status 是状态码，例如 200 OK；
response_headers 是一个元素为二元组的列表，例如[('Content-Type', 'text/plain;charset=utf-8')]；
exc_info 在错误处理的时候使用

start_response 返回的是 Response Header。APP 返回的可迭代对象是 Response Body。所以要在返回可迭代对象之前调用，先把 Response Header 发过去，之后发 Response Body：

返回网页的例子¶

from wsgiref.simple_server import make_server

# APP 应用程序
def application(environ, start_response):
    status = '200 OK'
    headers = [('Content-Type', 'text/html;charset=utf-8')]
    start_response(status, headers) # 先返回 Response Header`
    html = '<h1>欢迎你</h1>'.encode("utf-8")
    return [html]   # 再返回可迭代对象 Response Body

ip = '127.0.0.1'
port = 9999
server = make_server(ip, port, application)
server.serve_forever()
# server.handle_request() 这个是运行一次

WSGI 总结¶

WEB 服务器：

本质上就是一个 TCP 服务器，监听在特定端口上；
支持 HTTP 协议，能够将 HTTP 请求报文进行解析，能够把响应数据进行HTTP 协议的报文封装并返回浏览器端；
实现了 WSGI 协议，该协议约定了和应用程序之间接口。

APP 应用程序：

遵从 WSGI 协议；
本身是一个可调用对象；
调用 start_response，返回响应头部；
返回包含正文的可迭代对象。

2. 使用 webob 库解析 environ¶

环境数据有很多，都是存在字典中的，字典的存取方式没有对象的属性访问方便。使用第三方库 webob，可以把环境数据的解析、封装成对象。

1	~]$ pip install webob

webob 中使用了 MultiDict，MultiDict 允许一个 key 存放多个值。

webob.Request 对象¶

将环境参数解析并封装成 request 对象.

request = webob.Request(environ)
print(request.headers) # 类字典容器
print(request.method)
print(request.path)
print(request.query_string) # 查询字符串
print(request.GET) # GET 方法中的所有数据
print(request.POST) # POST 方法中的所有数据
print(request.params) # 所有数据，参数

不关心什么方法提交，只关心数据，可以使用 request.params，它里面是所有提交数据的封装。

webob.Response 对象¶

res = webob.Response()
print(res.status)   # 状态码
print(res.headerlist)   # head 信息，默认 [('Content-Type', 'text/html; charset=UTF-8'), ('Content-Length', '0')]
start_response(res.status, res.headerlist)
# 返回可迭代对象
html = '<h1>欢迎你</h1>'.encode("utf-8")
return [html]

webob.Response 还实现了 `call` 方法¶

先看 webob.Response 类的源码：

def __call__(self, environ, start_response):
    ...
    start_response(self.status, headerlist) # 返回 head 信息
    ...
    return self._app_iter # 返回可迭代对象

由此可以这样使用：

def application(environ:dict, start_response):
    # 请求处理
    request = webob.Request(environ)

    # 响应处理
    res = webob.Response() # [('Content-Type', 'text/html; charset=UTF-8'), ('Content-Length', '0')]
    res.status_code = 200 # 默认200 print(res.content_type)
    html = '<h1>欢迎你</h1>'.encode("utf-8")
    res.body = html
    return res(environ, start_response)

webob.dec 装饰器¶

webob 还可以作为装饰器使用。

先看下面的 app，以函数举例，以后就可以写成这样的形式，只接收一个参数，类型是 webob.Request。显然这并不符合 WSGI 接口定义的要能接收 environ 和 start_response 两参数，怎么办呢？经过 @wsgify 这个装饰器一包装，返回的一定是可以接收两参数，符合 WSGI 标准的。

这样的好处是 @wsgify 装饰器会为你传入 webob.Request 参数，它包括了 environ 中请求信息，并以对象形式包装了，以后就可以方便的通过 request.xxx 的方式直接获取请求中的某个值。而响应值也可以方便的直接 return 一个 webob.Response 类型的对象就可以了，这样可以专注的在写 APP 上:

from webob.dec import wsgify

@wsgify
def app(request:webob.Request) -> webob.Response:
    # 可以这样用
    res = webob.Response('<h1>欢迎你a </h1>')

    # 也可以这样
    res.status_code = 202
    res.body = 'hello MOTO'.encode()

    # 可以是 bytes
    my_bytes = b'<h1>hello</h1>'

    # 可以是个字符串
    my_str = '<h1>欢迎你</h1>'

    return res
    # return my_str
    # return my_bytes

wsgify 装饰器装饰的函数应该具有一个参数，这个参数是 webob.Request 类型，是对字典 environ 的对象化后的实例。

返回值
可以是一个 webob.Response 类型实例
可以是一个 bytes 类型实例，它会被封装成 webob.Response 类型实例的 body 属性
可以是一个字符串类型实例，它会被转换成 bytes 类型实例，然后会被封装成 webob.Response 类型实例的 body 属性
总之，返回值会被封装成 webob.Response 类型实例返回

由此之前的代码可以修改为：

from wsgiref.simple_server import make_server
import webob
from webob.dec import wsgify


@wsgify
def app(request: webob.Request) -> webob.Response:
    print(request.method)
    print(request.path)
    print(request.query_string)
    print(request.GET)
    print(request.POST)
    print('params = {}'.format(request.params))
    res = webob.Response('<h1>欢迎你</h1>')
    return res


if __name__ == '__main__':
    ip = '127.0.0.1'
    port = 9999
    server = make_server(ip, port, app)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        server.shutdown()
        server.server_close()

写成类就是这样：

class App:
    @wsgify
    def __call__(self, request:Request):
        return '<h1>欢迎你</h1>'

十一、数据库基础¶

SQL 语句分为：
- DDL：数据定义语言，负责数据库定义、数据库对象定义，由 CREATE、ALTER 与 DROP 三个语法所组成
- DML：数据操作语言，负责对数据库对象的操作，CRUD 增删改查
- DCL：数据控制语言，负责数据库权限访问控制，由 GRANT 和 REVOKE 两个指令组成
- TCL：事务控制语言，负责处理 ACID 事务，支持 commit、rollback 指令

1. DCL¶

GRANT 授权
REVOKE 撤销

1 2	GRANT ALL ON employees.* TO 'monster'@'%' IDENTIFIED by 'monster'; REVOKE ALL ON . FROM monster;

2. DDL¶

删除用户(慎用)！

1	DROP USER monster;

创建数据库¶

1	CREATE DATABASE IF NOT EXISTS gogs CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

CHARACTER SET 指定字符集。utf8mb4 是 utf8 的扩展，支持 4 字节 utf8mb4，需要 MySQL5.5.3+。
COLLATE 指定字符集的校对规则，用来做字符串的比较的。例如 a、A 谁大?

删除数据库¶

1	DROP DATABASE IF EXISTS gogs;

创建表¶

 CREATE TABLE `employees` (
  `emp_no` int(11) NOT NULL,
  `birth_date` date NOT NULL,
  `first_name` varchar(14) NOT NULL,
  `last_name` varchar(16) NOT NULL,
  `gender` enum('M','F') NOT NULL,
  `hire_date` date NOT NULL,
  PRIMARY KEY (`emp_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

反引号标注的名称，被认为是非关键字。

查看列信息¶

DESC

1	{DESCRIBE \| DESC} tbl_name [col_name \| wild]

Index：索引，可以看做是一本字典的目录，为了快速检索用的。空间换时间，显著提高查询效率。
- 主键索引：主键会自动建立主键索引，主键本身就是为了快速定位唯一记录的。
- 唯一索引：表中的索引列组成的索引必须唯一，但可以为空，非空值必须唯一
- 普通索引：没有唯一性的要求，就是建了一个字典的目录而已。
Constraint：约束
- UNIQUE：唯一键约束
- PRIMARY KEY：主键约束
- Foreign Key：外键约束
视图：也称虚表，看起来像表。它是由查询语句生成的。可以通过视图进行 CRUD 操作

3. DML – CRUD 增删改查¶

Insert 语句¶

向表中插入一行数据。自增字段、缺省值字段、允许为空的字段可以不写。

1	INSERT INTO table_name (col_name,...) VALUES (value1,...);

将 select 查询的结果插入到表中。

1	INSERT INTO table_name SELECT ... ;

如果主键冲突、唯一键冲突就执行 update 后的设置。

这条语句的意思就是，如主键不在，则新增记录；如主键在，就更新部分字段。

1	INSERT INTO table_name (col_name1,...) VALUES (value1,...) ON DUPLICATE KEY UPDATE col_name1=value1,...;

如果主键冲突、唯一键冲突就忽略错误，返回一个警告。

1	INSERT IGNORE INTO table_name (col_name,...) VALUES (value1,...);

Update 语句¶

UPDATE [IGNORE] tbl_name
  SET col_name1=expr1 [, col_name2=expr2 ...]
  [WHERE where_definition]

UPDATE reg SET name='张三' WHERE id=5;

IGNORE 的意义同 Insert 中的 IGNOER 语句

Delete 语句¶

1 2	DELETE [IGNORE] FROM tbl_name [WHERE where_definition] -- 删除符合条件的记录

Select 语句¶

语法¶

SELECT
    [DISTINCT]
    select_expr, ...
    [FROM table_references
    [WHERE where_definition]
    [GROUP BY {col_name | expr | position}
        [ASC | DESC], ... [《 WITH ROLLUP]]
    [HAVING where_definition]
    [ORDER BY {col_name | expr | position}
        [ASC | DESC] , ...]
    [LIMIT {[offset,] row_count | row_count OFFSET offset}]
    [FOR UPDATE | LOCK IN SHARE MODE]]

FOR UPDATE 会把行进行写锁定，这是排它锁。

查询¶

-- 数字合并
SELECT emp_no, first_name + last_name FROM employees;
-- 字符串合并
SELECT emp_no, CONCAT(first_name,' ',last_name) FROM employees;

-- AS 定义别名，可选。写 AS 是一个好习惯
SELECT emp_no as `no`, CONCAT(first_name,' ',last_name) name FROM employees emp;

Limit 子句¶

-- 返回 5 条记录
SELECT * FROM employees emp LIMIT 5;

-- 先偏移 18 条后，返回 5 条记录，
SELECT * FROM employees emp LIMIT 5 OFFSET 18;
-- 等效语法，但是推荐使用第 1 种
SELECT * FROM employees emp LIMIT 18, 5;

Where 子句¶

运算符	描述
=	等于
<>	不等于
>、<、>=、<=	大于、小于、大于等于、小于等于
BETWEEN	在某个范围之内，between a and b等价于[a, b]
LIKE	字符串模式匹配，%表示任意多个字符，_表示一个字符
IN	指定针对某个列的多个可能值
AND	与
OR	或

注意：如果很多表达式需要使用 AND、OR 计算逻辑表达式的值的时候，由于有结合律的问题，建议使用小括号来避免产生错误。

-- 条件查询
SELECT * FROM employees WHERE emp_no < 10015 and last_name LIKE 'P%';
SELECT * FROM employees WHERE emp_no BETWEEN 10010 AND 10015 AND last_name LIKE 'P%';
SELECT * FROM employees WHERE emp_no in (10001, 10002, 10010);

Order by 子句¶

对查询结果进行排序，可以升序 ASC、降序 DESC。

1 2	-- 降序 SELECT * FROM employees WHERE emp_no in (10001, 10002, 10010) ORDER BY emp_no DESC;

使用 DISTINCT 不返回重复记录

1
2
3

-- DISTINCT使用
SELECT DISTINCT dept_no from dept_emp;
SELECT DISTINCT emp_no from dept_emp;

COUNT、AVG、MIN、SUM 等聚合函数¶

描述	函数
COUNT(expr)	返回记录中记录的数目，如果指定列，则返回非 NULL 值的行数
COUNT(DISTINCT expr,[expr…])	返回不重复的非 NULL 值的行数
AVG([DISTINCT] expr)	返回平均值，返回不同值的平均值
MIN(expr), MAX(expr)	最小值，最大值
SUM([DISTINCT] expr)	求和，Distinct 返回不同值求和

1 2	-- 聚合函数 SELECT COUNT(*), AVG(emp_no), SUM(emp_no), MIN(emp_no), MAX(emp_no) FROM employees;

Group by 和 Having 分组查询¶

使用 Group by 子句，如果有条件，使用 Having 子句过滤分组、聚合过的结果。

-- 聚合所有
SELECT emp_no, SUM(salary), AVG(salary), COUNT(emp_no) from salaries;
-- 聚合被选择的记录
SELECT emp_no, SUM(salary), AVG(salary), COUNT(emp_no) from salaries WHERE emp_no < 10003;

-- 按照不同 emp_no 分组，每组分别聚合
SELECT emp_no, SUM(salary), AVG(salary), COUNT(emp_no) from salaries WHERE emp_no < 10003 GROUP BY emp_no;

-- HAVING 子句对分组结果过滤
SELECT emp_no, SUM(salary), AVG(salary), COUNT(emp_no) from salaries GROUP BY emp_no HAVING AVG(salary) > 45000;
-- 使用别名
SELECT emp_no, SUM(salary), AVG(salary) AS sal_avg, COUNT(emp_no) from salaries GROUP BY emp_no HAVING sal_avg > 60000;

-- 最后对分组过滤后的结果排序
SELECT emp_no, SUM(salary), AVG(salary) AS sal_avg, COUNT(emp_no) from salaries GROUP BY emp_no HAVING sal_avg > 60000 ORDER BY sal_avg;
--                                     ↑↑↑                         ↑↑↑              ↑↑↑             ↑↑↑                    ↑↑↑
-- SQL 语句的执行顺序：          3、FROM 之前的语句(投影)        1、FROM...WHERE(选择)   2、GROUPBY(分组)   4、HAVING(过滤)     5、ORDER BY(排序)

子查询¶

查询语句可以嵌套，内部查询就是子查询。
子查询必须在一组小括号中。
子查询中不能使用 Order by。


-- 子查询
SELECT * FROM employees WHERE emp_no in (SELECT emp_no from employees WHERE emp_no > 10015) ORDER BY emp_no DESC;

SELECT emp.emp_no, emp.first_name, gender FROM (SELECT * from employees WHERE emp_no > 10015) AS emp WHERE emp.emp_no < 10019 ORDER BY emp_no DESC;

连接 Join¶

交叉连接：cross join

笛卡尔乘积，全部交叉
在 MySQL 中，CROSS JOIN 从语法上说与 INNER JOIN 等同

内连接：inner join，省略为 join。
等值连接：只选某些 field 相等的元组(行)，使用 On 限定关联的结果。
自然连接：特殊的等值连接，会去掉重复的列。用的少。

-- 内连接，笛卡尔乘积 800行
SELECT * from employees JOIN salaries;
SELECT * from employees INNER JOIN salaries;

-- ON 等值连接 40行
SELECT * from employees JOIN salaries ON employees.emp_no = salaries.emp_no;

-- 自然连接，去掉了重复列，且自行使用 employees.emp_no = salaries.emp_no 的条件
SELECT * from employees NATURAL JOIN salaries;

外连接：outer join，分为：

左外连接，即左连接（左边是主表，从左往右看；左边不加限制）
右外连接，即右连接、全外连接

看表的数据的方向，谁是主表，谁的所有数据都显示

INNER JOIN: 如果表中有至少一个匹配，则返回行；
LEFT JOIN: 即使右表中没有匹配，也从左表返回所有的行；
RIGHT JOIN: 即使左表中没有匹配，也从右表返回所有的行；
FULL JOIN: 只要其中一个表中存在匹配，则返回行。

-- 左连接
SELECT * from employees LEFT JOIN salaries ON employees.emp_no = salaries.emp_no;

-- 右连接
SELECT * from employees RIGHT JOIN salaries ON employees.emp_no = salaries.emp_no;

-- 这个右连接等价于上面的左连接
SELECT * from salaries RIGHT JOIN employees ON employees.emp_no = salaries.emp_no;

4. 事务 Transaction¶

ACID 特性	描述
原子性(atomicity)：	要求事务中的所有操作，不可分割，不能做了一部分操作，还剩一部分操作；
一致性(consistency)：	多个事务并行执行的结果，应该和事务排队执行的结果一致。如果事务的并行执行和多线程读写共享资源一样不可预料，就不能保证一致性；
隔离性(isolation)：	就是指多个事务访问共同的数据了，应该互不干。隔离性，指的就是在一个事务处理期间，其他事务能不能访问的问题
持久性(durability)：	比较好理解，就是事务提交后，数据不能丢失。

在上述四个特性中，如果 隔离性 不好，事务的操作就会互相影响，带来不同严重程度的后果:

更新丢失 Lost Update：事务 A 和 B，更新同一个数据，它们都读取了初始值100，A 要减 10，B 要加 100，A 减去 10 后更新为 90，B 加 100 后更新为 200，A 就是更新丢失了，就像从来没有减过 10 一样。
脏读：事务 A 和 B，事务 B 读取到了事务 A 未提交的数据(这个数据可能是一个中间值，也可能事务 A 后来回滚事务)。事务 A 是否最后提交并不关心。只要读取到了这个被修改的数据就是脏读。
不可重复读 Unrepeatable read：事务 A 在事务执行中相同查询语句，得到了不同的结果，不能保证同一条查询语句重复读相同的结果就是不可以重复读。
例如，事务 A 查询了一次后，事务 B 修改了数据，事务 A 又查询了一次，发现数据不一致了。
- 注意，脏读讲的是可以读到相同的数据的，但是读取的是一个未提交的数据，不是提交的最终结果。
幻读 Phantom read：事务 A 中同一个查询要进行多次，事务 B 插入数据，导致 A 返回不同的结果集，如同幻觉，就是幻读。数据集有记录增加了，可以看做是增加了记录的不可重复读。

为了解决上述问题，提出了隔离级别，隔离级别由低到高，如下表：

隔离级别越高，串行化越高，数据库执行效率越低。当前事务处理的中间结果对其它事务不可见程度越高。

隔离级别	描述
READ UNCOMMITTED	读取到未提交的数据
READ COMMITTED	读已经提交的数据，ORACLE 默认隔离级别
REPEATABLE READ	可以重复读，MySQL 的默认隔离级别。
SERIALIZABLE	可串行化。事务间完全隔离，事务不能并发，只能串行执行

隔离相关语法¶

-- 设置会话级或者全局隔离级别
SET [SESSION | GLOBAL] TRANSACTION ISOLATION LEVEL
{READ UNCOMMITTED | READ COMMITTED | REPEATABLE READ | SERIALIZABLE}

-- 示例
SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
SET GLOBAL TRANSACTION ISOLATION LEVEL REPEATABLE READ;

-- 查询隔离级别
SELECT @@global.tx_isolation;
SELECT @@tx_isolation;

事务相关语法¶

-- 开始一个事务，
START TRANSACTION 或 BEGINSTART

-- 提交事务后，变更成为永久变更
COMMIT

-- 回滚变更，在提交事务之前，事务中的操作就如同没有发生过一样(原子性)。
ROLLBACK

-- 用于当前连接禁用或启用默认的 autocommit 模式。如果开启自动提交，如果有一个修改表的语句执行后，会立即把更新存储到磁盘。
SET AUTOCOMMIT = 0 | 1

5. 数据库和数据仓库的区别¶

本质上来说没有区别，都是存放数据的地方。但是：

数据库	数据仓库
关注数据的持久化、数据的关系，为业务系统提供支持，事务支持	为了分析或者发掘而设计的表结构，可以存储海量数据
存储在线交易数据 OLTP（联机事务处理 On-line Transaction Processing）	数据仓库存储历史数据，用于 OLAP （联机分析处理 On-Line Analytical Processing）
数据库支持在线业务，需要频繁增删改查	数据仓库一般囤积历史数据支持用于分析的 SQL，一般不建议删改

十二、数据库开发¶

MySQL 的驱动：

MySQLdb：最有名的库，对 MySQL 的 Client 封装实现，支持Python 2，不更新了，不支持 Python3；
MySQL 官方 Connector；
pymysql：语法兼容 MySQLdb，使用 Python 写的库，支持 Python3。

1. pymysql 的使用¶

连接 Connect¶

连接数据库必须建立一个传输数据的通道，使用 pymysql.connect() 方法：

import pymysql

try:
    # 首先，必须建立一个传输数据的通道。默认 Autocommit 是关闭的
    conn = pymysql.connect('host_ip', 'hostname', 'password', 'database')

    # ping() 方法，测试数据库服务器是否活着。有一个参数 reconnect 表示断开与服务器连接是否重连。
    print(conn.ping(reconnect=False))
finally:
    if conn:
        conn.close()
        print(conn.ping(reconnect=False))

操作数据库¶

操作数据库，必须使用 游标(Cursor)，需要先获取一个游标对象；

# 1. 返回一个新的游标对象。 连接没有关闭前，游标对象可以反复使用。
cursor = Connection.cursor(cursor=None)  # cursor 参数可以指定一个 Cursor 类。如果为 None，则使用默认 Cursor 类。

# 2. 数据库操作需要使用 Cursor 类的实例，提供 execute() 方法，执行 SQL 语句，成功返回影响的行数。
cursor.execute("insert into student (name,age) values('tom',20)")

事务管理¶

Connection 类有三个方法：

begin 开始事务；
commit 将变更提交；
rollback 回滚事务。

import pymysql

    conn = pymysql.connect('host_ip', 'hostname', 'password', 'school')
    cursor = conn.cursor()
try:
    insert_sql = "insert into student (name,age) values('tom',20)"
    rows = cursor.execute(insert_sql)
    print(rows)
    conn.commit()
except:
    conn.rollback()
finally:
    if cursor:
        cursor.close()
    if conn:
        conn.close()

结果集查询¶

Cursor 类的获取查询结果集的方法有（如果走到末尾，就返回空元组）：

fetchone()：获取结果集的下一行；
fetchmany(size=None)：size 指定返回的行数的行，None 则返回空元组；
fetchall()：获取所有行；
cursor.rownumber：返回当前行号，可以修改，支持负数；
cursor.rowcount：返回的总行数。

注意：fetch 操作的是结果集，结果集是保存在客户端的，也就是说 fetch 的时候，查询已经结束了。

带列名查询¶

Cursor 类有一个 Mixin 的子类 DictCursor。

只需要 cursor = conn.cursor(DictCursor) 就可以了。

1
2
3

# 返回结果
{'name': 'tom', 'age': 20, 'id': 4}
{'name': 'tom0', 'age': 20, 'id': 5}

SQL 注入攻击¶

现有功能如下，通过用户输入的 userid 来查询：

1	sql = f'SELECT * from student WHERE id = {userid}'

可是，如果用户提交的是 '5 or 1=1'呢？就成了：

1	sql = 'SELECT * from student WHERE id = 5 or 1=1''

运行的结果竟然是返回了整张表。

如何解决注入攻击?

使用 参数化 查询，可以有效防止注入攻击，并提高查询的效率。

# 语法：
Cursor.execute(sql, args=None)  # `args` 必须是元组、列表或字典。如果查询字符串使用 `%(name)s`，就必须使用字典。

# 示例：
cursor.execute('SELECT * from student WHERE id = %s', (userid,)) # 参数化查询

cursor.execute('SELECT * from student WHERE id = %(id)s', {'id':'5'}) # 参数化查询

参数化查询为什么提高效率？

原因就是 – SQL 语句缓存
数据库服务器一般会对 SQL 语句编译和缓存，编译只对 SQL 语句部分，所以参数中就算有 SQL 指令也不会被执行。
编译过程，需要词法分析、语法分析、生成 AST、优化、生成执行计划等过程，比较耗费资源。
服务端会先查找是否对同一条查询语句进行了缓存，如果缓存未失效，则不需要再次编译，从而降低了编译的成本，降低了内存消耗。
可以认为 SQL 语句字符串就是一个 key，如果使用拼接方案，每次发过去的 SQL 语句都不一样，都需要编译并缓存。
大量查询的时候，首选使用参数化查询，以节省资源。
开发时，应该使用参数化查询。

上下文支持¶

class Connection(object):
    def __enter__(self):
        """Context manager that returns a Cursor"""
        return self.cursor()

    def __exit__(self, exc, value, traceback):
        """On successful exit, commit. On exception, rollback"""
        if exc:
            self.rollback()
        else:
            self.commit()

2. SQLAlchemy 的使用¶

SQLAlchemy 是一个功能强大的 ORM 模型的著名实现，内部使用了连接池。

底层支持使用 pymysql 库，也可以使用 mysqldb 库，不过后者已经不更新了且不支持 python3。

SQLAlchemy 中大量使用了元编程，关于元编程的示例，查看 py 文件 mate_class.py。

(1) 创建连接¶

数据库连接的事情，交给引擎

import sqlalchemy

# pymysql 的连接格式：推荐使用
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
engine = sqlalchemy.create_engine("mysql+pymysql://root:123qwe@127.0.0.1:3306/test", echo=True)
# echo=True 引擎是否打印执行的语句，调试的时候建议打开。


# mysqldb 的连接，不支持 Python3
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
engine = sqlalchemy.create_engine("mysql+pymysql://root:123qwe@127.0.0.1:3306/test")

(2) Declare a Mapping 创建映射¶

创建基类：

from sqlalchemy.ext.declarative import declarative_base

# 创建基类（这是一个元类），用于实体类继承
Base = declarative_base()

创建实体类：

student 表：

CREATE TABLE student (
    id INTEGER NOT NULL AUTO_INCREMENT,
    name VARCHAR(64) NOT NULL,
    age INTEGER,
    PRIMARY KEY (id)
)

对应的 python 代码：

from sqlalchemy import Column, Integer, String

# 创建实体类，必须从 Base 类继承
class Student(Base):
    # 指定表名
    __tablename__ = 'student'

    # 定义属性对应字段
    id = Column(Integer, primary_key=True)
    name = Column(String(64))
    age = Column(Integer)
    # Column()内的第一参数是字段名，如果省略则使用属性名。如果和属性名不一致，一定要指定 # age = Column('age', Integer)

    def __repr__(self):
        return "{} id={} name={} age={}".format(
            self.__class__.__name__, self.id, self.name, self.age)

(3) 使用 SQLAlchemy 创建表¶

可以使用 SQLAlchemy 来创建、删除表。
生产环境很少这样创建表，都是系统上线的时候由脚本生成。
生成环境很少删除表，宁可废弃都不能删除。

# 删除继承自 Base 的所有表
Base.metadata.drop_all(engine)

# 创建继承自 Base 的所有表
Base.metadata.create_all(engine)

(4) 创建会话 session¶

在一个会话中操作数据库，会话建立在连接上，连接被引擎管理。

1
2
3

# 创建session
Session = sessionmaker(bind=engine) # 返回类
session = Session() # 实例化

session 对象线程不安全。所以不同线程要使用不用的 session 对象。
而 Session 类和 engine 都是线程安全的，有一个就行了。

(5) CRUD 操作¶

先从基类实例化，等于一条记录：

s = Student(name='tom')
print(s.name)

s.age = 20
print(s.age)

增¶

1
2
3

session.add(s)      # 添加一个
session.commit()     # commit 事务
session.add_all([s])  # 从指定的可迭代容器添加

查（简单查\无条件查询\无 where）¶

使用 query() 方法：将要查询的实体的类传入后，返回可迭代的 Query 对象，这时候并不去数据库查询。当迭代这个对象时，才执行 SQL 来查询数据库，并把查询结果封装到指定类的实例。
使用 get 方法：通过主键查询，直接执行 SQL 查询，会返回一条指定类的一个实例。

# 无条件查询
students = session.query(Student)
for student in students:
    print(student)
    print(type(student))
print('~~~~~~~~~~~~~')

# 通过主键查询
student = session.query(Student).get(2)

改¶

# 先查回来，修改后，再提交更改。
student = session.query(Student).get(2)
student.name = 'sam'
student.age = 30
session.add(student)
session.commit()

删¶

1 2	# s 对象必须是 persistent 状态 session.delete(s)

flush 方法¶

flush() 方法，主动把改变从 python 应用到数据库中去，但还是在同一个事务内，还未 commit。

(6) 实体的状态属性¶

每一个实体，都有一个状态属性 _sa_instance_state，其类型是sqlalchemy.orm.state.InstanceState，可以使用 sqlalchemy.inspect(entity) 函数查看状态。

常见的状态值有：transient、pending、persistent、deleted、detached。

状态	说明
transient	实体类刚刚创建出来，尚未加入到 session 中。
pending	状态为 transient 的实体被 `add()` 到 session 中时，状态会切换到 pending，但它还没有 `flush` 到数据库中。
persistent	这个状态表示这个实体对象，对应着数据库中真实的记录。当 pending 状态在提交成功后可以变成 persistent状态，或者查询成功返回的实体也是 persistent 状态。
deleted	实体被删除且已经 `flush` 但未 `commit` 完成。当事务提交成功后，实体变成 detached，如事务失败，会回到 persistent 状态
detached	实体删除成功后进入这个状态

举例：

transient：
- 新建一个实体还未 add()，状态是 transient 临时的。
pending：
- 一旦 add() 后，就从 transient 变成 pending 状态。
persistent：
- 成功 commit() 后，就从 pending 变成 persistent 状态；
- 或者查询成功后返回的实体对象，也是 persistent 状态；
- persistent 状态的实体，修改后依然是 persistent 状态。
deleted：
- 把 persistent 状态的实体 delete() 后并 flush 了，但没有 commit，就变成 deteled 状态。
detached：
- 若之后成功 commit 了，则变为 detached 状态；若提交失败，则还原到 persistent 状态。

删除、修改，需要对应一个真实的记录，所以要求实体对象是 persistent 状态。

(7) 复杂查询¶

使用 filter() 可以定义查询条件。

演示环境：以下库使用测试数据 test.sql 做基础。

实体类¶

基础通用代码，查询的演示以此为基础：

import sqlalchemy
from sqlalchemy import Column, Integer, String, Date, Enum, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declarative_base  # 用于创建基类，便于实体类继承
from sqlalchemy.orm import sessionmaker
import enum

# 创建连接 engine
conn_str = "{}://{}:{}@{}:{}/{}".format('mysql+pymysql', 'root', '123qwe', '192.168.50.30', 3306, 'test')
engine = sqlalchemy.create_engine(conn_str, echo=True)

# 创建基类
Base = declarative_base()

# 创建会话 session
Session = sessionmaker(bind=engine)
session = Session()


# 打印函数
def show(emps):
    for x in emps:
        print(x)
    print('~~~~~~~~~~~~~~~', end='\n\n')


# 用于性别中使用的枚举类型，不常用，而且可能有问题。生产中建议使用短 int 类型，用数字 0、1、2 来表示（要开放）
class MyEnum(enum.Enum):
    M = 'M'
    F = 'F'


# 创建实体类，必须从 Base 类继承
class Employee(Base):
    # 指定表名
    __tablename__ = 'employees'

    # 定义属性对应字段：Column() 内的第一参数是字段名，如果省略则使用属性名。如果和属性名不一致，一定要指定 # age = Column('age', Integer)
    emp_no = Column(Integer, primary_key=True)
    birth_date = Column(Date, nullable=False)
    first_name = Column(String(14), nullable=False)
    last_name = Column(String(16), nullable=False)
    gender = Column(Enum(MyEnum), nullable=False)
    hire_date = Column(Date, nullable=False)

    def __repr__(self):
        return "{} no={} name={} {} gender={}".format(
            self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender.value)

简单的条件查询（where）¶

1 2	emps = session.query(Employee).filter(Employee.emp_no > 10015) show(emps)

与或非¶

1	from sqlalchemy import or_, and_, not_

比较推荐使用运算符的方式。

# ======== AND 条件 ========
# 第一种写法
emps = session.query(Employee).filter(Employee.emp_no > 10015).filter(Employee.gender == MyEnum.F)
show(emps)
# 第二种写法
emps = session.query(Employee).filter(and_(Employee.emp_no > 10015, Employee.gender == MyEnum.M))
show(emps)
# 第三种写法    ---- 推荐
emps = session.query(Employee).filter((Employee.emp_no > 10015) & (Employee.gender == MyEnum.M))
                                                # 一定要注意 & 符号两边的表达式都要加括号
show(emps)
        # 等效 SQL：WHERE employees.emp_no > %(emp_no_1)s AND employees.gender = %(gender_1)s


# ======== OR 条件 ========
# 第一种写法    ---- 推荐
emps = session.query(Employee).filter((Employee.emp_no > 10018) | (Employee.emp_no < 10003))
show(emps)
# 第二种写法
emps = session.query(Employee).filter(or_(Employee.emp_no > 10018, Employee.emp_no < 10003))
show(emps)
        # 等效 SQL：WHERE employees.emp_no > %(emp_no_1)s OR employees.emp_no < %(emp_no_2)s


# ======== Not 条件 ========
# 第一种写法
emps = session.query(Employee).filter(not_(Employee.emp_no < 10018))
show(emps)
# 第二种写法    ---- 推荐
emps = session.query(Employee).filter(~(Employee.emp_no < 10018))
                            # 一定注意 ~ 后要加括号
show(emps)
        # 等效 SQL：WHERE employees.emp_no >= %(emp_no_1)s

总之，与或非的运算符 &、|、~，一定要在表达式上加上括号，否则因为运算符的优先级得到的结果可能不是你想要的

in、not in、like¶

# ======== in ========
emplist = [10010, 10015, 10018]
emps = session.query(Employee).filter(Employee.emp_no.in_(emplist))
show(emps)
        # 等效 SQL：WHERE employees.emp_no IN (%(emp_no_1)s, %(emp_no_2)s, %(emp_no_3)s)

# ========# not in ========
emplist = [10010, 10015, 10018]
emps = session.query(Employee).filter(~Employee.emp_no.in_(emplist))
show(emps)
        # 等效 SQL：WHERE employees.emp_no NOT IN (%(emp_no_1)s, %(emp_no_2)s, %(emp_no_3)s)

# ======== like ========
emps = session.query(Employee).filter(Employee.last_name.like('P%'))    # P 开头后边跟任意
show(emps)
        # 等效 SQL：WHERE employees.last_name LIKE %(last_name_1)s

一个子查询的使用示例¶

# 获取入度为 0 的点    这样使用子查询只需要去数据库查一次
zds = vertexes.filter(Vertex.id.notin_(db.session.query(Edge.head).filter(Edge.graph_id == graph.id))).all()
# 等价 SQL：子查询的方式
SELECT * FROM vertex WHERE graph_id =1 AND vertex.id NOT IN (SELECT edge.head from edge WHERE graph_id =1)

排序（order_by）¶

# ======== 升序 ========
# 第一种，使用默认升序
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no)
        # 等效 SQL：WHERE employees.emp_no > %(emp_no_1)s ORDER BY employees.emp_no
# 第二种，显示要求使用升序 asc，效果和第一种一样
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.asc())
show(emps)
        # 等效 SQL：WHERE employees.emp_no > %(emp_no_1)s ORDER BY employees.emp_no ASC

# ======== 降序 ========
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.desc())
show(emps)
        # 等效 SQL：WHERE employees.emp_no > %(emp_no_1)s ORDER BY employees.emp_no DESC

# ======== 多列排序 ========
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name).order_by(Employee.emp_no.desc())
show(emps)
        # 等效 SQL：WHERE employees.emp_no > %(emp_no_1)s ORDER BY employees.last_name, employees.emp_no DESC

分页（limit）¶

# 分页
emps = session.query(Employee).limit(4)
show(emps)
        # 等效 SQL：FROM employees LIMIT 4
        # 只取 4 个

emps = session.query(Employee).limit(4).offset(18)
show(emps)
        # 等效 SQL：FROM employees LIMIT 18, 4
        # 先偏移 18 个后，取 4 个

消费者方法¶

消费者方法调用后，Query 对象(可迭代)就转换成了一个容器。

消费者方法包括：

Query.count()：返回查询结果总数；
Query.all()：返回所有查询结果；
Query.first()：只返回第一个；
Query.one()：查询结果有且只能有一个。

# 总行数
emps = session.query(Employee)
print(len(list(emps)))  # 第一种（建议不用）：转换为 list 时会把数据库都查一遍，之后返回大量的结果集，速度慢
        # 等效 SQL：SELECT * FROM employees

print(emps.count())     # 第二种（建议使用）：使用这个时会帮你调用聚合函数 count(*) 来查询，速度快
        # 等效 SQL：SELECT count(*) AS count_1

# 取所有数据
print(emps.all())       # 返回个列表，查不到返回空列表。如果没有条件，等于全查询，会很慢！！！
        # 等于 (list(emps))

# 取首行
print(emps.first())     # 返回首行，查不到返回None
        # 等效 SQL：FROM employees LIMIT 1

# 有且只能有一行
#print(emps.one()) # 如果查询结果是多行抛异常
print(emps.limit(1).one())

# 删除 delete by query
session.query(Employee).filter(Employee.emp_no > 10018).delete()  # 注意：查询到的全部删除！！！
# session.commit() # 提交则删除

聚合、分组（max、min、avg、group_by）¶

# 聚合函数

# count
from sqlalchemy import func
query = session.query(func.count(Employee.emp_no))
print(query.one())      # 只能有一行结果，返回的是一个元组 (20,)
print(query.scalar())   # 取 one() 返回元组的第一个元素 20
        # 等效 SQL：SELECT count(employees.emp_no) AS count_1

# max/min/avg
print(session.query(func.max(Employee.emp_no)).scalar())
        # 等效 SQL：SELECT max(employees.emp_no)
print(session.query(func.min(Employee.emp_no)).scalar())
        # 等效 SQL：SELECT min(employees.emp_no)
print(session.query(func.avg(Employee.emp_no)).scalar())
        # 等效 SQL：SELECT avg(employees.emp_no)

# 分组
query = session.query(Employee.gender, func.count(Employee.emp_no)).group_by(Employee.gender)
query.all()
        # 等效 SQL：SELECT employees.gender AS employees_gender, count(employees.emp_no) AS count_1
        #          FROM employees GROUP BY employees.gender

(8) 关联查询（join）¶

test 库表关系如下：

员工表 employees 的主键字段是员工编号 emp_no
部门表 department 的主键字段是部门编号 dept_no
员工部门对应表 dept_emp 中有 emp_no 和 dept_no 的对应关系，是联合主键，有外键约束，是级联删除

关联查询的演示以此为基础，完整代码：

import sqlalchemy
from sqlalchemy import Column, Integer, String, Date, Enum, ForeignKey
from sqlalchemy.ext.declarative import declarative_base  # 用于创建基类，便于实体类继承
from sqlalchemy.orm import sessionmaker, relationship
import enum

# 创建连接 engine
conn_str = "{}://{}:{}@{}:{}/{}".format('mysql+pymysql', 'root', '123qwe', '192.168.50.30', 3306, 'test')
engine = sqlalchemy.create_engine(conn_str, echo=True)

# 创建基类
Base = declarative_base()

# 创建会话 session
Session = sessionmaker(bind=engine)
session = Session()

# 打印函数
def show(emps):
    for x in emps:
        print(x)
    print('~~~~~~~~~~~~~~~', end='\n\n')

# 用于性别中使用的枚举类型，不常用，而且可能有问题。生产中建议使用短 int 类型，用数字 0、1、2 来表示（要开放）
class MyEnum(enum.Enum):
    M = 'M'
    F = 'F'


# 创建员工表类，必须从 Base 类继承
class Employee(Base):
    # 指定表名
    __tablename__ = 'employees'

    # 定义属性对应字段：Column() 内的第一参数是字段名，如果省略则使用属性名。如果和属性名不一致，一定要指定 # age = Column('age', Integer)
    emp_no = Column(Integer, primary_key=True)
    birth_date = Column(Date, nullable=False)
    first_name = Column(String(14), nullable=False)
    last_name = Column(String(16), nullable=False)
    gender = Column(Enum(MyEnum), nullable=False)
    hire_date = Column(Date, nullable=False)

    # 当使用 join 查询时，使用 relationship 关联另一张表
    # 注意格式，括号内写要关联的类名  ↓
    dept_emps = relationship('Dept_emp')  # ！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！

    def __repr__(self):
        return "{} no={} name={} {} gender={} dept_emps={}".format(
            self.__class__.__name__, self.emp_no, self.first_name, self.last_name,
            self.gender.value, self.dept_emps)      # 这里如果不使用 dept_emps，结果也不会显示 dept_emps 表中的内容


# 创建部门表
class Department(Base):
    __tablename__ = 'departments'

    dept_no = Column(String(4), primary_key=True)
    dept_name = Column(String(40), nullable=False, unique=True)

    def __repr__(self):
        return "{} no={} name={}".format(
            type(self).__name__, self.dept_no, self.dept_name)


# 员工、部门之间的关系表
class Dept_emp(Base):
    __tablename__ = "dept_emp"

    # 定义外键约束，来自员工表和部门表
    emp_no = Column(Integer, ForeignKey('employees.emp_no', ondelete='CASCADE'), primary_key=True)
    dept_no = Column(String(4), ForeignKey('departments.dept_no', ondelete='CASCADE'), primary_key=True)
    from_date = Column(Date, nullable=False)
    to_date = Column(Date, nullable=False)q

    def __repr__(self):
        return "{} empno={} deptno={}".format(
            type(self).__name__, self.emp_no, self.dept_no)

# 查询10010员工所在的部门编号
results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010)
show(results.all())

需求如下：查询 10010 员工的所在的部门编号¶

第 1 种方式：隐式内连接¶

使用隐式内连接的方式（还没有使用 join）

# 查询10010员工所在的部门编号
results = session.query(Employee, Dept_emp).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010)
                        # 同时查两个表                                              # 两个 filter 等于 AND 操作
        # 等效 SQL：FROM employees, dept_emp
        #          WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s
print(results.all())

第 2 种方式：使用 join¶

# 第一种写法，建议不使用，因为如果不指定等值条件，SQLAlchrmy 会自己决定等值条件的字段，这样的结果可能不是你期待的结果
results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == 10010)
print(results.all())

# 第二种写法，建议使用，在 join 中明确指明等值条件                    #  ↓  建议自己明确指定 JOIN ON 时的等值条件
results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010)
print(results.all())
        # 等效 SQL：FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no
        #          WHERE employees.emp_no = 10010

# 第三种，和第二种效果一样，可以使用
results = session.query(Employee).join(Dept_emp, (Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010))
print(results.all())

relationship¶

注意：使用 join 时，要注意是否定义了 relationship，否则上面两种写法，返回的结果会像下面这样：
只有一行 Employee 表中字段的数据，却没有 Dept_emp 表中的字段，为什么？

1	[Employee no=10010 name=Duangkaew Piveteau gender=F]

原因在于 query(Employee) 只能返回一个 Employee 的实体对象，这和 Dept_emp 表本来就一毛钱关系都没有。

为了解决这个问题，需要使用 relationship 在实体类 Employee 中增加属性，用来存放 Dept_emp 表的信息。

语法：sqlalchemy.orm.relationship('实体类名字符串')

所以在 Employee 类中，要有如下属性：

1	dept_emps = relationship('Dept_emp')

这样定义后，当查询时，sqlalchemy 会去执行 2 条 SQL：

SELECT * FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no WHERE employees.emp_no = 10010
SELECT * FROM dept_emp WHERE dept_emp.emp_no = 10010

正确的查询结果：

1	[Employee no=10010 name=Duangkaew Piveteau gender=F dept_emps=[Dept_emp empno=10010 deptno=d004, Dept_emp empno=10010 deptno=d006]]

一对多、多对一¶

从 一查多 和从 多查一 时，从两个方向查，示例：

# 从一查多

class Alice(Base):
    __tablename__ = 'alice'

    id = Column(INTEGER(11), primary_key=True)

    bobs = relationship('Bob', foreign_keys='Bob.a_id')
    # 从一查多时，a.bobs 会查出同一个 a_id 的所有 Bob


# 从多查一
class Bob(Base):
    __tablename__ = 'bob'

    id = Column(INTEGER(11), primary_key=True)
    a_id = Column(ForeignKey('alice.id'), nullable=False)

    alice = relationship('Alice')
    # 从多查一时，b.alice.name 便可以获取 alice 的 name

使用时的技巧：

a = Alice()
db.session.add(a)

b = Bob()

# 由于 a 还未提交，并没有 id，而且这时也不想提交，那么 b.a_id 应该等于什么？怎么办？
# 因为在 model 中定义了外键，所以使用 b.alice 让 SQLALchemy 自己通过 relationship 找映射关系
b.alice = a   # 等价 b.a_id = a.id
db.session.add(b)

db.session.commit()
print(a.id)
print(b.a_id)

使用 join 提高效率¶

# 这样会查多次数据库
query = db.session.query(Track).filter(Track.pipeline_id == p_id).filter(Track.state.in_(state))

pipelines = []
for track in query:     # 不仅会查多次数据库，而且 query 中有多少 track，就会查几次，效率极差
    ret.append((track.pipeline.id, track.pipeline.name, track.pipeline.state,
                track.id, track.state,
                track.vertex.id, track.vertex.name, track.vertex.input, track.vertex.script))
print(pipelines)

# 使用 join 方式，只查一次数据库
pipelines = db.session.query(Pipeline.id, Pipeline.name, Pipeline.state,
                             Track.id, Track.state,
                             Vertex.id, Vertex.name, Vertex.input, Vertex.script) \
    .join(Track, Track.pipeline_id == Pipeline.id) \
    .join(Vertex, Vertex.id == Track.vertex_id) \
    .filter(Track.pipeline_id == p_id) \
    .filter(Track.state.in_(state))  # in 方法，判断是否在 state 内。   state 是个列表
print(pipelines)

ORM 总结¶

在开发中，一般都会采用 ORM 框架，这样就可以使用对象操作表了。

定义表映射的类，使用 Column 的描述器定义类属性，使用 ForeignKey 来定义外键约束。

如果在一个对象中，想查看其它表对应的对象的内容，就要使用 relationship 来定义关系。

一、些零散的函数¶

glob¶

分割字符串 partition 和 split 区别¶

检查对象是否为可调用 callable()¶

@inspect.isXXXXXX：用于检查数据类型¶

@inspect.signature：用于获取函数形参的详细信息¶

@functools.partial 偏函数¶

@functools.total_ordering 补全缺失的比较运算符的装饰器¶

Hashable 测试是否可 hash 值¶

startswith 判断字符串前缀的方法¶

replace 字符串替换¶

脚本模块 subprocess 和 临时文件模块 tempfile¶

IP 地址的约束 ipaddress 库¶

netifaces – 获取主机网络接口、ip 列表¶

获取主机名方式¶

二、小技巧¶

__slots__ 在字典上节省内存¶

未实现和未实现异常的区别¶

运算符重载中的反向方法 __radd__¶

单例（单实例）¶

三、设计原则¶

关于 return¶

关于数据容器类型的选择，list 还是 dict¶

注意！！关于 class 中的 __init__ 要不要写 return¶

四、文件¶

1. 文件 IO 常用操作¶

open¶

file¶

mode 模式¶

文件指针¶

buffering 缓冲区¶

newline 换行符号¶

read¶

readline¶

write¶

close¶

判断¶

2. StringIO 和 BytesIO¶

StringIO¶

BytesIO¶

file-like 对象¶

3. 路径操作¶

4. 文件操作¶

5. os 模块¶

6. shutil 模块¶

copy 复制¶

rm 删除¶

move 移动¶

7. csv、ini 和 configparser 模块¶

8. 序列化与反序列化（pickle 库、Json 和 MessagePack 库）¶

五、正则表达式¶

六、面向对象¶

1. 多继承¶

2. 查看继承的特殊属性和方法有¶

3. 特殊属性¶

4. 魔术方法¶

hash¶

bool¶

可视化¶

运算符重载¶

容器相关¶

可调用对象¶

上下文管理¶

5. 反射(reflaction)相关¶

反射的内键函数¶

反射的魔术方法¶

使用反射，动态创建属性的小技巧¶

6. 描述器(Descriptors)相关¶

七、多线程¶

threading 的属性和方法¶

start 和 run¶

线程安全¶

logging 日志输出¶

daemon 和 non-daemon¶

join¶

threading.local 类¶

Timer 定时器/延迟执行¶

Event¶

Lock 排他锁/互斥锁¶

Rlock 可重入锁¶

`@inspect.isXXXXXX`：用于检查数据类型¶

`@inspect.signature`：用于获取函数形参的详细信息¶

`@functools.partial` 偏函数¶

`@functools.total_ordering` 补全缺失的比较运算符的装饰器¶

`startswith` 判断字符串前缀的方法¶

`replace` 字符串替换¶

脚本模块 `subprocess` 和临时文件模块 `tempfile`¶

`slots` 在字典上节省内存¶

运算符重载中的反向方法 `radd`¶

关于数据容器类型的选择，`list` 还是 `dict`¶

注意！！关于 class 中的 `init` 要不要写 `return`¶

`threading` 的属性和方法¶

`start` 和 `run`¶

`logging` 日志输出¶

`daemon` 和 `non-daemon`¶

`join`¶

`threading.local` 类¶

`Timer` 定时器/延迟执行¶

`Event`¶

`Lock` 排他锁/互斥锁¶

`Rlock` 可重入锁¶

`Condition` 生产者、消费者¶

`Barrier` 屏障/栅栏¶

`Semaphore` 信号量¶

`BoundedSemaphore` 受限信号量¶

webob.Response 还实现了 `call` 方法¶