import time import functools from collections import OrderedDict
deflogger(log_file='app.log'): defdecorator(func): @functools.wraps(func) defwrapper(*args, **kwargs): start = time.time() result = func(*args, **kwargs) end = time.time() log_msg = ( f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] " f"{func.__name__} called with args={args}, kwargs={kwargs}, " f"result={result}, cost={end-start:.3f}s\n" ) withopen(log_file, 'a') as f: f.write(log_msg) return result return wrapper return decorator
defcache(max_size=128, ttl=300): defdecorator(func): cache_data = OrderedDict() @functools.wraps(func) defwrapper(*args, **kwargs): key = (args, tuple(sorted(kwargs.items()))) if key in cache_data: value, timestamp = cache_data[key] if time.time() - timestamp < ttl: return value else: del cache_data[key] value = func(*args, **kwargs) cache_data[key] = (value, time.time()) iflen(cache_data) > max_size: cache_data.popitem(last=False) return value return wrapper return decorator
defretry(times=3, delay=1): defdecorator(func): @functools.wraps(func) defwrapper(*args, **kwargs): for i inrange(times): try: return func(*args, **kwargs) except Exception as e: if i == times - 1: raise time.sleep(delay) return wrapper return decorator
classField: def__init__(self, default=None, required=False): self.default = default self.required = required self.name = None def__get__(self, instance, owner): if instance isNone: return self return instance.__dict__.get(self.name, self.default) def__set__(self, instance, value): if self.required and value isNone: raise ValueError(f"{self.name} is required") instance.__dict__[self.name] = value
classIntegerField(Field): def__set__(self, instance, value): if value isnotNoneandnotisinstance(value, int): raise TypeError(f"{self.name} must be an integer") super().__set__(instance, value)
classCharField(Field): def__set__(self, instance, value): if value isnotNoneandnotisinstance(value, str): raise TypeError(f"{self.name} must be a string") super().__set__(instance, value)
classModelMeta(type): def__new__(mcs, name, bases, attrs): fields = {k: v for k, v in attrs.items() ifisinstance(v, Field)} for field_name, field in fields.items(): field.name = field_name attrs['_fields'] = fields attrs['_table_name'] = name.lower() + 's' new_class = super().__new__(mcs, name, bases, attrs) ModelRegistry.register(new_class) return new_class
classModel(metaclass=ModelMeta): def__init__(self, **kwargs): for key, value in kwargs.items(): setattr(self, key, value) defsave(self): # 模拟保存到数据库 print(f"Saving {self} to {self._table_name}") defdelete(self): # 模拟删除 print(f"Deleting {self} from {self._table_name}") @classmethod defobjects(cls): return QuerySet(cls)
# 优化前：30 秒 defprocess_users_slow(user_ids): results = [] for user_id in user_ids: user = User.objects.get(id=user_id) orders = Order.objects.filter(user_id=user_id) for order in orders: items = OrderItem.objects.filter(order_id=order.id) for item in items: results.append({ 'user': user.name, 'order': order.id, 'item': item.name, 'price': item.price }) return results
# 优化后：2 秒 defprocess_users_fast(user_ids): # 预加载用户 users = User.objects.in_bulk(user_ids) # 预加载订单 orders = Order.objects.filter( user_id__in=user_ids ).select_related('user') # 预加载订单项 order_ids = [order.idfor order in orders] items = OrderItem.objects.filter( order_id__in=order_ids ) # 组织数据 order_items = {} for item in items: if item.order_id notin order_items: order_items[item.order_id] = [] order_items[item.order_id].append(item) # 生成结果 results = [] for order in orders: user = users[order.user_id] for item in order_items.get(order.id, []): results.append({ 'user': user.name, 'order': order.id, 'item': item.name, 'price': item.price }) return results
微信进群
