sql

name	SQL生成器
description	当生成SQL查询、验证SQL、优化查询或调试SQL错误时，在执行到生产数据库之前验证SQL。
license	MIT

SQL生成器技能

概述

SQL是强大且永久的。无效的SQL会导致数据丢失或损坏。在执行到生产数据库之前彻底测试SQL。

核心原则: SQL是永久的。先测试，后执行。

何时使用

始终:

• 在执行到生产环境之前
• 编写新查询
• 优化慢查询
• 创建迁移
• 调试SQL错误
• 数据库设计

触发短语:

• "生成SQL查询"
• "优化这个SQL"
• "验证SQL语法"
• "创建数据库表"
• "生成插入语句"
• "SQL调试"

SQL生成功能

查询生成

• SELECT查询构建
• INSERT语句生成
• UPDATE语句创建
• DELETE语句生成
• 复杂查询组合

数据库设计

• 表结构生成
• 索引创建
• 约束定义
• 外键关系
• 视图创建

SQL优化

• 查询性能分析
• 索引建议
• 执行计划分析
• 慢查询优化
• 资源使用优化

常见SQL问题

SQL注入攻击

问题:
SQL注入导致安全漏洞

错误示例:
"SELECT * FROM users WHERE name = '" + userName + "'"

解决方案:
使用参数化查询:
"SELECT * FROM users WHERE name = ?"

或使用ORM框架:
User.objects.filter(name=userName)

慢查询

问题:
查询执行时间过长

错误示例:
SELECT * FROM orders WHERE customer_id IN (
    SELECT id FROM customers WHERE status = 'active'
)

解决方案:
1. 添加索引
2. 使用JOIN替代子查询
3. 优化WHERE条件
4. 分页查询

死锁问题

问题:
并发操作导致死锁

错误示例:
事务1: 锁定表A，然后锁定表B
事务2: 锁定表B，然后锁定表A

解决方案:
1. 统一锁定顺序
2. 减少事务时间
3. 使用乐观锁
4. 设置合理的超时

代码实现示例

SQL生成器

import re
import json
from typing import Dict, List, Any, Optional, Union, Tuple
from dataclasses import dataclass
from enum import Enum
import sqlparse
from sqlparse.sql import IdentifierList, Identifier, Function
from sqlparse.tokens import Keyword, DML, Name

class SQLOperation(Enum):
    SELECT = "SELECT"
    INSERT = "INSERT"
    UPDATE = "UPDATE"
    DELETE = "DELETE"
    CREATE = "CREATE"
    DROP = "DROP"
    ALTER = "ALTER"

class SQLJoinType(Enum):
    INNER = "INNER JOIN"
    LEFT = "LEFT JOIN"
    RIGHT = "RIGHT JOIN"
    FULL = "FULL OUTER JOIN"
    CROSS = "CROSS JOIN"

@dataclass
class SQLColumn:
    """SQL列定义"""
    name: str
    data_type: str
    nullable: bool = True
    primary_key: bool = False
    foreign_key: Optional[str] = None
    default_value: [] = 
    unique:  = 
    auto_increment:  = 


 :
    
    name: 
    columns: [SQLColumn]
    indexes: [[, ]] = 
    constraints: [[, ]] = 


 :
    
    operation: SQLOperation
    tables: []
    columns: []
    conditions: []
    joins: [[, ]] = 
    group_by: [] = 
    having: [] = 
    order_by: [] = 
    limit: [] = 
    offset: [] = 

 :
    
    
     ():
        .data_type_mapping = {
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : ,
            : 
        }
    
     () -> :
        
        columns_sql = []
        
         column  table.columns:
            column_sql = ._build_column_definition(column)
            columns_sql.append(column_sql)
        
        
        primary_keys = [col.name  col  table.columns  col.primary_key]
         (primary_keys) > :
            columns_sql.append()
        
        
        foreign_keys = [col  col  table.columns  col.foreign_key]
         fk  foreign_keys:
            columns_sql.append()
        
        
        unique_columns = [col.name  col  table.columns  col.unique]
         unique_columns:
            columns_sql.append()
        
        sql = 
        sql += .join(columns_sql)
        sql += 
        
         sql
    
     () -> :
        
        sql = 
        
          column.nullable:
            sql += 
        
         column.auto_increment:
            sql += 
        
         column.default_value   :
            sql += 
        
         column.primary_key  ([col  col  [column]  col.primary_key]) == :
            sql += 
        
         sql
    
     () -> :
        
        columns = (data.keys())
        values = (data.values())
        
        
        formatted_values = []
         value  values:
             (value, ):
                formatted_values.append()
             (value, ):
                formatted_values.append(  value  )
             value  :
                formatted_values.append()
            :
                formatted_values.append((value))
        
        sql = 
        sql += 
        
         sql
    
     () -> :
        
        set_clauses = []
         column, value  data.items():
             (value, ):
                set_clauses.append()
             (value, ):
                set_clauses.append()
             value  :
                set_clauses.append()
            :
                set_clauses.append()
        
        where_clauses = []
         column, value  conditions.items():
             (value, ):
                where_clauses.append()
             (value, ):
                where_clauses.append()
             value  :
                where_clauses.append()
            :
                where_clauses.append()
        
        sql = 
         where_clauses:
            sql += 
        
         sql
    
     () -> :
        
        
         query.columns == []:
            select_clause = 
        :
            select_clause = .join(query.columns)
        
        sql = 
        
        
         query.joins:
             join  query.joins:
                join_type = join.get(, SQLJoinType.INNER.value)
                table = join[]
                condition = join[]
                sql += 
        
        
         query.conditions:
            sql += 
        
        
         query.group_by:
            sql += 
        
        
         query.having:
            sql += 
        
        
         query.order_by:
            sql += 
        
        
         query.limit   :
            sql += 
             query.offset   :
                sql += 
        
         sql
    
     () -> :
        
        where_clauses = []
         column, value  conditions.items():
             (value, ):
                where_clauses.append()
             (value, ):
                where_clauses.append()
             value  :
                where_clauses.append()
            :
                where_clauses.append()
        
        sql = 
         where_clauses:
            sql += 
        
         sql
    
     () -> :
        
        index_type =   unique  
        sql = 
         sql
    
     () -> SQLTable:
        
        columns = []
        
         col_name, col_info  table_dict[].items():
            column = SQLColumn(
                name=col_name,
                data_type=col_info[],
                nullable=col_info.get(, ),
                primary_key=col_info.get(, ),
                foreign_key=col_info.get(),
                default_value=col_info.get(),
                unique=col_info.get(, ),
                auto_increment=col_info.get(, )
            )
            columns.append(column)
        
         SQLTable(
            name=table_dict[],
            columns=columns,
            indexes=table_dict.get(, []),
            constraints=table_dict.get(, [])
        )
    
     () -> []:
        
        migrations = []
        
        
        old_columns = {col.name: col  col  old_table.columns}
        new_columns = {col.name: col  col  new_table.columns}
        
        
         col_name, column  new_columns.items():
             col_name   old_columns:
                alter_sql = 
                migrations.append(alter_sql)
        
        
         col_name  old_columns:
             col_name   new_columns:
                alter_sql = 
                migrations.append(alter_sql)
        
        
         col_name, new_column  new_columns.items():
             col_name  old_columns:
                old_column = old_columns[col_name]
                 old_column.data_type != new_column.data_type:
                    alter_sql = 
                    migrations.append(alter_sql)
        
         migrations


```python
 :
    
    
     ():
        .reset()
    
     ():
        
        ._operation = 
        ._tables = []
        ._columns = []
        ._conditions = []
        ._joins = []
        ._group_by = []
        ._having = []
        ._order_by = []
        ._limit = 
        ._offset = 
         
    
     ():
        
        ._operation = SQLOperation.SELECT
        ._columns = (columns)  columns  []
         
    
     ():
        
        ._tables = (tables)
         
    
     ():
        
        ._conditions.append(condition)
         
    
     ():
        
         (values, (, )):
            formatted_values = []
             v  values:
                 (v, ):
                    formatted_values.append()
                :
                    formatted_values.append((v))
            condition = 
        :
            condition = 
        
        ._conditions.append(condition)
         
    
     ():
        
        ._conditions.append()
         
    
     ():
        
        ._joins.append({
            : join_type.value,
            : table,
            : condition
        })
         
    
     ():
        
         .join(table, condition, SQLJoinType.LEFT)
    
     ():
        
         .join(table, condition, SQLJoinType.RIGHT)
    
     ():
        
        ._group_by = (columns)
         
    
     ():
        
        ._having.append(condition)
         
    
     ():
        
        ._order_by = (columns)
         
    
     ():
        
        ._order_by = [  col  columns]
         
    
     ():
        
        ._limit = count
         
    
     ():
        
        ._offset = count
         
    
     () -> :
        
          ._operation   ._tables:
             ValueError()
        
        query = SQLQuery(
            operation=._operation,
            tables=._tables,
            columns=._columns,
            conditions=._conditions,
            joins=._joins,
            group_by=._group_by,
            having=._having,
            order_by=._order_by,
            limit=._limit,
            offset=._offset
        )
        
        generator = SQLGenerator()
         generator.generate_select(query)
    
     ():
        
        ._columns = []
         
    
     ():
        
        ._columns = [  col  ._columns]
         


 ():
    generator = SQLGenerator()
    
    
    users_table = SQLTable(
        name=,
        columns=[
            SQLColumn(, , primary_key=, auto_increment=),
            SQLColumn(, , unique=, nullable=),
            SQLColumn(, , unique=, nullable=),
            SQLColumn(, , default_value=)
        ]
    )
    
    create_sql = generator.create_table(users_table)
    ()
    (create_sql)
    
    
    user_data = {
        : ,
        : 
    }
    
    insert_sql = generator.generate_insert(, user_data)
    ()
    (insert_sql)
    
    
    query_builder = QueryBuilder()
    query = (query_builder
             .select(, , )
             .from_table()
             .where()
             .order_by_desc()
             .limit()
             .build())
    
    ()
    (query)
    
    
    update_sql = generator.generate_update(
        , 
        {: }, 
        {: }
    )
    ()
    (update_sql)

 __name__ == :
    main()

SQL验证器

class SQLValidator:
    """SQL验证器"""
    
    def __init__(self):
        self.reserved_keywords = {
            'SELECT', 'FROM', 'WHERE', 'INSERT', 'UPDATE', 'DELETE',
            'CREATE', 'DROP', 'ALTER', 'INDEX', 'TABLE', 'DATABASE',
            'JOIN', 'INNER', 'LEFT', 'RIGHT', 'FULL', 'OUTER',
            'GROUP', 'BY', 'HAVING', 'ORDER', 'LIMIT', 'OFFSET',
            'UNION', 'DISTINCT', 'COUNT', 'SUM', 'AVG', 'MIN', 'MAX'
        }
    
    def validate_sql(self, sql: str) -> Dict[str, Any]:
        """验证SQL语法"""
        result = {
            'valid': True,
            'errors': [],
            'warnings': [],
            'suggestions': []
        }
        
        try:
            
            parsed = sqlparse.parse(sql)
            
              parsed:
                result[] = 
                result[].append()
                 result
            
             statement  parsed:
                ._validate_statement(statement, result)
            
            
            ._check_injection_risks(sql, result)
            
            
            ._performance_analysis(sql, result)
            
         Exception  e:
            result[] = 
            result[].append()
        
         result
    
     ():
        
        tokens = (statement.flatten())
        
        
          (token.ttype  DML  token  tokens):
            result[].append()
        
        
        identifiers = [token  token  tokens  (token, Identifier)]
         identifier  identifiers:
             identifier.get_real_name()  .reserved_keywords:
                result[].append()
    
     ():
        
        
           sql    sql:
            result[].append()
        
        
        patterns = [
            ,  
            ,   
                  
        ]
        
         pattern  patterns:
             re.search(pattern, sql):
                result[].append()
    
     ():
        
        
           sql.upper():
            result[].append()
        
        
        upper_sql = sql.upper()
           upper_sql    upper_sql:
                upper_sql:
                result[].append()
        
        
           upper_sql  upper_sql.count() > :
            result[].append()
    
     () -> :
        
        :
            formatted = sqlparse.(sql, reindent=, keyword_case=)
             formatted
        :
             sql


 ():
    validator = SQLValidator()
    
    
    test_sql =  + user_input + 
    
    result = validator.validate_sql(test_sql)
    
    ()
    ()
    
     result[]:
        ()
         error  result[]:
            ()
    
     result[]:
        ()
         warning  result[]:
            ()
    
     result[]:
        ()
         suggestion  result[]:
            ()

 __name__ == :
    main()

SQL最佳实践

查询优化

1. 索引使用: 为常用查询字段创建索引
2. **避免SELECT ***: 只查询需要的列
3. 分页查询: 使用LIMIT和OFFSET
4. 批量操作: 减少数据库往返次数

安全性

1. 参数化查询: 防止SQL注入
2. 最小权限: 应用程序使用最小必要权限
3. 数据加密: 敏感数据加密存储
4. 审计日志: 记录重要操作

数据库设计

1. 规范化: 遵循数据库范式
2. 数据类型: 选择合适的数据类型
3. 约束定义: 使用外键和约束
4. 命名规范: 统一的命名约定

相关技能

• sql-optimizer - SQL优化
• database-design - 数据库设计
• data-migration - 数据迁移
• performance-tuning - 性能调优