1. Generators & Coroutines — Less Code, Faster Execution
Generators are like “lazy workers” — they only do enough work to push the task forward.
📌 Background Scenario
Suppose we are building a real-time log processing system:
•
Logs are continuously generated (e.g., access logs, transaction logs).
•
We need to:
•
Read logs one by one (using generators to save memory);
•
Real-time count occurrences of certain keywords (using coroutines to receive & process data).
📕 Code Implementation
import random
import time
# Simulate log generator — returns one log at a time (saves memory)
def log_generator():
levels = ['INFO', 'WARNING', 'ERROR']
level = random.choice(levels)
while True:
yield f'{time.strftime("%H:%M:%S")} - {level} - Something happened!'
time.sleep(0.5) # Simulate interval
# Coroutine: Count occurrences of ERROR
def error_counter():
count = 0
try:
while True:
log = (yield) # Coroutine receives log
if 'ERROR' in log:
count += 1
print(f'Counted ERROR occurrences: {count}')
except GeneratorExit:
print(f'Coroutine closed, final ERROR count: {count}')
if __name__ == '__main__':
logs = log_generator() # Start log generator
counter = error_counter()
next(counter) # Activate coroutine (must call once)
try:
for _ in range(10): # Simulate reading 10 logs
log = next(logs)
print(f'[Log] {log}')
counter.send(log) # Send log to coroutine
finally:
counter.close() # Close coroutine
🔑 Explanation
•
Generator (<span>log_generator</span>) : Simulates a “data source” that produces one log at a time, saving memory.
•
Coroutine (<span>error_counter</span>) : Receives data via <span>yield</span> and processes it in real-time (here counting occurrences of ERROR).
•
<span>counter.send(log)</span> → Pushes the log to the coroutine for processing.
•
<span>counter.close()</span> → Gracefully closes the coroutine.
✅ Running Effect
[Log] 00:00:00 - ERROR - Something happened!
Counted ERROR occurrences: 1
[Log] 00:00:00 - ERROR - Something happened!
Counted ERROR occurrences: 2
[Log] 00:00:01 - ERROR - Something happened!
Counted ERROR occurrences: 3
[Log] 00:00:01 - ERROR - Something happened!
Counted ERROR occurrences: 4
[Log] 00:00:02 - ERROR - Something happened!
Counted ERROR occurrences: 5
[Log] 00:00:02 - ERROR - Something happened!
Counted ERROR occurrences: 6
[Log] 00:00:03 - ERROR - Something happened!
Counted ERROR occurrences: 7
[Log] 00:00:03 - ERROR - Something happened!
Counted ERROR occurrences: 8
[Log] 00:00:04 - ERROR - Something happened!
Counted ERROR occurrences: 9
[Log] 00:00:04 - ERROR - Something happened!
Counted ERROR occurrences: 10
Coroutine closed, final ERROR count: 10
🚀 Advantages
•
Memory efficient
•
Suitable for processing large datasets
•
Uses <span>yield</span> instead of <span>return</span>
•
Can combine <span>next()</span> and <span>send()</span> to achieve coroutine behavior
🔥 Advanced Bonus: Combine with async to get asynchronous generators. This is true “performance cooking”.
Scenario
We want to implement a log collection service:
•Asynchronously pull logs from multiple servers;•Use asynchronous generators to “produce while pulling”;•Asynchronous consumers process logs in real-time (e.g., filter ERROR and count).
📕 Code Implementation
import asyncio
import random
import time
# Asynchronous log source: Simulate fetching logs from different servers
async def fetch_logs(server_name: str):
levels = ['INFO', 'WARNING', 'ERROR']
for _ in range(5): # Each server produces 5 logs
await asyncio.sleep(random.uniform(0.2, 1.0)) # Simulate network delay
level = random.choice(levels)
yield f"{server_name} | {time.strftime('%Y-%m-%d %H:%M:%S')} | {level}"
# Asynchronous consumer: Process logs in real-time
async def process_logs(server_name: str):
async for log in fetch_logs(server_name): # Asynchronous generator fetches data
print(f'[Consume] {log}')
if 'ERROR' in log:
print(f'[⚠️Warning] Found error log: {log}')
# Main entry: Process multiple loggers simultaneously
async def main():
servers = ['Server-A', 'Server-B', 'Server-C']
tasks = [asyncio.create_task(process_logs(server)) for server in servers]
await asyncio.gather(*tasks)
if __name__ == '__main__':
asyncio.run(main())
✅ Running Effect
[Consume] Server-A | 2025-08-21 00:20:16 | WARNING
[Consume] Server-C | 2025-08-21 00:20:17 | WARNING
[Consume] Server-B | 2025-08-21 00:20:17 | INFO
[Consume] Server-C | 2025-08-21 00:20:17 | INFO
[Consume] Server-A | 2025-08-21 00:20:17 | INFO
[Consume] Server-B | 2025-08-21 00:20:17 | WARNING
[Consume] Server-B | 2025-08-21 00:20:18 | INFO
[Consume] Server-A | 2025-08-21 00:20:18 | WARNING
[Consume] Server-C | 2025-08-21 00:20:18 | WARNING
[Consume] Server-B | 2025-08-21 00:20:18 | ERROR
[⚠️Warning] Found error log: Server-B | 2025-08-21 00:20:18 | ERROR
[Consume] Server-B | 2025-08-21 00:20:18 | WARNING
[Consume] Server-C | 2025-08-21 00:20:19 | ERROR
[⚠️Warning] Found error log: Server-C | 2025-08-21 00:20:19 | ERROR
[Consume] Server-A | 2025-08-21 00:20:19 | ERROR
[⚠️Warning] Found error log: Server-A | 2025-08-21 00:20:19 | ERROR
[Consume] Server-C | 2025-08-21 00:20:19 | WARNING
[Consume] Server-A | 2025-08-21 00:20:20 | INFO
🔑 Highlights Analysis
•
<span>async def fetch_logs</span> + <span>yield</span> → Asynchronous Generators: Each log is “produced” at different time points, rather than loading all at once.
•
async for: Consumers asynchronously fetch logs one by one in the event loop, without blocking other tasks.
•
<span>asyncio.gather</span>: Concurrently runs multiple log sources, achieving real-time collection from multiple servers.
•
Performance Advantages
•
If implemented synchronously, must wait for one server to return before processing the next;
•
Here, the asynchronous method allows 3 servers to pull logs simultaneously, with performance nearly linearly improved.
2. Descriptors
Function
Descriptors are a protocol (protocol = specific method agreement) provided by Python, allowing us to insert custom logic duringattribute access (<span>get</span>, <span>set</span>, <span>delete</span>). The core methods are three:
•
<span>__get__(self, instance, owner)</span> → Triggered when accessing an attribute
•
<span>__set__(self, instance, value)</span> → Triggered when modifying an attribute
•
<span>__delete__(self, instance)</span> → Triggered when deleting an attribute
As long as these methods are defined in a class, it becomes a “descriptor object” that controls how another class accesses its attributes.
Usage Scenarios
•
Attribute validation (e.g., ensuring age > 0, email is valid)
•
Type checking (ensuring assignment is <span>int/str/...</span>)
•
Lazy loading
•
ORM frameworks (SQLAlchemy uses descriptors to implement field and database mapping)
•
Proxy access (accessing an attribute actually calls an external resource, like Redis cache)
🎯 Example: Age Validation
class PositiveInteger:
"""Descriptor for validating positive integers."""
def __get__(self, instance, owner):
return instance.__dict__.get(self.name, None)
def __set__(self, instance, value):
if not isinstance(value, int) or value <= 0:
raise ValueError(f"{self.name} must be a positive integer")
instance.__dict__[self.name] = value
def __set_name__(self, owner, name):
# Get the name of the attribute automatically
self.name = name
class Person:
age = PositiveInteger()
def __init__(self, age):
self.age = age
if __name__ == '__main__':
p = Person(30)
print(p.age) # Output: 30
p.age = 20
print(p.age) # Output: 20
p.age = -10 # Raises ValueError: age must be a positive integer
Advantages
•
Encapsulates attribute access logic: Automatically triggers validation during attribute assignment, reading, and deletion, avoiding repetitive <span>setter/getter</span><span>.</span>
•
High reusability: A descriptor class can be reused across multiple business classes, standardizing specifications.
•
Decoupled & Elegant: Separates validation/control logic from business logic, making code clearer.
•
Framework-level Core Technology: Django ORM, SQLAlchemy field mapping, property decorators, all fundamentally based on descriptor protocols.
Descriptors are the “magic hooks” in Python’s object-oriented programming, allowing you to finely control attribute read/write behavior, widely used in data validation, ORM mapping, caching proxies, etc.
3. Metaclasses — The Ultimate Boss of Python
Function
Metaclasses are the classes that create classes.
•
In Python, everything is an object: objects are created by classes, and classes themselves are created by metaclasses.
•
By default, Python uses <span>type</span> as the metaclass:
class Foo: pass
print(type(Foo)) # <class 'type'>
•In other words:Metaclasses allow you to dynamically modify class definitions (attributes, methods, inheritance) during the class creation process.
Usage Scenarios
•
ORM frameworks (Django, SQLAlchemy)
•
Model classes (User, Article) → Automatically map to database table structures
•
Metaclasses intercept class definitions, collect field information, and generate SQL mappings
•
Singleton pattern, interface constraints: Use metaclasses to automatically transform classes during creation, avoiding repetitive code
•
Automatic registration mechanism (plugin systems, factory patterns): Automatically add classes to a global registry upon definition for dynamic loading
•
Code standard checks / injection features
•
Check if methods are implemented or conform to naming conventions when creating classes
•
Automatically inject common methods (like logging, configuration loading)
🎯 Example: Automatic Registration of Subclasses (Common in Plugin Systems)
class RegistryMeta(type):
registry = {}
def __new__(mcls, name, bases, attrs):
cls = super().__new__(mcls, name, bases, attrs)
if name != 'Base':
RegistryMeta.registry[name] = cls
return cls
class Base(metaclass=RegistryMeta):
pass
class PluginA(Base):
def run(self):
return "PluginA running"
class PluginB(Base):
def run(self):
return "PluginB running"
# Automatic registration successful
print(RegistryMeta.registry)
# {'PluginA': <class '__main__.PluginA'>, 'PluginB': <class '__main__.PluginB'>}
# Dynamic usage
plugin = RegistryMeta.registry['PluginA']()
print(plugin.run()) # PluginA running
Advantages
•
Higher level of abstraction capability: Can automatically handle logic during the “class generation phase”, reducing boilerplate code.
•
Framework-level core technology: Advanced frameworks like Django, SQLAlchemy, Pydantic extensively use metaclasses.
•
Unified specifications: Can enforce subclasses to implement certain methods, similar to interface constraints.
•
Automation & Decoupling: Plugin systems, automatic registration mechanisms make extensions more elegant and flexible.
Metaclasses (Metaclass) are the “class factories” in Python, capable of intercepting and modifying class definitions during the class creation phase, commonly used in ORM, plugin systems, automatic registration, code constraints, and are the backbone of advanced frameworks.
4. Decorators (Decorating Decorators)
You know decorators, but did you know that decorators themselves can also be decorated?
from functools import wraps
def smart_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("Before...")
result = func(*args, **kwargs)
print("After...")
return result
return wrapper
🔁 Common Applications:
•
Caching (<span>@lru_cache</span>)
•
Authentication logic
•
Automatic retries
•
Logging
•
Memoization functions
Writing Python in production, this is definitely a must-have skill.
5. AsyncIO and Concurrency — Doing More with Less Resources
Sync code? Simple.Parallel code? Powerful.Asynchronous code? ✨ Magic ✨
import asyncio
async def say_hello():
await asyncio.sleep(1)
print("Hello!")
async def main():
await asyncio.gather(say_hello(), say_hello())
asyncio.run(main())
# Hello!
# Hello!
🕹️ Practical Applications:
•
Web crawlers
•
WebSocket applications
•
Real-time dashboards
•
Asynchronous APIs (FastAPI + async SQLAlchemy = ❤️)
6. Type Hints & Static Analysis — Making Code More Controllable for the Future
Python’s static typing is no longer just an “option”. Combined with <span>mypy</span>, <span>pyright</span>, <span>pydantic</span>, it feels like having superpowers.
def add(x: int, y: int) -> int:
return x + y
🔧 Why is it Important?
•
Safer refactoring
•
Stronger IDE support
•
Self-documenting code
•
Fewer runtime errors
✨ Professional Advice: Use <span>TypedDict</span>, <span>Protocol</span>, <span>dataclasses</span> to design robust and maintainable code.
Final Words
If basic Python is your bicycle 🚲 — then advanced Python is your Tesla ⚡.
As you gradually master metaclasses, async, decorators, generators and other advanced skills, you will no longer just be writing code, but building Pythonic systems.
And the best part is:
The deeper you go, the simpler your code will become.
Clean, elegant, efficient. This is the Python way 🐍.
Thanks for your reading! Enjoy coding!
Recommended Reading👇👇👇
•Efficient Python Programming: 6 Features of functools to Avoid Repeating Yourself!•Efficient Use of Python Dictionaries: 8 Inefficient Usages to Avoid and Advanced Techniques•These Lesser-Known Techniques — Significantly Improve Your Python Code Performance (Most Developers Overlook)!•10 “Outdated” Python Practices, Please Stop Using Them Immediately!•A Beginner’s Memo on Python Lists•A Beginner’s Memo on Basic Concepts of Python•Must-Know Python: These 5 Lesser-Known Python Treasures, Almost No One Will Teach You!
🌟 If you find this article helpful and are willing to support me, you can : 🌟
•👍 Like, so more people can see it•⤴️ Share, to pass the content to your peers•❤️ Recommend, to let the article impact more people•🧑🤝🧑 Help me meet more like-minded friends•👏 Welcome to leave comments, your feedback is very important to me
👉 Follow me, to get more practical tips and the latest trends on Python, data analysis, machine learning.
•Let’s explore the power of data together and unlock the potential of AI.•Your every interaction is the greatest encouragement to me and the driving force for our mutual growth! 🚀
👇👇👇 Follow me, to get more high-quality content sharing, see you next time!