In our design pattern series, let’s now look at a more advanced design pattern used in SQLAlchemy.
One of the most ingenious patterns in SQLAlchemy is how it dynamically generates query methods. While the full implementation is complex, here's a simplified version showing the core concept:
class ModelMetaclass(type):
def __new__(cls, name, bases, attrs):
# Generate methods for common queries
for operation in ['find_by', 'filter_by']:
for field in attrs.get('fields', []):
method_name = f'{operation}_{field}'
attrs[method_name] = cls.generate_query_method(operation, field)
return super().__new__(cls, name, bases, attrs)
@staticmethod
def generate_query_method(operation, field):
def query_method(self, value):
if operation == 'find_by':
return f"SELECT * FROM {self.__table__} WHERE {field} = {value}"
elif operation == 'filter_by':
return f"SELECT * FROM {self.__table__} WHERE {field} LIKE '%{value}%'"
return query_method
class Model(metaclass=ModelMetaclass):
__table__ = 'users'
fields = ['name', 'email', 'status']
# Usage
user_model = Model()
print(user_model.find_by_email('user@example.com'))
print(user_model.filter_by_name('John'))
This pattern is brilliant because it:
Reduces boilerplate code dramatically
Creates methods dynamically based on model attributes
Maintains consistent query interface across models
Allows for easy extension and modification of query generation
This pattern demonstrate how Python's flexibility allows for elegant solutions to complex problems. It shows how good design can make code more maintainable, efficient, and easier to understand.
Let's explore how we can apply this pattern in different contexts.
Automatic Event Handler Registration
Imagine we're building a game engine where different game objects need to handle various events. We can use metaclasses to automatically register event handlers based on method names:
class GameObjectMetaclass(type):
def __new__(cls, name, bases, attrs):
# Store event handlers
handlers = {}
# Look for methods starting with 'on_'
for method_name, method in attrs.items():
if method_name.startswith('on_'):
event_name = method_name[3:] # Remove 'on_' prefix
handlers[event_name] = method
# Add handlers dictionary to class
attrs['_event_handlers'] = handlers
# Add convenience method for handling events
def handle_event(self, event_name, *args, **kwargs):
handler = self._event_handlers.get(event_name)
if handler:
return handler(self, *args, **kwargs)
return None
attrs['handle_event'] = handle_event
return super().__new__(cls, name, bases, attrs)
# Usage example
class Player(metaclass=GameObjectMetaclass):
def __init__(self, name):
self.name = name
self.health = 100
def on_collision(self, other_object):
print(f"{self.name} collided with {other_object}")
def on_damage(self, amount):
self.health -= amount
print(f"{self.name} took {amount} damage. Health: {self.health}")
def on_heal(self, amount):
self.health += amount
print(f"{self.name} healed for {amount}. Health: {self.health}")
# Using the class
player = Player("Hero")
player.handle_event("collision", "wall") # Automatically calls on_collision
player.handle_event("damage", 20) # Automatically calls on_damage
Automatic API Endpoint Generation
Here's how we could use metaclasses to automatically generate API endpoints for a service class, similar to how FastAPI or Flask work but with our own twist:
from functools import wraps
import inspect
class APIEndpointMetaclass(type):
def __new__(cls, name, bases, attrs):
endpoints = {}
# Look for methods with endpoint decorator
for method_name, method in attrs.items():
if hasattr(method, '_endpoint_info'):
endpoints[method._endpoint_info['path']] = {
'method': method,
'http_method': method._endpoint_info['http_method'],
'requires_auth': method._endpoint_info.get('requires_auth', False)
}
attrs['_endpoints'] = endpoints
return super().__new__(cls, name, bases, attrs)
def endpoint(path, http_method='GET', requires_auth=False):
def decorator(func):
# Store endpoint information in the function object
func._endpoint_info = {
'path': path,
'http_method': http_method,
'requires_auth': requires_auth
}
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
# Usage example
class UserService(metaclass=APIEndpointMetaclass):
def __init__(self):
self.users = {}
@endpoint('/users', http_method='GET')
def get_users(self):
return list(self.users.values())
@endpoint('/users/{user_id}', http_method='GET')
def get_user(self, user_id):
return self.users.get(user_id)
@endpoint('/users', http_method='POST', requires_auth=True)
def create_user(self, user_data):
user_id = len(self.users) + 1
self.users[user_id] = user_data
return {"id": user_id, **user_data}
# Simple router implementation
class Router:
def __init__(self, service):
self.service = service
def handle_request(self, path, http_method, auth=None, **kwargs):
if path in self.service._endpoints:
endpoint_info = self.service._endpoints[path]
if endpoint_info['http_method'] != http_method:
return {"error": "Method not allowed"}
if endpoint_info['requires_auth'] and not auth:
return {"error": "Authentication required"}
return endpoint_info['method'](self.service, **kwargs)
return {"error": "Not found"}
# Using the service
service = UserService()
router = Router(service)
# Simulate HTTP requests
print(router.handle_request('/users', 'GET'))
print(router.handle_request('/users', 'POST', auth='token', user_data={"name": "John"}))