526互联

mysql table to proto message

发布时间 2023-12-19 14:26:32作者: NorseLZJ

用 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 模式中。

如何使用

  1. 配置： 在脚本中设置 db_name 和 db_url 的适当值，以匹配你的 MySQL 数据库。

  2. 运行脚本： 执行脚本，它将自动连接到数据库，检索模式信息，并生成 Protobuf 消息。

  3. 输出文件： 生成的 Protobuf 消息保存在 out_py.proto 文件中，可以直接包含在你的 Protobuf 模式中。

依赖项

  • SQLAlchemy：Python 中强大的 SQL 工具包和对象关系映射（ORM）库。

  • MySQL Connector：Python 中用于 MySQL 的驱动程序。

结论

这个脚本简化了将 MySQL 数据库架构信息集成到 Protobuf 模式的过程，节省了时间，确保数据库和 Protobuf 数据模型之间的一致性。

随时根据项目需求定制脚本，并让它成为你软件开发旅程中有用的工具。

```python

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:123456@192.168.31.54: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.")

```