如何用Python类统一管理企业数据的多场景字段需求？

问题背景 #

我在Python中用字典存储企业数据，字段表头如下：

HEADERS['companies-details'] = ["company_id","company_url","company_name",
"free_places","accepted_perc","domain_raw",
"domain_cleaned","email","email_domain","phone_main","phone_main_cleaned",
"founded_year","employee_count","employee_count_cleaned","employee_count_cat",
"revenue","industry_raw","industry_cat","industry_cat_prob","address_raw",
"address_street","postal_code","city","date" ];

目前生成CSV、创建SQL表、执行SQL插入、初始化数据时，需要重复编写维护这些字段，增删字段时要同步修改多处代码，极易出错。我希望创建一个Company类，统一实现：

• CSV表头（返回列表）
• SQL建表语句（返回字符串）
• SQL插入语句（返回字符串）
• SQL查询语句（返回字符串）
• 数据初始化与填充（在__init__中处理）
  我尝试用Pydantic的BaseModel和typing.Optional，想确认方向是否正确，并获取具体实现方式。

实现方案 #

用Pydantic BaseModel完全是正确的方向——它能让你在一处定义所有字段及其类型，后续所有操作都基于这个模型自动生成，彻底避免重复维护字段的问题。以下是具体实现：

1. 定义Company模型 #

导入必要模块，定义带类型注解的Company模型，用Optional标记允许为空的字段：

from pydantic import BaseModel
from typing import Optional, List
from datetime import date

class Company(BaseModel):
    company_id: str
    company_url: Optional[str] = None
    company_name: str
    free_places: Optional[int] = None
    accepted_perc: Optional[float] = None
    domain_raw: Optional[str] = None
    domain_cleaned: Optional[str] = None
    email: Optional[str] = None
    email_domain: Optional[str] = None
    phone_main: Optional[str] = None
    phone_main_cleaned: Optional[str] = None
    founded_year: Optional[int] = None
    employee_count: Optional[str] = None
    employee_count_cleaned: Optional[int] = None
    employee_count_cat: Optional[str] = None
    revenue: Optional[str] = None
    industry_raw: Optional[str] = None
    industry_cat: Optional[str] = None
    industry_cat_prob: Optional[float] = None
    address_raw: Optional[str] = None
    address_street: Optional[str] = None
    postal_code: Optional[str] = None
    city: Optional[str] = None
    date: date

    # 生成CSV表头
    @classmethod
    def get_csv_header(cls) -> List[str]:
        return list(cls.model_fields.keys())

    # 生成SQL建表语句
    @classmethod
    def get_create_table_sql(cls, table_name: str = "companies") -> str:
        type_mapping = {
            str: "VARCHAR(255)",
            int: "INT",
            float: "DECIMAL(10,2)",
            date: "DATE",
            Optional[str]: "VARCHAR(255)",
            Optional[int]: "INT",
            Optional[float]: "DECIMAL(10,2)",
            Optional[date]: "DATE"
        }
        columns = []
        for field_name, field in cls.model_fields.items():
            sql_type = type_mapping.get(field.annotation, "VARCHAR(255)")
            if field_name == "company_id":
                columns.append(f"`{field_name}` {sql_type} PRIMARY KEY")
            else:
                columns.append(f"`{field_name}` {sql_type}")
        return f"CREATE TABLE IF NOT EXISTS `{table_name}` ({', '.join(columns)});"

    # 生成SQL插入语句（单条数据）
    def get_insert_sql(self, table_name: str = "companies") -> tuple[str, list]:
        fields = list(self.model_fields.keys())
        placeholders = ["%s"] * len(fields)
        values = [getattr(self, field) for field in fields]
        return (
            f"""INSERT INTO `{table_name}` ({', '.join(f'`{f}`' for f in fields)}) 
VALUES ({', '.join(placeholders)});""",
            values
        )

    # 生成SQL查询语句（可指定字段，默认查所有）
    @classmethod
    def get_select_sql(cls, table_name: str = "companies", fields: Optional[List[str]] = None) -> str:
        target_fields = fields if fields else list(cls.model_fields.keys())
        selected_fields = ', '.join(f'`{f}`' for f in target_fields)
        return f"SELECT {selected_fields} FROM `{table_name}`;"

    # 从爬取字典快速初始化实例
    @classmethod
    def from_scraped_data(cls, data_dict: dict):
        filtered_data = {k: v for k, v in data_dict.items() if k in cls.model_fields}
        return cls(**filtered_data)

2. 使用示例 #

# 获取CSV表头
print(Company.get_csv_header())

# 生成建表SQL
create_sql = Company.get_create_table_sql()
print(create_sql)

# 初始化企业数据（数据采集场景）
scraped_data = {
    "company_id": "COMP001",
    "company_name": "Tech Corp",
    "date": date.today(),
    "city": "New York"
}
company = Company.from_scraped_data(scraped_data)

# 生成插入SQL和对应值
insert_sql, insert_values = company.get_insert_sql()
print(insert_sql)
print(insert_values)

# 生成指定字段的查询SQL
select_sql = Company.get_select_sql(fields=["company_id", "company_name", "city"])
print(select_sql)

方案优势 #

• 单一数据源：所有字段仅在Company模型中定义一次，增删字段只需修改模型，所有生成逻辑自动同步。
• 类型安全：Pydantic自动校验数据类型，避免非法数据流入。
• 扩展性强：后续新增功能（如Excel导出、JSON Schema生成），只需在模型上新增方法即可。

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