用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/microwind/ai-skills --skill sql命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
基于 SOC 职业分类
正在显示 SKILL.md
| name | SQL生成器 |
| description | 当生成SQL查询、验证SQL、优化查询或调试SQL错误时，在执行到生产数据库之前验证SQL。 |
| license | MIT |
SQL是强大且永久的。无效的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. 设置合理的超时
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()
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()