python

modules

关于命名空间和作用域

要点:

• 作用域是静态的，就是一段代码文本区域。

• 命名空间是抽象的，是运行时被创建的。

• 作用域本质就是命名空间的访问权限。

• 命名空间大致分为：

  • 函数的局部命名空间
  • 模块的全局命名空间（这个全局很有误导性，不是真全局，而是相当于模块内部的比如函数，类来说的全局）
  • 内置命名空间

• 对于作用域来说，任何一个位置的作用域，可以最多访问4层命名空间，按顺序分别为：

  1. 当前所处的局部命名空间
  2. 闭包的局部命名空间
  3. 模块的全局命名空间
  4. 内置命名空间（也可以理解为真全局命名空间）

  这也是变量的查找顺序。

关于模块导入

本质就是就是将导入的内容加入到当前模块的命名空间中。要注意命名重复的情况。

import M: 将M: module加入到当前模块的命名空间dict中 from M import a, b, c: 将a: a, b: b, c: c加入到当前模块的命名空间dict中

Note

一个模块被导入时，只会执行一次，不管被导入了几次。

关于__main__模块

当一个py文件或代码被解释器顶层调用执行时（比如以脚本的形式执行python文件），此时实际属于__main__模块。

Note

所以为什么经常看到if __name__ == '__main__'判断，因为这个文件即可以当作脚本用，也可以当作模块被导入。

模块搜索路径

1. 内置模块，即从sys.builtin_module_names中搜索
2. sys.path路径下，查看对应文件 sys.path从下面位置初始化：
   1. 脚本所在目录（如果是链接，则指向真实目录）
   2. PYTHONPATH(可以手动添加路径)
   3. 库文件夹，如site-packages，dist-packages等

Tip

不要和库模块同名，因为脚本所在目录优先级更高，除非你明确要这么做。

关于包

给模块上又包了一层命名空间，可以防止模块重名。

一个目录下必须要有一个__init__.py文件，才会被识别为包(本质就是用这个文件代表这个包，此时包又可以当作模块，和lua中的init.lua效果很像)。

from package import *会发生什么？ 如果存在__init__，这个文件此时就是要导出的模块，和from module import *效果一样。 和常规模块不同在于，可以在__init__中设置一个__all__ list，明确要导出的子模块。

Tip

要明白package的作用，所以不管是import package还是from package import *，都是把package当作模块 这不是一个好的编程实践，不建议这么做。 常规做法依然是from package import module

包的__path__属性可以获取包的路径。

注意包和模块的关系可以用目录和文件类比，但是模块和包并不是必须来自文件系统，比如命名空间包。

import a.b.c,实际上是导入a,a.b,a.b.c三个模块，都会被缓存在sys.modules中，只有当模块对象没有时才会去加载模块。

sys.modules是可写的，所以可以del来强制下次导入重新加载，也可以赋值为None来触发ModuleNotFoundError。

多次执行一个模块代码，得到的对象是不同的，所以如果reload不当，可能出现一个模块多个实例同时存在的情况。

如果要reload，建议使用importlib.reload。

Note

,impoortlib提供了丰富的API来与导入系统交互。

相对导入

相对的是当前模块名，所以主模块不能使用，因为主模块名是__main__。 不是使用./,../而是.,..。(这也是奇葩)

exception

异常

异常没有被捕获，将导致程序终止。

try:
    pass
# 一个或多个except，但是只有一个会被匹配
# 每个except可以有一个或多个Exception
except Exception as e:
    pass
else: # 无异常发生的情况
    pass
# finally一定会被执行，执行在try..except..else语句结束前(try语句中的return,break,continue之前)，不管是正常结束，还是非正常结束
# 没有捕获的异常，以及except和else中的异常会在finally执行后再次抛出，即非正常结束前
# finally中如果有return,break,continue，则相关异常不会再次抛出，程序已正常退出，后面的代码不会执行
finally:
    pass

BaseExecption是异常的基类，Exception是所有非致命异常的基类。 不是Exception的异常通常不需要处理，因为此时程序应该终止。

抛出异常

raise exception # 支持直接抛出异常类，但是本身是调用无参构造函数，所以意义不大

常见的比如只记录不处理，记录后可再次抛出给上层调用者处理：

try:
    pass
except Exception as e:
    logging.error(e)
    raise # 再次抛出

异常链：

在一个except中再次发生异常，这2个异常存在关联，可以使用from字句链接，表明因果关系。

try:
    pass
except Exception as e:
    raise AnotherException(e) from e

会输出异常链上的所有信息，如果想隐藏from-exception，可以使用from None。

自定义异常

异常应该从Exception继承，而不是BaseException。

异常类应该简洁，更多的是一种类型的标记，不应该做额外的工作。

class MyException(Exception):
    def __init__(self, message):
       super().__init__(message)
    def __str__(self):
        return f'MyException: {self.message}'

预定义清理操作

虽然finally很适合做清理，但是过于繁琐，很多对象都定义了标准的清理操作，可以使用with语法糖来自动清理。 就像java中的try-with-resources一样。

with open('file.txt') as f:
    print(f.read())

不再需要在finally中显示close。

异常打包

可以收集多个异常，然后一次raise

errors = [OSError("os error"), ValueError("value error")]
raise ExceptionGroup("two errors occurred", errors)

ExceptionGroup本身也是一个异常，所以可以实现嵌套。

对于打包的异常，可以单独捕获:

try:
    pass
except* OSError:
    # 只处理OSError
except* ValueError:
    # 只处理ValueError

没有被捕获的异常会被打印出来，已被捕获的异常则不会。

在ExceptionGroup嵌套的场景，对于except*来说，异常是扁平的。

补充异常信息

可以添加上下文信息，作为异常的补充描述。

e.add_note("context")

class

基本特征

1. python的类支持多重继承（多基类），且可以访问父类同名方法。
2. 运行时创建，创建后可修改，就像修改普通的对象一样，比如添加属性、方法。
3. 类有独立命名空间，用作局部作用域。
4. 类本身也是对象，本质是类的命名空间的包装器，和普通对象不同的是类还可以被调用（实例化）。
5. 默认定义的属性和方法都是实例属性和方法。可以使用ClassVar定义类变量，@classmethod,@staticmethod标注类方法和静态方法。

类对象

class本身也是对象，从语法上类中所有的都是类对象的属性。但是从语义上它们有些是实例的，有些是类的。注意语义有些是类型系统提供的，运行时效果参考语法上的定义。

可以使用ClassVar定义类属性，@classmethod,@staticmethod标注类方法和静态方法。

@classmethod标注的方法，被类调用时会自动传入类对象：

class MyClass:
    @classmethod
    def bar(cls: Type['MyClass']) -> str:
        return cls.__doc__
 
MyClass.bar() # equal MyClass.bar(cls=MyClass)

@staticmethod标注的方法，被类调用时不会自动传入类对象，就是常规的静态方法：

class MyClass:
    @staticmethod
    def baz(x):
        return x > 0
 
MyClass.baz(1) # True

构造函数

__init__方法是类的构造函数，在实例化类时自动调用。无显示构造函数的类，默认返回”空”实例。

过程类似：

class MyClass:
    def __init__(self, x):
        self.x = x
    def foo(self):
        return self.x
 
# 1. 创建空实例
o = MyClass()
# 2. 调用__init__方法，还是利用了动态添加属性的机制
o.__init__(11)
 
# 当然实际构造时，直接调用即可，但是实际操作和上面是一样的
ob = MyClass(11)

Note

构造函数__init__也是会被继承的，本质来说它也是一个普通方法。

实例对象

实例对象支持的操作就是对属性和方法的访问。 用命名空间的方式来理解就是，实例可访问对象和类(包括继承链上的类)的命名空间。 类和实例都是独立对象，实例对象命名空间中访问不到的会去类命名空间中找。

由于动态语言的特征，所以运行时可以添加属性和del属性，这带来便利性，也带来一些安全性问题。

Note

动态属性的特征造成的后果就是，不管你怎么定义，你都必须显示或隐式的给对象加上这个属性，否则对象都不会有这个属性，没有这个属性并不意味着不能访问，因为能访问到其他命名空间。

方法和函数

注意方法和函数的区别，虽然本质都是一个，但是参数是不同的，方法少一个self参数。

o = MyClass()
# 方法调用
o.foo() # ok equal MyClass.foo(o); o.foo.__class__ is method
fx = MyClass.foo
# 函数调用
fx() # error too few arguments;fx.__class__ is function

python没有属性的访问控制，类似get和set方法，所以就算你进行了某种约定，一样可以绕过直接修改属性。

方法可访问方法、模块、内置命名空间。

继承

class A(B):
    pass

A extends B 继承有个隐患，子类重写基类某个方法，这个方法在基类中被其他方法调用，子类中调用这个其他方法，其他方法中不会调用基类的相关方法，而是调用子类重写的方法，因为此时self指向了子类对象。

继承检测机制：

• isinstance,isinstance(obj, A),ob是否为A类型
• issubclass,issubclass(A, B),A是否为B类型的子类

多重继承

class A(B,C,D):
    pass

多重继承的命名空间访问顺序问题，并不是一定的深度优先，从左到右。而是采用动态算法将搜索顺序线性化,也就是MRO，即能提供对super()的支持，还能避免某些基类被访问多次，因为基类可能继承至同一个类，比如object。 所以super本质并不是指向意义上的父类，而是MRO链的下一个类，所以继承顺序的改变可能会导致一些不可预料的问题。

类继承执行机制

class A:
    __init_subclass__(cls, **kwargs):
        ...
class B(A):
    __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        ...
class C(B):
    pass

__init_subclass__提供了继承后的反馈机制，一但创建了子类，父类的这个方法会被执行，这和__init__不同，__init__只会实例话时执行。 同时注意多层继承时，要想传递到顶层，需要super()调用，并不会自动传递。

这个方法还可以接收参数，那在哪传递参数呢：

class SubClass(BaseClass, key=value): # 这本质上就是创建类时传递的实参，创建子类，所以参数是给父类的，父类的`__init_subclass__`和父类的元类的`__new__`方法，都可以接收参数
    pass

这有什么用：

• 给子类增加属性和方法

• 检查子类的属性是否合格

• 一些动态注册机制，比如实现插件注册

• ORM model注册

  class Model:
  _registry = {}
   
  def __init_subclass__(cls, table_name=None, **kwargs):
      super().__init_subclass__(**kwargs)
      cls._table_name = table_name or cls.__name__.lower()
      cls._registry[cls._table_name] = cls
      print(f"✓ 注册模型: {cls.__name__} -> 表 '{cls._table_name}'")
   
  @classmethod
  def get_model(cls, table_name):
      return cls._registry.get(table_name)

总之： 这是一个偏框架开发使用的方法。

私有变量

python是一个约定式语言，虽然没有私有属性，但是一般把带有_前缀的当作私有属性。

并且为了实现这种私有访问，还支持名称改写，当变量满足^_{2,}\w+_?$，会被改写为_classname_\w+_，这样就算继承了也不冲突，因为类名不同，方法名也就不同了。 这个特性还解决了上面的继承隐患。

迭代器

本质一个对象如果定义了一个__iter__方法，返回了一个实现了__next__方法的对象，那这个对象就是可迭代的，就可以使用for循环。

class Reverse:
  """example for iterator
 
  This class is an example for iterator. It can be used to iterate over a list in reverse order.
  """
  def __init__(self, x: list[int]):
      self.x = x
      # private field, will be renamed to _Reverse__index
      self.__index = len(x)
  def __iter__(self):
      return self
  def __next__(self):
      if self.__index == 0:
          raise StopIteration
      else:
          self.__index -= 1
          return self.x[self.__index]
 
rl = Reverse([1,2,3,4,5])
 
# print(dir(rl))
 
# next(rl) # rl.__next__()
 
for i in rl:
  print(i)

生成器

生成器可以快速生成迭代器，但是这个迭代器，迭代的原理和普通的迭代器不同。

• 普通迭代器：next函数，执行迭代体，每次执行return一个值，直到raise StopIteration。迭代体在重复的执行。
• 生成器迭代器：next函数，会调用生成器的迭代体，每次执行yield一个值，并暂停，函数的栈帧被保存，再次调用next时，恢复栈帧继续执行，直到遇到下一个yield或者迭代体结束，结束时会raise StopIteration。迭代体就执行了一次。

生成器函数的执行会生成一个迭代器，调用next()才会执行。

from typing import Iterator,Generator
def reverse(data: list[int]):
    for i in range(len(data)-1, -1, -1):
        yield data[i]
 
it = reverse([1,2,3,4,5])
next(it)
isinstance(it, Iterator) # True
issubclass(Generator, Iterator) # True

yield from 将外部的next信号传递进去：

def sub_generator():
    yield 1
    yield 2
    yield 3
 
def main_generator():
    yield 'Start'
    yield from sub_generator()  # Delegates to sub_generator
    yield 'End'
 
for item in main_generator():
    print(item)
 
# output
# Start
# 1
# 2
# 3
# End

生成器表达式

生成器表达式语法基本同列表推导式，但是用()，且会被外层函数立即使用，即立即生成一个迭代器。所以外层的函数接收的参数应该支持迭代器。

sum(i*i for i in range(5)) # ()中是生成器，sum(iterator)

async

关于异步

异步是允许”同时”运行多个任务，即可以中途切换到其他任务。 对一个任务来说，这并不是允许改变它的执行顺序，而是可以暂停等待异步完成后再执行，它本身的顺序是严格按逻辑执行的。 所以js的Promise，和python的原生协程都有类似改变异步对象状态的逻辑，由事件循环改变对象状态，然后唤醒任务继续执行。

异步的本质就是中途切换出去，等待状态完成再切换回来。

实现方式

python中的异步，通过协程实现

有2种实现方式：

1. generator生成器协程 利用栈帧的保存恢复机制实现 生成器函数会返回一个迭代器，通过next()方法调用

2. 原生协程 async函数返回一个协程对象，这个协程对象只有在run或者await时才会被交给事件循环去执行，事件循环会从ready队列里取出协程对象，执行它的同步部分，直到遇到await后，挂起当前协程，然后将await后面的协程继续交给event loop，event loop在遇到真正的I/O等阻塞情况时会丢给操作系统去处理，然后继续执行ready队列中的协程，当操作系统完成I/O后会，事件循环将就绪的协程放在ready队列中，等待调用执行。 event loop 维护了2个队列： 挂起的，ready

   事件循环是会堵塞当前线程的，直到没有协程需要执行了才会继续执行后面的代码。

   注意： 协程对象只能在事件循环中执行，await实际是注册协程到事件循环去执行并挂起等待协程完成，当协程对象状态为done才会继续执行，也就是模拟同步操作

   假设有协程对象A,B,C,D，D中有一个异步操作，调用栈为A → B → C → D，A通过事件循环调用，其他通过await注册，当D中sleep完成，恢复D执行，D变成done，C继续执行出栈，然后B，最后A结束

   import asyncio
    
   async def D():
       await asyncio.sleep(1)
    
   async def C():
       await D()
    
   async def B():
       await C()
    
   async def A():
       await B()
    
   asyncio.run(A())

   协程被调度执行时，同步代码会顺序执行，直到遇到异步代码，而异步代码则会导致协程被挂起，等待异步操作完成，再恢复执行 await会等待后面的awaitable对象变成done，再继续执行。

   asyncio.sleep()函数的源码进行了解，它是通过Future这个底层对象来实现的。

   关于什么是异步操作，表面上看就是不需要CPU介入的都是异步操作，比如I/O，从实现上看就是能将控制权交给事件循环的，并且能够注册回调，让事件循环在未来某个时刻调用回调，改变awaitable的状态，然后事件循环会唤醒对应的协程继续执行，这就是异步，可以看

异步生成器

import asyncio
 
async def async_gen():
    await asyncio.sleep(1)
    yield 1
    await asyncio.sleep(1)
    yield 2

这个函数返回async_generator异步生成器，每次迭代都返回一个awaitable，通过await去调用执行。 可以使用async for和anext调用:

async def main():
    async for item in async_gen():
        print(item)
 
    # anext返回一个awaitable对象
    # 等效上面的async for
    # ag = async_gen()
    # print(await anext(ag))
    # print(await anext(ag))
 
asyncio.run(main())

async with

异步资源管理(try..source)

import aiofiles
async def main():
    async with open('file.txt', 'r') as f:
        # 在async with中会自动 await __aenter__() 和 await __aexit__()
        content = await f.read()

async with最重要的特征是必须处于async def协程中，会异步执行__aenter__和__aexit__，同时也可以异步执行过程代码

可等待对象

主要有3种：协程，Task，Futures

任务用于并行调度协程，通过asyncio.create_task(coro)创建任务，该协程会被自动调度执行，也就是封装为任务即是加入调度，在合适的时候会被执行，就像await一样，但是只是加入调度并没等待完成的逻辑，所以一般还需要await task。

Note

Task更像是协程的一个强引用，管理协程的入口，比如可以获取协程的状态，取消协程等等。

async def hello_time(task: int):
    print(f"hello {task}")
    await asyncio.sleep(task)
    print(f"world {task}")
 
async def task_test():
    task = asyncio.create_task(hello_time(3))
    print("before await")
    await say_hello(1)
    # 没有await task，task还没有完成，但是程序已经结束
    # await task
 

Future表示异步操作的结果，是一个底层对象，比如sleep方法就用到，Future被等待时，协程将保持等待直到Future完成。

Note

对比协程和Task，Future只是一个结果标记，并不是可执行体。

任务组

asyncio.TaskGroup类似Promise.all，并行执行直到所有任务完成，任何一个任务如果抛出asyncio.CancelledError之外的异常，分组中的剩余任务会被取消。 asyncio.gather效果类似，但是TaskGroup更“安全”，gather不会主动取消剩余任务，除非整个gather被取消。

asyncio.wait(aws,*,timeout=None,return_when=ALL_COMPLETED)这个函数，通过return_when可以模拟Promise.all,Promise.race等逻辑

async def do_it(time: int):
    await asyncio.sleep(time, result=time)
 
 
async def race():
    aws = [asyncio.create_task(do_it(i)) for i in range(1, 4)]
done, pending = await asyncio.wait(aws, return_when=asyncio.FIRST_COMPLETED)
print(len(done))  # 1
print(len(pending))  # 2

还可以使用asyncio.as_completed，返回一个迭代器，每个迭代元素都可以被await，然后返回一个最先完成的结果，注意并不是返回对应awaitable的结果。

async def as_completed():
    import random
 
    aws = [asyncio.create_task(do_it(i)) for i in range(1, 4)]
    random.shuffle(aws) # 随机打乱
    for task in asyncio.as_completed(aws):
        result = await task
        print(result) # 结果顺序不变

其他线程运行

常规的异步处理，是因为操作能够异步，可以切换到其他协程执行需要CPU参与的任务，但是有些不能切换的（比如同步IO操作），如何提高效率，多线程。 在开启了GIL的python实现中，这也是有用的，因为IO并不占用CPU，主线程依然可以继续执行。

Note

GIL的核心是不能并行进行CPU执行，所以CPU密集型的任务多线程毫无意义。

更灵活的跨线程调度： asyncio.run_coroutine_threadsafe(coro, loop)，这个函数的作用是在一个线程中将一个协程提交到另一个线程的事件循环。

正则

python的正则实现较为完整。 常规用法：

• re.findall(pattern, string, flags=0)，返回匹配到的所有子串
• re.search(pattern, string, flags=0)，从任意位置，返回匹配到的第一个子串
• re.match(pattern, string, flags=0)，从头开始匹配，返回匹配到的第一个子串
• re.fullmatch(pattern, string, flags=0)，匹配整个字符串，类似re.match('^<pattern>$')效果
• re.split(pattern, string, maxsplit=0, flags=0)，分割字符串，返回分割后的子串列表
• re.sub(pattern, repl, string, count=0, flags=0)，子串替换，返回替换后的字符串
• re.escape(string)，转义字符串，将字符串中的特殊字符转义，可以查看哪些字符被转义了

特殊标记

flags参数效果同这些标记：

• (?aiLmsux) : 开启特殊标记
  • re.A (ASCII-only matching)
  • re.I (ignore case)
  • re.L (locale dependent)
  • re.M (multi-line
  • re.S (dot matches all)
  • re.U (Unicode matching)
  • re.X (verbose))
• (?aiLmsux-imsx:...): 开启特殊标记，-之前的标记是对所有pattern有效，-之后的标记只对:后面的pattern有效

注意

:表示标记只对后面的正则有效，意思是:后面的pattern有新的含义了，重点还是pattern，不要混淆了

比如(?s:.)这个标记的含义是:后面的.开启re.S标记，可以匹配包括\n在内的任意字符，不再是普通的不能匹配\n了。

(?)用法

• (?P<name><pattern>)这仅仅是给匹配组命名，并不会影响匹配的效果，可以通过match.group()`访问匹配组。

• (?P=<name>)，引用最近的这个命名匹配组，类似变量的效果

• (?:<pattern>)，这是不捕获这个匹配组的意思，可以减少group数量，避免干扰，比如有时候只是想要用()表达一个整体，比如(?:\d+\.?){2,}。

• (?><pattern>), 表示固化分组，禁止回溯，可以提高性能，比如abcabc，使用(?>abc)+abc时，将匹配None,因为(?>abc)+匹配了完整的abcabc,并且不会吐出来，导致后面的abc不能匹配。 关于贪婪模式，?表示禁止贪婪，即尽可能少的匹配，比如*?,+?,??，而+表示贪婪匹配，且不会回溯，这本质上和?>效果一样，所以下面的含义一样:

  • (?>a+) 和 a++
  • (?>a*) 和 a*+
  • (?>a?) 和 a?+

• 断言类型：这些都不消耗字符，也就是只判断，并且也不会当作匹配组

  断言	名称	含义	消耗字符
  (?=...)	正向前瞻	后面必须是	❌ 否
  (?!...)	负向前瞻	后面不能是	❌ 否
  (?<=...)	正向后顾	前面必须是	❌ 否
  (?<!...)	负向后顾	前面不能是	❌ 否

匹配次数的问题

上面的例子(\d+\.?){2,}，要匹配类似1.2.3之类的，但是这个匹配组只会捕获最后一次，也就是group(1)此时为3，这是{m,n}的问题，可以使用((?:\d+\.?){2,})再包裹一层来解决，当然内部的匹配组建议不捕获，因为没有意义了