不使用sqlalchemy等框架如何将Python类转换为PostgreSQL数据表

第一步：定义基础字段类 #

先封装PostgreSQL字段类型对应的Python类，用于后续识别类的属性为数据库字段：

# 基础字段父类
class BaseField:
    def __init__(self, pg_type, **kwargs):
        self.pg_type = pg_type
        self.nullable = kwargs.get("nullable", False)
        self.primary_key = kwargs.get("primary_key", False)
        self.default = kwargs.get("default", None)

# 对应PG可变长字符类型
class CharField(BaseField):
    def __init__(self, max_length=255, **kwargs):
        super().__init__(f"varchar({max_length})", **kwargs)

# 对应PG整数类型
class IntegerField(BaseField):
    def __init__(self, **kwargs):
        super().__init__("int4", **kwargs)

# 对应PG自增主键类型
class SerialField(BaseField):
    def __init__(self, **kwargs):
        super().__init__("serial", primary_key=True, **kwargs)

第二步：编写类转建表SQL的工具函数 #

该函数会遍历目标类的所有属性，自动拼接为合法的PostgreSQL建表语句：

def generate_create_table_sql(model_cls) -> str:
    # 提取表名，默认取类名小写
    tablename = getattr(model_cls, "tablename", model_cls.__name__.lower())
    field_defs = []
    # 筛选类属性中的字段实例，拼接单条字段定义
    for attr_name, attr_value in model_cls.__dict__.items():
        if isinstance(attr_value, BaseField):
            single_def = [attr_name, attr_value.pg_type]
            if attr_value.primary_key:
                single_def.append("PRIMARY KEY")
            if not attr_value.nullable and not attr_value.primary_key:
                single_def.append("NOT NULL")
            if attr_value.default is not None:
                single_def.append(f"DEFAULT {attr_value.default}")
            field_defs.append(" ".join(single_def))
    # 拼接完整建表语句
    return f"CREATE TABLE IF NOT EXISTS {tablename} ({','.join(field_defs)});"

第三步：结合现有代码完成调用 #

自定义模型类示例 #

import psycopg2

class User:
    tablename = "users" # 注意user是PG保留关键字，建议改名为users避免转义
    id = SerialField()
    name = CharField(max_length=50, nullable=False)
    age = IntegerField(nullable=True)

完整调用流程 #

# 你已实现的建表函数
def create_table(
    sql_query: str, 
    conn: psycopg2.extensions.connection, 
    cur: psycopg2.extensions.cursor
) -> None:
    try:
        cur.execute(sql_query)
    except Exception as e:
        print(f"{type(e).__name__}: {e}")
        print(f"Query: {cur.query}")
        conn.rollback()
        cur.close()
    else:
        conn.commit()

# 数据库连接（替换为你自己的连接参数）
conn = psycopg2.connect(
    dbname="你的数据库名",
    user="你的数据库用户名",
    password="你的数据库密码",
    host="127.0.0.1",
    port="5432"
)
cur = conn.cursor()

# 生成建表SQL并执行
create_table_sql = generate_create_table_sql(User)
create_table(create_table_sql, conn, cur)

# 关闭资源
cur.close()
conn.close()

效果验证 #

执行完成后可在PostgreSQL命令行执行\d users查看生成的表结构，最终生成的建表语句如下：

CREATE TABLE IF NOT EXISTS users (id serial PRIMARY KEY,name varchar(50) NOT NULL,age int4);

内容的提问来源于stack exchange，提问作者Ivan