在现代软件开发中数据库结构文档的维护往往是一项繁琐而重要的工作。手动编写和更新数据库文档不仅耗时还容易出错。今天我将分享一个使用 Python 自动分析 PostgreSQL 数据库结构并生成完整 Markdown 文档的解决方案。完整代码实现以下是完整的 Python 脚本它能够连接到 PostgreSQL 数据库分析所有表结构并生成详细的 Markdown 文档。importpsycopg2fromurllib.parseimporturlparsefromcontextlibimportcontextmanagercontextmanagerdefget_db_connection(connection_url):数据库连接上下文管理器try:parsed_urlurlparse(connection_url)hostparsed_url.hostname portparsed_url.portor5432databaseparsed_url.path[1:]userparsed_url.username passwordparsed_url.password connpsycopg2.connect(hosthost,portport,databasedatabase,useruser,passwordpassword)yieldconnexceptpsycopg2.Errorase:raiseException(f数据库连接失败:{str(e)})exceptExceptionase:raiseException(f连接参数解析失败:{str(e)})finally:try:conn.close()except:pass# 防止连接未成功建立时的错误defget_table_details(cursor,table_name):获取单个表的详细信息# 获取列信息cursor.execute( SELECT c.column_name, c.data_type, CASE WHEN c.character_maximum_length IS NOT NULL THEN c.data_type || ( || c.character_maximum_length || ) ELSE c.data_type END AS full_data_type, c.is_nullable, c.column_default, col_description((c.table_schema||.||c.table_name)::regclass, c.ordinal_position) AS column_comment FROM information_schema.columns c WHERE c.table_name %s ORDER BY c.ordinal_position; ,(table_name,))columnscursor.fetchall()# 获取主键信息cursor.execute( SELECT kcu.column_name FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu ON tc.constraint_name kcu.constraint_name AND tc.table_name kcu.table_name WHERE tc.table_name %s AND tc.constraint_type PRIMARY KEY ORDER BY kcu.ordinal_position; ,(table_name,))primary_keys[row[0]forrowincursor.fetchall()]# 获取外键信息cursor.execute( SELECT tc.constraint_name, kcu.column_name, ccu.table_name AS foreign_table_name, ccu.column_name AS foreign_column_name FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu ON tc.constraint_name kcu.constraint_name AND tc.table_name kcu.table_name JOIN information_schema.constraint_column_usage ccu ON ccu.constraint_name tc.constraint_name WHERE tc.table_name %s AND tc.constraint_type FOREIGN KEY ORDER BY kcu.ordinal_position; ,(table_name,))foreign_keyscursor.fetchall()# 获取唯一约束信息cursor.execute( SELECT tc.constraint_name, string_agg(kcu.column_name, , ORDER BY kcu.ordinal_position) AS columns FROM information_schema.table_constraints tc JOIN information_schema.key_column_usage kcu ON tc.constraint_name kcu.constraint_name AND tc.table_name kcu.table_name WHERE tc.table_name %s AND tc.constraint_type UNIQUE GROUP BY tc.constraint_name; ,(table_name,))unique_constraintscursor.fetchall()# 获取索引信息包括普通索引和唯一索引cursor.execute( SELECT t.relname as table_name, i.relname as index_name, array_to_string(array_agg(a.attname), , ) as column_names, ix.indisunique as is_unique, ix.indisprimary as is_primary FROM pg_class t, pg_class i, pg_index ix, pg_attribute a WHERE t.oid ix.indrelid AND i.oid ix.indexrelid AND a.attrelid t.oid AND a.attnum ANY(ix.indkey) AND t.relkind r AND t.relname %s GROUP BY t.relname, i.relname, ix.indisunique, ix.indisprimary ORDER BY t.relname, i.relname; ,(table_name,))indexescursor.fetchall()# 获取触发器信息cursor.execute( SELECT t.tgname as trigger_name, pg_get_functiondef(t.tgfoid) as trigger_function_def, CASE WHEN (t.tgtype 2) ! 0 THEN BEFORE WHEN (t.tgtype 4) ! 0 THEN AFTER WHEN (t.tgtype 8) ! 0 THEN INSTEAD OF END as trigger_timing, CASE WHEN (t.tgtype 16) ! 0 THEN INSERT WHEN (t.tgtype 32) ! 0 THEN DELETE WHEN (t.tgtype 64) ! 0 THEN UPDATE WHEN (t.tgtype 128) ! 0 THEN TRUNCATE END as trigger_event, CASE WHEN (t.tgtype 1) ! 0 THEN ROW ELSE STATEMENT END as trigger_level FROM pg_trigger t JOIN pg_class c ON c.oid t.tgrelid WHERE c.relname %s AND NOT t.tgisinternal ORDER BY t.tgname; ,(table_name,))triggerscursor.fetchall()# 获取表统计信息cursor.execute( SELECT n_tup_ins, n_tup_upd, n_tup_del, n_tup_hot_upd, n_live_tup, n_dead_tup, last_vacuum, last_autovacuum, last_analyze, last_autoanalyze, pg_size_pretty(pg_table_size(c.oid)) as table_size, pg_size_pretty(pg_indexes_size(c.oid)) as indexes_size, pg_size_pretty(pg_total_relation_size(c.oid)) as total_size FROM PG_CLASS c LEFT JOIN PG_STAT_USER_TABLES s ON c.oid s.relid WHERE c.relname %s; ,(table_name,))statisticscursor.fetchone()returncolumns,primary_keys,foreign_keys,unique_constraints,indexes,triggers,statisticsdefconnect_and_get_schema(connection_url): 使用URL连接数据库并获取完整的表结构信息 withget_db_connection(connection_url)asconn:withconn.cursor()ascursor:# 获取所有表名包括schemacursor.execute( SELECT table_schema, table_name FROM information_schema.tables WHERE table_type BASE TABLE AND table_schema NOT IN (information_schema, pg_catalog) ORDER BY table_schema, table_name; )tables_with_schemascursor.fetchall()# 获取视图信息cursor.execute( SELECT table_schema, table_name, view_definition FROM information_schema.views WHERE table_schema NOT IN (information_schema, pg_catalog) ORDER BY table_schema, table_name; )views_infocursor.fetchall()# 获取函数信息cursor.execute( SELECT n.nspname AS schema_name, p.proname AS function_name, pg_get_function_arguments(p.oid) AS arguments, t.typname AS return_type, l.lanname AS language FROM pg_proc p LEFT JOIN pg_namespace n ON p.pronamespace n.oid LEFT JOIN pg_type t ON p.prorettype t.oid LEFT JOIN pg_language l ON p.prolang l.oid WHERE n.nspname NOT IN (information_schema, pg_catalog) AND p.prokind f ORDER BY n.nspname, p.proname; )functions_infocursor.fetchall()# 为每个表收集结构信息table_structures{}forschema_name,table_nameintables_with_schemas:full_table_namef{schema_name}.{table_name}columns,primary_keys,foreign_keys,unique_constraints,indexes,triggers,statisticsget_table_details(cursor,table_name)table_structures[(schema_name,table_name)]{columns:columns,primary_keys:primary_keys,foreign_keys:foreign_keys,unique_constraints:unique_constraints,indexes:indexes,triggers:triggers,statistics:statistics}return{tables:table_structures,views:views_info,functions:functions_info}defgenerate_table_schema_markdown(table_name,table_info,schema_namepublic): 为指定表生成完整的Schema信息 columnstable_info[columns]primary_keystable_info[primary_keys]foreign_keystable_info[foreign_keys]unique_constraintstable_info[unique_constraints]indexestable_info[indexes]triggerstable_info[triggers]statisticstable_info[statistics]markdown_contentf###{table_name}\n\n# 字段信息表格markdown_content#### 字段信息\n\nmarkdown_content| 字段名 | 类型 | 是否为空 | 默认值 | 主键 | 注释 |\nmarkdown_content|--------|------|----------|--------|------|------|\nforcolincolumns:col_name,data_type,full_data_type,is_nullable,default_val,commentcol# 检查是否为主键is_pk✓ifcol_nameinprimary_keyselse# 清理默认值显示ifdefault_val:clean_defaultdefault_val.replace(::text,).replace(::integer,).replace(::bigint,)else:clean_default# 处理注释ifcommentisNone:commentmarkdown_contentf|{col_name}|{full_data_type}|{is_nullable}|{clean_default}|{is_pk}|{comment}|\nmarkdown_content\n# 外键信息ifforeign_keys:markdown_content#### 外键约束\n\nmarkdown_content| 约束名 | 字段 | 引用表 | 引用字段 |\nmarkdown_content|--------|------|--------|--------|\nforconstraint_name,col_name,ref_table,ref_columninforeign_keys:markdown_contentf|{constraint_name}|{col_name}|{ref_table}|{ref_column}|\nmarkdown_content\n# 唯一约束信息ifunique_constraints:markdown_content#### 唯一约束\n\nmarkdown_content| 约束名 | 字段 |\nmarkdown_content|--------|------|\nforconstraint_name,columns_strinunique_constraints:markdown_contentf|{constraint_name}|{columns_str}|\nmarkdown_content\n# 索引信息ifindexes:markdown_content#### 索引\n\nmarkdown_content| 索引名 | 字段 | 类型 |\nmarkdown_content|--------|------|------|\nfor_,index_name,column_names,is_unique,is_primaryinindexes:# 跳过主键索引已由主键约束自动创建ifis_primary:index_type主键索引elifis_unique:index_type唯一索引else:index_type普通索引markdown_contentf|{index_name}|{column_names}|{index_type}|\nmarkdown_content\n# 触发器信息iftriggers:markdown_content#### 触发器\n\nmarkdown_content| 触发器名 | 触发时机 | 触发事件 | 触发级别 |\nmarkdown_content|----------|----------|----------|----------|\nfortrigger_name,func_def,timing,event,levelintriggers:markdown_contentf|{trigger_name}|{timing}|{event}|{level}|\nmarkdown_content\n# 统计信息ifstatistics:markdown_content#### 表统计信息\n\nmarkdown_content| 统计项 | 值 |\nmarkdown_content|--------|----|\nn_tup_ins,n_tup_upd,n_tup_del,n_tup_hot_upd,n_live_tup,n_dead_tup,\ last_vacuum,last_autovacuum,last_analyze,last_autoanalyze,\ table_size,indexes_size,total_sizestatistics markdown_contentf| 插入记录数 |{n_tup_insor0}|\nmarkdown_contentf| 更新记录数 |{n_tup_updor0}|\nmarkdown_contentf| 删除记录数 |{n_tup_delor0}|\nmarkdown_contentf| 活跃记录数 |{n_live_tupor0}|\nmarkdown_contentf| 死亡记录数 |{n_dead_tupor0}|\nmarkdown_contentf| 表大小 |{table_sizeor未知}|\nmarkdown_contentf| 索引大小 |{indexes_sizeor未知}|\nmarkdown_contentf| 总大小 |{total_sizeor未知}|\niflast_analyze:markdown_contentf| 最后分析时间 |{last_analyze}|\nmarkdown_content\nreturnmarkdown_contentdefgenerate_complete_schema_documentation(connection_url): 生成包含每张表完整Schema信息的Markdown文档 schema_dataconnect_and_get_schema(connection_url)markdown_content# 数据库结构文档\n\n# 表列表概述markdown_content## 表列表\n\nfor(schema,table_name),_inschema_data[tables].items():markdown_contentf- [{table_name}](#{table_name.lower()})\nmarkdown_content\n# 视图列表ifschema_data[views]:markdown_content## 视图列表\n\nforschema,view_name,_inschema_data[views]:markdown_contentf- [{view_name}](#{view_name.lower()}-view)\nmarkdown_content\n# 函数列表ifschema_data[functions]:markdown_content## 函数列表\n\nforschema,func_name,args,return_type,langinschema_data[functions]:markdown_contentf-{schema}.{func_name}({args}) -{return_type}[{lang}]\nmarkdown_content\n# 详细表结构信息markdown_content## 详细表结构\n\nfor(schema_name,table_name),infoinschema_data[tables].items():markdown_contentgenerate_table_schema_markdown(table_name,info,schema_name)# 视图详细信息ifschema_data[views]:markdown_content## 视图定义\n\nforschema,view_name,definitioninschema_data[views]:markdown_contentf###{view_name}(View)\n\nmarkdown_contentf**定义:**\nsql\n{definition}\n\n\nreturnmarkdown_content# 主程序if__name____main__:# 根据提供的参数构建连接URLconnection_urlpostgresql://postgres:123456172.29.44.161:5432/mydatabasetry:print(正在连接数据库并获取表结构...)markdown_contentgenerate_complete_schema_documentation(connection_url)# 保存到文件withopen(schema.md,w,encodingutf-8)asf:f.write(markdown_content)print(数据库表结构文档已生成到: schema.md)# 统计信息table_countlen([lineforlineinmarkdown_content.split(\n)ifline.startswith(### )])print(f共处理了{table_count}个表/视图)exceptExceptionase:print(f错误:{e})print(请确保:)print(- 数据库服务正在运行)print(- 连接参数正确)print(- 有足够的权限访问数据库)print(- PostgreSQL版本兼容)安装依赖在运行此脚本之前您需要安装psycopg2库pipinstallpsycopg2如何使用修改连接参数在主程序部分修改connection_url变量填入您的数据库连接信息运行脚本执行python script_name.py查看结果脚本会在当前目录生成database_schema.md文件结论这个工具为 PostgreSQL 数据库的文档管理提供了一个完整、自动化的解决方案。通过简单的配置就能生成详尽的数据库结构文档大大提高了开发效率和文档质量。无论是个人项目还是团队协作这个工具都能发挥重要作用。