You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
79 lines
2.2 KiB
79 lines
2.2 KiB
import pymysql
|
|
import yaml
|
|
|
|
with open("./config.yml", "r", encoding="utf-8") as f:
|
|
cfg = yaml.safe_load(f)
|
|
DB = cfg["db"]
|
|
|
|
|
|
def get_conn():
|
|
return pymysql.connect(**DB, cursorclass=pymysql.cursors.DictCursor)
|
|
|
|
|
|
def get_table(table_name):
|
|
sql = """
|
|
SELECT table_name, table_comment
|
|
FROM information_schema.tables
|
|
WHERE table_schema=%s AND table_name=%s
|
|
"""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (DB["database"], table_name))
|
|
return cur.fetchone()
|
|
|
|
|
|
def get_all_tables(database=None):
|
|
"""获取指定数据库(默认当前库)的所有表"""
|
|
schema = database or DB["database"]
|
|
sql = """
|
|
SELECT table_name, table_comment
|
|
FROM information_schema.tables
|
|
WHERE table_schema=%s AND table_type='BASE TABLE'
|
|
ORDER BY table_name
|
|
"""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (schema,))
|
|
return cur.fetchall()
|
|
|
|
|
|
def get_columns(table_name):
|
|
sql = """
|
|
SELECT column_name, data_type, column_comment
|
|
FROM information_schema.columns
|
|
WHERE table_schema=%s AND table_name=%s
|
|
ORDER BY ordinal_position
|
|
"""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql, (DB["database"], table_name))
|
|
return cur.fetchall()
|
|
|
|
|
|
def mysql_to_java(mysql_type):
|
|
mapping = {
|
|
"bigint": "Long",
|
|
"int": "Integer",
|
|
"varchar": "String",
|
|
"datetime": "LocalDateTime",
|
|
"decimal": "BigDecimal",
|
|
}
|
|
return mapping.get(mysql_type, "String")
|
|
|
|
|
|
def get_enum_field_names():
|
|
"""查询 enums 表所有未删除的 distinct enums_name。
|
|
|
|
这些 enums_name 与业务表的 column_name (snake_case) 对应，
|
|
脚手架生成时据此判断哪些字段应走 Select 下拉框。
|
|
"""
|
|
sql = """
|
|
SELECT DISTINCT enums_name
|
|
FROM enums
|
|
WHERE deleted_flag = 0 AND enums_name IS NOT NULL AND enums_name <> ''
|
|
"""
|
|
with get_conn() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(sql)
|
|
rows = cur.fetchall()
|
|
return [r["enums_name"] for r in rows if r.get("enums_name")]
|
|
|