Menü
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/microwind/ai-skills --skill sqlDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
Basierend auf der SOC-Berufsklassifikation
SKILL.md wird angezeigt
| 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()