#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 获取 university 数据库所有表的 schema 信息 使用 pymysql 连接数据库 """ import pymysql import sys # 数据库连接配置 DB_CONFIG = { 'host': '127.0.0.1', 'user': 'root', 'password': '123456', 'database': 'dbsclab2026', 'charset': 'utf8mb4', 'cursorclass': pymysql.cursors.DictCursor } def get_connection(): """获取数据库连接""" try: print("正在连接数据库...", flush=True) conn = pymysql.connect(**DB_CONFIG) print("✓ 数据库连接成功!", flush=True) return conn except Exception as e: print(f"✗ 数据库连接失败: {e}", flush=True) return None def get_all_tables(cursor): """获取所有表和视图""" cursor.execute(""" SELECT TABLE_NAME, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s ORDER BY TABLE_TYPE, TABLE_NAME """, (DB_CONFIG['database'],)) return cursor.fetchall() def get_table_columns(cursor, table_name): """获取表的列信息""" cursor.execute(""" SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, COLUMN_KEY, COLUMN_DEFAULT, EXTRA, COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s ORDER BY ORDINAL_POSITION """, (DB_CONFIG['database'], table_name)) return cursor.fetchall() def get_table_constraints(cursor, table_name): """获取表的主键和外键约束""" # 主键 cursor.execute(""" SELECT COLUMN_NAME FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND CONSTRAINT_NAME = 'PRIMARY' ORDER BY ORDINAL_POSITION """, (DB_CONFIG['database'], table_name)) primary_keys = [row['COLUMN_NAME'] for row in cursor.fetchall()] # 外键 cursor.execute(""" SELECT kcu.CONSTRAINT_NAME, kcu.COLUMN_NAME, kcu.REFERENCED_TABLE_NAME, kcu.REFERENCED_COLUMN_NAME, rc.DELETE_RULE FROM information_schema.KEY_COLUMN_USAGE kcu LEFT JOIN information_schema.REFERENTIAL_CONSTRAINTS rc ON kcu.CONSTRAINT_NAME = rc.CONSTRAINT_NAME AND kcu.TABLE_SCHEMA = rc.CONSTRAINT_SCHEMA WHERE kcu.TABLE_SCHEMA = %s AND kcu.TABLE_NAME = %s AND kcu.REFERENCED_TABLE_NAME IS NOT NULL ORDER BY kcu.CONSTRAINT_NAME, kcu.ORDINAL_POSITION """, (DB_CONFIG['database'], table_name)) foreign_keys = cursor.fetchall() # CHECK 约束 - MySQL 8.0+ try: cursor.execute(""" SELECT tc.CONSTRAINT_NAME, cc.CHECK_CLAUSE FROM information_schema.TABLE_CONSTRAINTS tc JOIN information_schema.CHECK_CONSTRAINTS cc ON tc.CONSTRAINT_NAME = cc.CONSTRAINT_NAME WHERE tc.TABLE_SCHEMA = %s AND tc.TABLE_NAME = %s AND tc.CONSTRAINT_TYPE = 'CHECK' """, (DB_CONFIG['database'], table_name)) check_constraints = cursor.fetchall() except Exception: check_constraints = [] return primary_keys, foreign_keys, check_constraints def get_view_definition(cursor, view_name): """获取视图定义""" cursor.execute(""" SELECT VIEW_DEFINITION FROM information_schema.VIEWS WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s """, (DB_CONFIG['database'], view_name)) result = cursor.fetchone() return result['VIEW_DEFINITION'] if result else None def print_schema_info(conn): """打印并保存所有表的 schema 信息""" cursor = conn.cursor() # 获取所有表 tables = get_all_tables(cursor) output_lines = [] output_lines.append("=" * 60) output_lines.append(f"数据库: {DB_CONFIG['database']} (university)") output_lines.append("所有表的 Schema 信息") output_lines.append("=" * 60) output_lines.append(f"\n共 {len(tables)} 个对象：") base_tables = [t for t in tables if t['TABLE_TYPE'] == 'BASE TABLE'] views = [t for t in tables if t['TABLE_TYPE'] == 'VIEW'] output_lines.append(f" - {len(base_tables)} 个基础表") output_lines.append(f" - {len(views)} 个视图\n") # 输出基础表 output_lines.append("=" * 60) output_lines.append("基础表 (BASE TABLE)") output_lines.append("=" * 60) for idx, table in enumerate(base_tables, 1): table_name = table['TABLE_NAME'] output_lines.append(f"\n{'─' * 60}") output_lines.append(f"{idx}. {table_name}") output_lines.append(f"{'─' * 60}") # 列信息 columns = get_table_columns(cursor, table_name) output_lines.append(f"\n{'字段名':<20} {'类型':<20} {'空值':<6} {'键':<8} {'默认值':<25} {'说明'}") output_lines.append("-" * 120) for col in columns: col_name = col['COLUMN_NAME'] col_type = col['COLUMN_TYPE'] is_nullable = col['IS_NULLABLE'] col_key = col['COLUMN_KEY'] col_default = col['COLUMN_DEFAULT'] extra = col['EXTRA'] comment = col['COLUMN_COMMENT'] nullable_str = 'NO' if is_nullable == 'NO' else 'YES' key_str = col_key if col_key else '' default_str = str(col_default) if col_default else 'NULL' if extra: default_str += f" ({extra})" comment_str = comment if comment else '' output_lines.append(f"{col_name:<20} {col_type:<20} {nullable_str:<6} {key_str:<8} {default_str:<25} {comment_str}") # 约束信息 primary_keys, foreign_keys, check_constraints = get_table_constraints(cursor, table_name) if primary_keys: output_lines.append(f"\n主键: {', '.join(primary_keys)}") if foreign_keys: output_lines.append("外键:") for fk in foreign_keys: col_name = fk['COLUMN_NAME'] ref_table = fk['REFERENCED_TABLE_NAME'] ref_col = fk['REFERENCED_COLUMN_NAME'] delete_rule = fk['DELETE_RULE'] output_lines.append(f" {col_name} -> {ref_table}({ref_col}) ON DELETE {delete_rule}") if check_constraints: output_lines.append("检查约束:") for check in check_constraints: output_lines.append(f" CHECK ({check['CHECK_CLAUSE']})") # 输出视图 output_lines.append("\n" + "=" * 60) output_lines.append("视图 (VIEW)") output_lines.append("=" * 60) for idx, view in enumerate(views, 1): view_name = view['TABLE_NAME'] output_lines.append(f"\n{'─' * 60}") output_lines.append(f"{idx}. {view_name}") output_lines.append(f"{'─' * 60}") # 视图定义 view_def = get_view_definition(cursor, view_name) if view_def: output_lines.append(f"\n定义:\n{view_def}") # 视图字段 columns = get_table_columns(cursor, view_name) output_lines.append(f"\n{'字段名':<20} {'类型':<20} {'说明'}") output_lines.append("-" * 60) for col in columns: col_name = col['COLUMN_NAME'] col_type = col['COLUMN_TYPE'] comment = col['COLUMN_COMMENT'] comment_str = comment if comment else '' output_lines.append(f"{col_name:<20} {col_type:<20} {comment_str}") # 打印到控制台 output_text = '\n'.join(output_lines) print(output_text, flush=True) # 保存到文件 output_file = 'university_schema_info_python.txt' with open(output_file, 'w', encoding='utf-8') as f: f.write(output_text) print(f"\n✓ Schema 信息已保存到: {output_file}", flush=True) cursor.close() def main(): """主函数""" print("\n" + "=" * 60, flush=True) print("开始获取数据库 schema 信息...", flush=True) print("=" * 60 + "\n", flush=True) conn = get_connection() if conn: try: print_schema_info(conn) except Exception as e: print(f"✗ 执行失败: {e}", flush=True) import traceback traceback.print_exc() finally: conn.close() print("\n✓ 数据库连接已关闭。", flush=True) else: print("✗ 无法连接数据库", flush=True) if __name__ == '__main__': main()