Python 数据库优化：查询与索引优化

Python 数据库优化：查询与索引优化

1. 技术分析

1.1 数据库性能瓶颈

数据库操作是应用性能的常见瓶颈：

性能瓶颈类型 查询慢: 缺乏索引 连接池耗尽: 连接管理不当 数据量大: 未分页 锁竞争: 并发写入

1.2 数据库优化层次

1.3 查询优化原则

查询优化原则 只查需要的列: SELECT specific_columns 使用索引: WHERE条件列要有索引 避免全表扫描: 使用LIMIT 批量操作: 减少往返次数

2. 核心功能实现

2.1 SQL查询优化

import sqlite3 class QueryOptimizer: @staticmethod def slow_query(db, user_id): cursor = db.cursor() cursor.execute("SELECT * FROM orders WHERE user_id = ?", (user_id,)) return cursor.fetchall() @staticmethod def fast_query(db, user_id): cursor = db.cursor() cursor.execute( "SELECT id, order_date, amount FROM orders WHERE user_id = ?", (user_id,) ) return cursor.fetchall() @staticmethod def batch_insert(db, items): cursor = db.cursor() cursor.executemany( "INSERT INTO items (name, price) VALUES (?, ?)", [(item['name'], item['price']) for item in items] ) db.commit() @staticmethod def create_index(db, table, column): cursor = db.cursor() cursor.execute(f"CREATE INDEX IF NOT EXISTS idx_{table}_{column} ON {table}({column})") db.commit() def optimize_complex_query(db): cursor = db.cursor() query = """ SELECT u.name, COUNT(o.id) as order_count FROM users u JOIN orders o ON u.id = o.user_id WHERE u.created_at > ? GROUP BY u.id HAVING COUNT(o.id) > 5 ORDER BY order_count DESC LIMIT 10 """ cursor.execute(query, ('2024-01-01',)) return cursor.fetchall()

2.2 连接池管理

from contextlib import contextmanager import threading class ConnectionPool: def __init__(self, max_connections=10): self.max_connections = max_connections self.connections = [] self.lock = threading.Lock() def _create_connection(self): return sqlite3.connect('example.db') def get_connection(self): with self.lock: if self.connections: return self.connections.pop() if len(self.connections) < self.max_connections: return self._create_connection() raise Exception("连接池已满") def release_connection(self, conn): with self.lock: if len(self.connections) < self.max_connections: self.connections.append(conn) @contextmanager def connection(self): conn = self.get_connection() try: yield conn finally: self.release_connection(conn) class DatabaseManager: def __init__(self, pool_size=10): self.pool = ConnectionPool(max_connections=pool_size) def execute_query(self, query, params=None): with self.pool.connection() as conn: cursor = conn.cursor() cursor.execute(query, params or ()) return cursor.fetchall() def execute_batch(self, query, items): with self.pool.connection() as conn: cursor = conn.cursor() cursor.executemany(query, items) conn.commit()

2.3 ORM优化

from sqlalchemy import create_engine, select, func from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base class ORMOptimizer: def __init__(self, db_url): self.engine = create_engine(db_url) self.Session = sessionmaker(bind=self.engine) def get_users_with_orders(self, min_orders=5): session = self.Session() query = ( select(User.name, func.count(Order.id).label('order_count')) .join(Order) .group_by(User.id) .having(func.count(Order.id) > min_orders) .order_by(func.count(Order.id).desc()) .limit(10) ) result = session.execute(query).all() session.close() return result def batch_insert_users(self, users): session = self.Session() try: session.add_all(users) session.commit() except: session.rollback() raise finally: session.close() class QueryProfiler: def __init__(self, engine): self.engine = engine def profile_query(self, query): import time start = time.time() result = self.engine.execute(query).fetchall() elapsed = time.time() - start return { 'result': result, 'time': elapsed, 'row_count': len(result) }

3. 性能对比

3.1 查询优化效果

3.2 连接池效果

3.3 索引效果

4. 最佳实践

4.1 数据库优化模式

def optimize_database_performance(db_config): db = DatabaseManager(pool_size=db_config.get('pool_size', 10)) indexes = [ ('users', 'email'), ('orders', 'user_id'), ('orders', 'order_date') ] for table, column in indexes: QueryOptimizer.create_index(db.pool.get_connection(), table, column) return db class DatabaseOptimizationWorkflow: def __init__(self): pass def run(self): self._analyze_queries() self._identify_slow_queries() self._create_indexes() self._optimize_queries() self._configure_connection_pool()

4.2 SQL审查清单

class SQLReviewChecker: @staticmethod def check(query): issues = [] if 'SELECT *' in query: issues.append("避免SELECT *，只查询需要的列") if 'WHERE' in query and 'INDEX' not in query: issues.append("检查WHERE条件列是否有索引") if 'JOIN' in query and 'ON' not in query: issues.append("确保JOIN有ON条件") return issues

5. 总结

数据库优化是提升应用性能的关键：

1. 索引优化：最有效的优化手段
2. 查询优化：避免SELECT *，使用LIMIT
3. 连接池：减少连接开销
4. 批量操作：减少数据库往返

对比数据如下：

• 索引可以带来100-1000倍性能提升
• 连接池减少90%以上连接开销
• 批量插入提升20倍性能
• 推荐定期审查慢查询日志