用 Python 从 MySQL 信息模式生成 Protobuf 结构
在许多软件项目中，特别是使用 MySQL 数据库的项目中，通常使用 Protocol Buffers（Protobuf）进行高效的数据序列化。如果你发现自己需要将 MySQL 数据库架构信息转换为 Protobuf 消息，这个 Python 脚本可能是一个有价值的工具。
概览
这个 Python 脚本利用 SQLAlchemy 和 MySQL 数据库的信息模式，动态生成 Protobuf 消息结构。它从 MySQL 的 information_schema 中检索关于表和列的信息，然后创建相应的 Protobuf 消息定义。
功能
-
动态生成： 脚本根据指定 MySQL 数据库中的表和列动态生成 Protobuf 消息。
-
灵活的类型映射： 它智能地将 MySQL 数据类型映射到它们对应的 Protobuf 类型，为生成的 Protobuf 消息提供了灵活性和准确性。
-
结构化输出： 生成的 Protobuf 消息以结构化的方式组织，方便集成到你的 Protobuf 模式中。
如何使用
-
配置： 在脚本中设置
db_name和db_url的适当值，以匹配你的 MySQL 数据库。 -
运行脚本： 执行脚本，它将自动连接到数据库，检索模式信息，并生成 Protobuf 消息。
-
输出文件： 生成的 Protobuf 消息保存在
out_py.proto文件中，可以直接包含在你的 Protobuf 模式中。
依赖项
-
SQLAlchemy：Python 中强大的 SQL 工具包和对象关系映射（ORM）库。
-
MySQL Connector：Python 中用于 MySQL 的驱动程序。
结论
这个脚本简化了将 MySQL 数据库架构信息集成到 Protobuf 模式的过程，节省了时间，确保数据库和 Protobuf 数据模型之间的一致性。
随时根据项目需求定制脚本，并让它成为你软件开发旅程中有用的工具。
import re
from sqlalchemy import create_engine, Column as SQLColumn, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.decl_api import declarative_base
Base = declarative_base()
class Table(Base):
__tablename__ = 'TABLES'
TABLE_NAME = SQLColumn('TABLE_NAME', String, primary_key=True)
TABLE_SCHEMA = SQLColumn('TABLE_SCHEMA', String)
class MyColumn(Base):
__tablename__ = 'COLUMNS'
COLUMN_NAME = SQLColumn('COLUMN_NAME', String, primary_key=True)
DATA_TYPE = SQLColumn('DATA_TYPE', String)
COLUMN_COMMENT = SQLColumn('COLUMN_COMMENT', String)
TABLE_NAME = SQLColumn('TABLE_NAME', String)
TABLE_SCHEMA = SQLColumn('TABLE_SCHEMA', String)
def generate_message_structure(db_session, db_name, table):
columns = db_session.query(MyColumn).filter(MyColumn.TABLE_SCHEMA == db_name, MyColumn.TABLE_NAME == table).all()
txt = ""
for index, col in enumerate(columns):
if col.COLUMN_COMMENT:
txt += f" {pb_type(col.DATA_TYPE)} {camel_case(col.COLUMN_NAME)} = {index + 1}; //{col.COLUMN_COMMENT}\n"
else:
txt += f" {pb_type(col.DATA_TYPE)} {camel_case(col.COLUMN_NAME)} = {index + 1};\n"
txt = f"message {tb_name(table)} {{\n{txt}}}\n"
return txt
def tb_name(table_name):
words = re.split(r'_', table_name)
return ''.join(word.capitalize() for word in words)
def camel_case(s):
parts = re.split(r'_', s)
return parts[0].lower() + ''.join(word.capitalize() for word in parts[1:])
def pb_type(mysql_type):
if mysql_type in ["varchar", "char", "text", "mediumtext", "longtext"]:
return "string"
elif mysql_type in ["timestamp", "datetime", "date", "time"]:
return "google.protobuf.Timestamp"
elif mysql_type == "bigint":
return "int64"
elif mysql_type in ["int", "mediumint", "smallint", "tinyint"]:
return "int32"
elif mysql_type in ["double", "decimal"]:
return "double"
elif mysql_type == "float":
return "float"
elif mysql_type == "json":
return "google.protobuf.Any"
elif mysql_type in ["enum", "set"]:
return "string"
elif mysql_type in ["binary", "varbinary", "blob", "longblob", "mediumblob", "tinyblob"]:
return "bytes"
else:
return "string"
def generate_proto_file(db_session, db_name):
tables = db_session.query(Table).filter(Table.TABLE_SCHEMA == db_name).all()
message = """syntax = "proto3";
package pbdef;
import "google/protobuf/any.proto";
import "google/protobuf/timestamp.proto";
"""
for table in tables:
message += generate_message_structure(db_session, db_name, table.TABLE_NAME) + "\n"
with open("out_py.proto", "w", encoding='utf-8') as file:
file.write(message)
if __name__ == "__main__":
db_name = "dopai"
db_url = "mysql+mysqlconnector://lzj:[email protected]:3306/information_schema"
engine = create_engine(db_url)
# Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
db_session = Session()
generate_proto_file(db_session, db_name)
print("Protobuf structure has been saved to out.proto.")