Advanced Python Topics: Metaclasses, Descriptors, Concurrency & Packaging
1. What are Metaclasses in Python?
Q: What are metaclasses?
Metaclasses are classes that define how other classes are created. They are the "class of a class" and allow customization of class creation. The default metaclass in Python is type. Metaclasses are defined by inheriting from type and overriding methods like __new__ or __init__.
Class Creation Process: When a class is defined, Python uses its metaclass to create it.
Key Methods:
__new__(mcs, name, bases, namespace): Creates the class object.__init__(cls, name, bases, namespace): Initializes the class after creation.
Use Case: Enforcing class-level constraints, logging class creation, or adding attributes/methods dynamically.
Example: Singleton Metaclass
class SingletonMeta(type):
_instances = {}
def __call__(cls, *args, **kwargs):
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
class SingletonClass(metaclass=SingletonMeta):
def __init__(self, value):
self.value = value
# Testing Singleton
obj1 = SingletonClass(1)
obj2 = SingletonClass(2)
print(f"obj1.value: {obj1.value}, obj2.value: {obj2.value}")
print(f"Same instance: {obj1 is obj2}")
Output:
obj1.value: 1, obj2.value: 1
Same instance: True
Note: SingletonMeta ensures only one instance of SingletonClass exists. __call__ intercepts instance creation to enforce the singleton pattern. Metaclasses are powerful but should be used sparingly due to complexity.
2. What are Descriptors in Python?
Q: What are descriptors?
Descriptors are objects that define how attribute access is handled, implementing the descriptor protocol.
Protocol Methods:
__get__(self, obj, owner): Called when accessing the attribute.__set__(self, obj, value): Called when setting the attribute.__delete__(self, obj): Called when deleting the attribute.
Types:
- Data descriptors: Implement
__get__and__set__. - Non-data descriptors: Implement only
__get__.
Use Case: Validating attributes, lazy loading, or custom attribute behavior.
Example: Positive Number Descriptor
class PositiveNumber:
def __init__(self, name):
self.name = name
def __get__(self, obj, owner):
return getattr(obj, f"_{self.name}")
def __set__(self, obj, value):
if not isinstance(value, (int, float)) or value <= 0:
raise ValueError(f"{self.name} must be a positive number")
setattr(obj, f"_{self.name}", value)
class Product:
price = PositiveNumber("price")
def __init__(self, name, price):
self.name = name
self.price = price
# Testing descriptor
try:
prod = Product("Laptop", 1000)
print(f"Price: {prod.price}")
prod.price = 1500
print(f"Updated price: {prod.price}")
prod.price = -100 # Raises ValueError
except ValueError as e:
print(f"Error: {e}")
Output:
Price: 1000
Updated price: 1500
Error: price must be a positive number
Note: PositiveNumber enforces that price is a positive number. __set__ validates the value; __get__ retrieves it. Stores actual value in _ to avoid recursion.
3. Advanced Concurrency Techniques with AsyncIO and Threading
Q: Advanced concurrency?
AsyncIO (Advanced):
- Event Loop: Manages async tasks using
asyncio.get_event_loop(). - Tasks and Futures: Schedule coroutines with
asyncio.create_task()orasyncio.ensure_future(). - Synchronization: Use
asyncio.Lock,asyncio.Semaphorefor async safety. - Use Case: High-concurrency I/O-bound tasks (e.g., web servers, async APIs).
Advanced Threading:
- Thread Pools: Use
concurrent.futures.ThreadPoolExecutorfor managing thread lifecycles. - Synchronization: Use
threading.Lock,threading.Condition, orqueue.Queue. - Use Case: I/O-bound tasks with thread safety requirements.
Key Difference: AsyncIO uses a single thread with cooperative multitasking; threading uses multiple threads with preemptive scheduling.
Limitation: Threading is GIL-limited for CPU-bound tasks; AsyncIO requires async-compatible libraries.
Example: AsyncIO and Threading
import asyncio
import threading
import concurrent.futures
import time
import queue
# AsyncIO with Lock
async def async_worker(name, lock, delay):
async with lock:
print(f"Async worker {name} starting")
await asyncio.sleep(delay)
print(f"Async worker {name} done")
return name
# Threading with Queue
def thread_worker(name, q):
time.sleep(1) # Simulate I/O
q.put(f"Thread {name} completed")
async def main_async():
lock = asyncio.Lock()
tasks = [async_worker(f"worker{i}", lock, 1) for i in range(3)]
results = await asyncio.gather(*tasks)
return results
def main_threading():
q = queue.Queue()
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(thread_worker, f"worker{i}", q) for i in range(3)]
concurrent.futures.wait(futures)
results = []
while not q.empty():
results.append(q.get())
return results
if __name__ == "__main__":
# AsyncIO
start_time = time.time()
async_results = asyncio.run(main_async())
print(f"AsyncIO results: {async_results}")
print(f"AsyncIO time: {time.time() - start_time:.2f} seconds\n")
# Threading
start_time = time.time()
thread_results = main_threading()
print(f"Threading results: {thread_results}")
print(f"Threading time: {time.time() - start_time:.2f} seconds")
4. Python Project Packaging
Q: How to package Python projects?
Packaging allows distribution via PyPI. Modern approach uses pyproject.toml with build and twine.
Basic Structure:
my_package/
├── pyproject.toml
├── src/
│ └── my_package/
│ ├── __init__.py
│ └── module.py
├── README.md
└── LICENSE
pyproject.toml Example:
[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"
[project]
name = "my-package"
version = "0.1.0"
description = "A sample package"
authors = [{name = "Your Name", email = "[email protected]"}]
Build & Upload: python -m build → creates dist/. twine upload dist/*.
Use Case: Sharing reusable code or publishing open-source libraries.
5. Common Mistakes & Best Practices
Q: Common mistakes?
- Metaclasses: Overcomplicating class creation, reducing readability. Incorrectly overriding
__new__or__init__, breaking class creation. - Descriptors: Forgetting to store values in a private attribute, causing recursion errors. Not handling
__get__for all cases (e.g., owner access). - Concurrency: Using blocking calls in AsyncIO (e.g.,
time.sleepinstead ofasyncio.sleep). Not synchronizing shared resources in threading, causing race conditions. - Packaging: Missing
pyproject.tomlor incorrect metadata, preventing distribution. Not testing package installation in a clean environment. - General: Overusing advanced features when simpler solutions suffice. Not documenting complex logic in metaclasses or descriptors.
Q: Best practices?
- Metaclasses: Use sparingly for specific needs (e.g., singletons, validation). Document metaclass behavior clearly with docstrings.
- Descriptors: Store data in private attributes (e.g.,
_). Implement__get__,__set__, and__delete__as needed. - Concurrency: Use
asyncio.Lockorthreading.Lockfor synchronization. Useconcurrent.futuresfor thread/process pools. Profile performance withtimeorcProfile. - Packaging: Use
pyproject.tomlfor modern packaging withsetuptools. Test packages in a virtual environment. IncludeREADME.mdand clear metadata. - General: Follow PEP 8 for clear naming and indentation. Use linters (e.g.,
flake8) to catch errors. Test with edge cases (e.g., invalid inputs, concurrency conflicts).