Python3 advanced主题tutorial

Learning Python3 in advancedprogrammingconcepts, including装饰器, 生成器, asynchronousprogrammingetc.

advanced主题Introduction

Python is afunctions强 big programminglanguage, 除了basic 语法 and functions out , 它还providing了许 many advanced features, 这些features可以helping我们writing更简洁, 更 high 效, 更优雅 code. 本tutorial将介绍Pythonin一些最常用 advancedprogrammingconcepts, helping你提升programming技能.

本tutorial涵盖 advanced主题including:

装饰器

装饰器 is Pythonin一种特殊 function, 它可以modifyotherfunction behavior. 装饰器通常用于以 under 场景:

basic装饰器

def simple_decorator(func):
    def wrapper():
        print(" in function执行 before 做一些事情")
        func()
        print(" in function执行 after 做一些事情")
    return wrapper

@simple_decorator
def say_hello():
    print("Hello, World!")

# 调用function
say_hello()
# 输出:
#  in function执行 before 做一些事情
# Hello, World!
#  in function执行 after 做一些事情

带parameter 装饰器

def decorator_with_args(func):
    def wrapper(*args, **kwargs):
        print(f"function {func.__name__} 被调用, parameter: {args}, {kwargs}")
        result = func(*args, **kwargs)
        print(f"function {func.__name__} 执行完毕, return value: {result}")
        return result
    return wrapper

@decorator_with_args
def add(a, b):
    return a + b

@decorator_with_args
def greet(name, greeting="Hello"):
    return f"{greeting}, {name}!"

# 调用function
print(add(5, 3))
print(greet("Alice"))
print(greet("Bob", greeting="Hi"))

带parameter 装饰器工厂

def repeat(num_times):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for i in range(num_times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(3)
def say_hello(name):
    print(f"Hello, {name}!")

# 调用function
say_hello("Alice")
# 输出:
# Hello, Alice!
# Hello, Alice!
# Hello, Alice!

保留function元data 装饰器

import functools

def decorator(func):
    @functools.wraps(func)  # 保留function元data
    def wrapper(*args, **kwargs):
        """package装function"""
        print("执行 before ")
        result = func(*args, **kwargs)
        print("执行 after ")
        return result
    return wrapper

@decorator
def original_function():
    """原始function"""
    print("原始function执行")

# 调用function
original_function()

# 查看function元data
print(f"function名: {original_function.__name__}")
print(f"functiondocumentation: {original_function.__doc__}")

生成器 and iterators

iterators

iterators is a implementation了__iter__() and __next__()method object. iterators可以逐个返回元素, 直 to 没 has 元素 for 止, 此时会抛出StopIterationexception.

class MyIterator:
    def __init__(self, start, end):
        self.start = start
        self.end = end
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.start < self.end:
            value = self.start
            self.start += 1
            return value
        else:
            raise StopIteration

# usingiterators
my_iter = MyIterator(1, 5)
for num in my_iter:
    print(num)

# 输出:
# 1
# 2
# 3
# 4

生成器

生成器 is a特殊 iterators, 它usingyield语句来return value. 生成器比iterators更简洁, 更 easy implementation.

def my_generator(start, end):
    while start < end:
        yield start
        start += 1

# using生成器
for num in my_generator(1, 5):
    print(num)

# 输出:
# 1
# 2
# 3
# 4

# 生成器表达式
squares = (x**2 for x in range(5))
for square in squares:
    print(square)

# 输出:
# 0
# 1
# 4
# 9
# 16

生成器 advanced用法

def fibonacci(n):
    """生成斐波那契数列  before n个数"""
    a, b = 0, 1
    count = 0
    while count < n:
        yield a
        a, b = b, a + b
        count += 1

# using生成器
for num in fibonacci(10):
    print(num)

# 输出:
# 0
# 1
# 1
# 2
# 3
# 5
# 8
# 13
# 21
# 34

# 生成器serving as协程
def echo_generator():
    print("开始生成器")
    while True:
        received = yield
        print(f"收 to : {received}")

# using生成器serving as协程
gen = echo_generator()
next(gen)  # 启动生成器
gen.send("Hello")
gen.send("World")
gen.close()  # 关闭生成器

# 输出:
# 开始生成器
# 收 to : Hello
# 收 to : World

on under 文management器

on under 文management器 is aimplementation了__enter__() and __exit__()method object, 它可以 in code块执行 before after 执行specific operation. on under 文management器通常usingwith语句来using, 最common 例子 is fileoperation.

自定义 on under 文management器

class Timer:
    def __enter__(self):
        import time
        self.start_time = time.time()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        self.end_time = time.time()
        self.elapsed_time = self.end_time - self.start_time
        print(f"执行时间: {self.elapsed_time:.6f} 秒")
        # 返回False, 表示exception会被传播
        return False

# using on  under 文management器
with Timer() as timer:
    # 执行一些耗时operation
    total = 0
    for i in range(1000000):
        total += i
    print(f"计算结果: {total}")

# 输出:
# 计算结果: 499999500000
# 执行时间: 0.123456 秒

usingcontextmanager装饰器

from contextlib import contextmanager

@contextmanager
def timer():
    import time
    start_time = time.time()
    try:
        yield  #  yield之 before  code in 进入 on  under 文时执行
    finally:
        end_time = time.time()
        elapsed_time = end_time - start_time
        print(f"执行时间: {elapsed_time:.6f} 秒")

# using on  under 文management器
with timer():
    # 执行一些耗时operation
    total = 0
    for i in range(1000000):
        total += i
    print(f"计算结果: {total}")

# 输出:
# 计算结果: 499999500000
# 执行时间: 0.123456 秒

on under 文management器 application

from contextlib import contextmanager
import tempfile
import os

@contextmanager
def temporary_file(mode='w', suffix=''):
    """creation临时file, using完毕 after 自动delete"""
    fd, path = tempfile.mkstemp(suffix=suffix)
    try:
        with os.fdopen(fd, mode) as f:
            yield f
    finally:
        if os.path.exists(path):
            os.unlink(path)

# using on  under 文management器
with temporary_file('w', '.txt') as f:
    f.write('Hello, World!')
    f.write('\nThis is a temporary file.')
    print(f"临时filepath: {f.name}")

# file in 此处已自动delete
print(f"临时file is 否存 in : {os.path.exists(f.name)}")

元programming

元programming is 指writingable tooperationcode code. in Pythonin, 元programming主要through以 under 方式implementation:

反射

class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age
    
    def say_hello(self):
        print(f"Hello, my name is {self.name}.")

# creationobject
p = Person("Alice", 30)

# using反射获取property
print(f"Name: {getattr(p, 'name')}")
print(f"Age: {getattr(p, 'age')}")

# checkproperty is 否存 in 
print(f"Has name: {hasattr(p, 'name')}")
print(f"Has gender: {hasattr(p, 'gender')}")

# 设置property
setattr(p, 'gender', 'female')
print(f"Gender: {p.gender}")

# 获取method并调用
method = getattr(p, 'say_hello')
method()

# 获取class 所 has property and method
print(f"\n所 has property and method: {dir(p)}")

元class

class Meta(type):
    def __new__(mcs, name, bases, attrs):
        #  in creationclass之 before modifyproperty
        attrs['added_by_meta'] = 'This attribute was added by the metaclass'
        return super().__new__(mcs, name, bases, attrs)
    
    def __init__(cls, name, bases, attrs):
        super().__init__(name, bases, attrs)
        #  in classcreation after 做一些事情
        print(f"class {name} 被creation")

class MyClass(metaclass=Meta):
    def __init__(self, value):
        self.value = value

# creationobject
obj = MyClass(42)
print(f"Value: {obj.value}")
print(f"Added by meta: {obj.added_by_meta}")

describes符

class Descriptor:
    def __get__(self, instance, owner):
        print("Getting value")
        return instance._value
    
    def __set__(self, instance, value):
        print(f"Setting value to {value}")
        instance._value = value
    
    def __delete__(self, instance):
        print("Deleting value")
        del instance._value

class MyClass:
    descriptor = Descriptor()
    
    def __init__(self, value):
        self._value = value

# usingdescribes符
obj = MyClass(42)
print(f"Value: {obj.descriptor}")
obj.descriptor = 100
print(f"Value: {obj.descriptor}")
del obj.descriptor

many thread and many process

Pythonproviding了 many thread and many process support, 用于implementationconcurrentprogramming. concurrentprogramming可以improving程序 执行efficiency, 特别 is in processingIO密集型task时.

many thread

import threading
import time

def worker(name, delay):
    print(f"thread {name} 开始")
    time.sleep(delay)
    print(f"thread {name} 结束")

# creationthread
thread1 = threading.Thread(target=worker, args=("A", 2))
thread2 = threading.Thread(target=worker, args=("B", 1))

# 启动thread
thread1.start()
thread2.start()

# etc.待thread结束
print("主threadetc.待thread结束")
thread1.join()
thread2.join()
print("所 has thread已结束")

threadsecurity

import threading
import time

class Counter:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()
    
    def increment(self):
        with self.lock:
            # critical section
            temp = self.value
            time.sleep(0.001)  # mock耗时operation
            self.value = temp + 1

# creation计数器
counter = Counter()

# creation many 个thread
def worker():
    for _ in range(1000):
        counter.increment()

threads = []
for i in range(10):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

# etc.待所 has thread结束
for t in threads:
    t.join()

print(f"最终计数: {counter.value}")

many process

import multiprocessing
import time

def worker(name, delay):
    print(f"process {name} 开始")
    time.sleep(delay)
    print(f"process {name} 结束")

if __name__ == "__main__":
    # creationprocess
    process1 = multiprocessing.Process(target=worker, args=("A", 2))
    process2 = multiprocessing.Process(target=worker, args=("B", 1))

    # 启动process
    process1.start()
    process2.start()

    # etc.待process结束
    print("主processetc.待process结束")
    process1.join()
    process2.join()
    print("所 has process已结束")

asynchronousprogramming

asynchronousprogramming is aprogramming范式, 它允许程序 in etc.待IOoperationcompletion时执行othertask, 而不 is 阻塞etc.待. Python 3.5+引入了asynciomodule and async/await语法, 用于implementationasynchronousprogramming.

basic asynchronousfunction

import asyncio

async def say_hello():
    print("Hello")
    # mockIOoperation
    await asyncio.sleep(1)
    print("World")

async def main():
    print("开始")
    # runasynchronousfunction
    await say_hello()
    print("结束")

# run主function
asyncio.run(main())

concurrent执行 many 个asynchronoustask

import asyncio
import time

async def task(name, delay):
    print(f"task {name} 开始")
    await asyncio.sleep(delay)
    print(f"task {name} 结束")
    return f"task {name} completion"

async def main():
    start_time = time.time()
    print("开始")
    
    # concurrent执行 many 个task
    tasks = [
        task("A", 2),
        task("B", 1),
        task("C", 3)
    ]
    
    # etc.待所 has taskcompletion
    results = await asyncio.gather(*tasks)
    print(f"结果: {results}")
    
    end_time = time.time()
    print(f"总执行时间: {end_time - start_time:.2f} 秒")

# run主function
asyncio.run(main())

asynchronousIOoperation

import asyncio
import aiofiles

async def read_file(filename):
    async with aiofiles.open(filename, 'r') as f:
        content = await f.read()
    return content

async def write_file(filename, content):
    async with aiofiles.open(filename, 'w') as f:
        await f.write(content)

async def main():
    # 写入file
    await write_file('test.txt', 'Hello, Async IO!')
    print("file写入completion")
    
    # 读取file
    content = await read_file('test.txt')
    print(f"file in 容: {content}")

# run主function
asyncio.run(main())

闭package

闭package is 指一个function可以访问其 out 部作用域in variable, 即使 out 部function已经执行完毕. 闭package通常用于以 under 场景:

def make_counter():
    count = 0
    
    def counter():
        nonlocal count
        count += 1
        return count
    
    return counter

# creation计数器
c1 = make_counter()
c2 = make_counter()

# using计数器
print(f"c1: {c1()}")  # 1
print(f"c1: {c1()}")  # 2
print(f"c2: {c2()}")  # 1
print(f"c1: {c1()}")  # 3
print(f"c2: {c2()}")  # 2

# function工厂
def make_multiplier(factor):
    def multiplier(x):
        return x * factor
    return multiplier

# creation乘法function
double = make_multiplier(2)
triple = make_multiplier(3)

# using乘法function
print(f"double(5): {double(5)}")  # 10
print(f"triple(5): {triple(5)}")  # 15

function式programming

Pythonsupportfunction式programming范式, function式programming强调using纯function, 不可变data and high 阶function. Pythonin function式programmingtoolincluding:

# lambda表达式
add = lambda x, y: x + y
print(f"add(5, 3): {add(5, 3)}")

# mapfunction
numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x**2, numbers))
print(f"平方: {squared}")

# filterfunction
even_numbers = list(filter(lambda x: x % 2 == 0, numbers))
print(f"偶数: {even_numbers}")

# reducefunction
from functools import reduce
sum_result = reduce(lambda x, y: x + y, numbers)
print(f"总 and : {sum_result}")

# list推导式
squared = [x**2 for x in numbers]
print(f"list推导式平方: {squared}")

even_squared = [x**2 for x in numbers if x % 2 == 0]
print(f"偶数 平方: {even_squared}")

# 生成器表达式
squared_gen = (x**2 for x in numbers)
print(f"生成器表达式平方: {list(squared_gen)}")

实践case: cache装饰器

creation一个cache装饰器, 用于cachefunction 计算结果, improvingfunction 执行efficiency.

import functools
import time

def cache(func):
    """
    cache装饰器, 用于cachefunction 计算结果
    """
    # cachedictionary
    cache_dict = {}
    
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 生成cache键
        key = str(args) + str(kwargs)
        
        # checkcachein is 否存 in 
        if key in cache_dict:
            print(f" from cachein获取结果")
            return cache_dict[key]
        
        # 计算结果
        print(f"计算结果")
        result = func(*args, **kwargs)
        
        # store to cache
        cache_dict[key] = result
        return result
    
    return wrapper

@cache
def fibonacci(n):
    """
    计算斐波那契数列 第n个数
    """
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

@cache
def slow_function(a, b):
    """
    mock耗时function
    """
    time.sleep(1)
    return a + b

# testcache装饰器
print("testfibonaccifunction:")
start_time = time.time()
print(f"fibonacci(30) = {fibonacci(30)}")
end_time = time.time()
print(f"执行时间: {end_time - start_time:.2f} 秒")

print("\n再次调用fibonacci(30):")
start_time = time.time()
print(f"fibonacci(30) = {fibonacci(30)}")
end_time = time.time()
print(f"执行时间: {end_time - start_time:.2f} 秒")

print("\ntestslow_functionfunction:")
start_time = time.time()
print(f"slow_function(10, 20) = {slow_function(10, 20)}")
end_time = time.time()
print(f"执行时间: {end_time - start_time:.2f} 秒")

print("\n再次调用slow_function(10, 20):")
start_time = time.time()
print(f"slow_function(10, 20) = {slow_function(10, 20)}")
end_time = time.time()
print(f"执行时间: {end_time - start_time:.2f} 秒")

互动练习

练习1: 自定义装饰器

creation一个装饰器, 用于记录function 调用次数 and 执行时间.

import functools
import time

def monitor(func):
    """
    monitor装饰器, 记录function调用次数 and 执行时间
    """
    # implementation你 code
    pass

@monitor
def slow_function(n):
    """
    mock耗时function
    """
    total = 0
    for i in range(n):
        total += i
    time.sleep(0.5)
    return total

# test装饰器
print("调用slow_function(1000000):")
result1 = slow_function(1000000)
print(f"结果: {result1}")

print("\n再次调用slow_function(1000000):")
result2 = slow_function(1000000)
print(f"结果: {result2}")

print("\n调用slow_function(500000):")
result3 = slow_function(500000)
print(f"结果: {result3}")

练习2: 生成器练习

creation一个生成器, 用于生成素数.

def is_prime(n):
    """
    判断一个数 is 否 for 素数
    """
    if n <= 1:
        return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0:
            return False
    return True

def prime_generator(limit):
    """
    生成 small 于etc.于limit 所 has 素数
    """
    # implementation你 code
    pass

# test生成器
print("生成 small 于etc.于100 素数:")
for prime in prime_generator(100):
    print(prime, end=" ")
print()

练习3: asynchronousprogramming练习

creation一个asynchronous程序, mock many 个networkrequest concurrent执行.

import asyncio
import time

async def fetch_data(url, delay):
    """
    mocknetworkrequest
    """
    print(f"开始request {url}")
    await asyncio.sleep(delay)
    print(f"request {url} completion")
    return f"{url}  responsedata"

async def main():
    """
    主function
    """
    start_time = time.time()
    print("开始")
    
    # implementation你 code
    # concurrentrequest many 个URL
    
    end_time = time.time()
    print(f"总执行时间: {end_time - start_time:.2f} 秒")

# run主function
asyncio.run(main())