使用python脚本连接SQL Server数据库导出表结构
一. 准备工作
Mac 系统安装freetds
brew install freetds
安装pymssql
pip3 install pymssql
二.导出指定表的结构:
import pymssql# 配置数据库连接参数(根据实际情况修改)
server = '' # 内网服务器地址或IP
database = '' # 数据库名称
port = '1433'
username = '' # 登录账号
password = '' # 登录密码
table_name = '' # 需要导出结构的表名
output_file = 'table_structure.txt' # 输出文件名try:conn = pymssql.connect(server=server, port=port, user=username, password=password, database=database,tds_version="7.0")# 创建游标对象cursor = conn.cursor()# 查询表结构query = ("SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, IS_NULLABLE ""FROM INFORMATION_SCHEMA.COLUMNS "f"WHERE TABLE_NAME = '{table_name}'")cursor.execute(query)columns = cursor.fetchall()if not columns:print(f"表 '{table_name}' 不存在或没有列信息")else:# 构建输出内容output = []output.append(f"表结构:{table_name}\n")header = "列名 | 数据类型 | 最大长度 | 允许空 | 主键"separator = "--- | --- | --- | --- | ---"output.extend([header, separator])for col in columns:col_name, data_type, max_len, is_null = colcol_info = [col_name,data_type,str(max_len) if max_len else "N/A","是" if is_null == 'YES' else "否"]output.append(" | ".join(col_info))# 写入文件with open(output_file, 'w', encoding='utf-8') as f:f.write('\n'.join(output))print(f"表结构已成功导出到 {output_file}")except pymssql.Error as e:print(f"数据库连接错误: {str(e)}")
except Exception as e:print(f"发生错误: {str(e)}")
finally:# 确保关闭数据库连接if 'conn' in locals():conn.close()
注意修改脚本中的以下参数值:
server = ''
database = ''
port = '1433'
username = ''
password = ''
同时要注意提供的username要有指定数据库的权限。
三. 导出指定数据库的所有非空表的结构,包含字段名称,类型,注释。
import pymssql# 配置数据库连接参数(根据实际情况修改)
server = ''
database = ''
port = '1433'
username = ''
password = ''
output_file = 'table_structure.txt'try:# 建立数据库连接conn = pymssql.connect(server=server, port=port, user=username, password=password, database=database,tds_version="7.0")cursor = conn.cursor()# 获取所有表名和对应的架构cursor.execute("SELECT TABLE_SCHEMA, TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE'")all_tables = cursor.fetchall()tables_with_data = []for schema, table_name in all_tables:try:# 查询表是否有数据data_query = f"SELECT COUNT(*) FROM [{schema}].[{table_name}]"cursor.execute(data_query)count = cursor.fetchone()[0]if count > 0:# 查询表注释comment_query = f"""SELECT ISNULL(ep.value, '')FROM sys.extended_properties epWHERE ep.major_id = OBJECT_ID('[{schema}].[{table_name}]') AND ep.minor_id = 0"""cursor.execute(comment_query)result = cursor.fetchone()# 处理表注释为空的情况table_comment = result[0] if result else ""tables_with_data.append((schema, table_name, table_comment))except pymssql.Error as e:print(f"查询表 {schema}.{table_name} 时出现数据库错误: {str(e)}")if not tables_with_data:print("未找到有数据的表")else:output = []for schema, table_name, table_comment in tables_with_data:output.append(f"表名:{schema}.{table_name}")output.append(f"表注释:{table_comment}")output.append("表结构:")try:# 查询表结构及字段备注column_query = f"""SELECT c.COLUMN_NAME, c.DATA_TYPE, c.CHARACTER_MAXIMUM_LENGTH, c.IS_NULLABLE,ISNULL(ep.value, '') AS COLUMN_COMMENTFROM INFORMATION_SCHEMA.COLUMNS cLEFT JOIN sys.extended_properties ep ON ep.major_id = OBJECT_ID('[{schema}].[{table_name}]') AND ep.minor_id = c.ORDINAL_POSITIONWHERE c.TABLE_NAME = '{table_name}' AND c.TABLE_SCHEMA = '{schema}'"""cursor.execute(column_query)columns = cursor.fetchall()header = "列名 | 数据类型 | 最大长度 | 允许空 | 字段备注"separator = "--- | --- | --- | --- | ---"output.extend([header, separator])for col in columns:col_name, data_type, max_len, is_null, col_comment = colcol_info = [col_name,data_type,str(max_len) if max_len else "N/A","是" if is_null == 'YES' else "否",col_comment]output.append(" | ".join(col_info))except pymssql.Error as e:print(f"查询表 {schema}.{table_name} 的结构时出现数据库错误: {str(e)}")output.append("\n")# 写入文件with open(output_file, 'w', encoding='utf-8') as f:f.write('\n'.join(output))print(f"有数据的表结构已成功导出到 {output_file}")except pymssql.Error as e:print(f"数据库连接错误: {str(e)}")
except Exception as e:print(f"发生未知错误: {str(e)}")
finally:# 确保关闭数据库连接if 'conn' in locals():conn.close()
同样注意修改以下内容:
server = ''
database = ''
port = '1433'
username = ''
password = ''