Learning Python3 in advancedprogrammingconcepts, including装饰器, 生成器, asynchronousprogrammingetc.
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 场景:
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 做一些事情
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"))
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!
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__}")
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
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器 is aimplementation了__enter__() and __exit__()method object, 它可以 in code块执行 before after 执行specific operation. on under 文management器通常usingwith语句来using, 最common 例子 is fileoperation.
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 秒
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 秒
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 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 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}")
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
Pythonproviding了 many thread and many process support, 用于implementationconcurrentprogramming. concurrentprogramming可以improving程序 执行efficiency, 特别 is in processingIO密集型task时.
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已结束")
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}")
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 is aprogramming范式, 它允许程序 in etc.待IOoperationcompletion时执行othertask, 而不 is 阻塞etc.待. Python 3.5+引入了asynciomodule and async/await语法, 用于implementationasynchronousprogramming.
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())
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())
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 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
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)}")
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} 秒")
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}")
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()
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())