playwrite/docs/DATABASE_ARCHITECTURE.md

数据库架构文档

概述

本文档描述了 Playwrite 自动化项目的数据库架构，包括设计模式、组件关系、数据流和数据库表结构。

系统简介

本项目是一个 Python 自动化框架，使用 Playwright 与中国 ERP 系统（用友 YonBIP）交互。数据库层采用抽象工厂模式和 DAO（Data Access Object）模式，支持多种数据库类型。

支持的数据库类型

• SQL Server (Microsoft SQL Server)

  • 驱动：ODBC Driver 18 for SQL Server
  • 连接库：pyodbc
  • 占位符：?

• MySQL

  • 连接库：mysql-connector-python
  • 占位符：%s

关键特性

• 多数据库支持：通过抽象基类实现统一的接口
• 自动 SQL 转换：处理不同数据库的表名格式差异
• 上下文管理器：支持 with 语句自动管理连接生命周期
• 批处理优化：根据数据库参数限制自动分批处理
• 事务管理：自动提交和回滚

---

架构设计

类层次结构

classDiagram
    class BaseDatabaseConnection {
        <<abstract>>
        +config: Dict
        +connection: Connection
        +connect() Connection
        +disconnect() void
        +execute_query(sql, params) List~Dict~
        +execute_update(sql, params) int
        +get_placeholder() string
    }

    class SQLServerConnection {
        +connect() pyodbc.Connection
        +disconnect() void
        +execute_query(sql, params) List~Dict~
        +execute_update(sql, params) int
        +get_placeholder() string
    }

    class MySQLConnection {
        +connect() MySQLConnection
        +disconnect() void
        +execute_query(sql, params) List~Dict~
        +execute_update(sql, params) int
        +get_placeholder() string
    }

    class BaseDAO {
        <<abstract>>
        #db: BaseDatabaseConnection
        #_db_type: DatabaseType
        +__enter__() BaseDAO
        +__exit__() void
        +_convert_sql(sql) string
        +_get_placeholder() string
    }

    class BIPUsersDAO {
        +authenticate(username, password) Dict
        +get_all_users() List~Dict~
        +create_user(username, password, user_type) bool
        +update_user_type(username, user_type) bool
        +update_password(username, new_password) bool
        +delete_user(username) bool
    }

    class DiscreteMaterialPlanDAO {
        +save_dataframe_with_replace(df) Dict
        +query_by_plan_number(plan_number) List~Dict~
        +query_by_plan_numbers(plan_numbers) List~Dict~
        +query_by_source_numbers(source_numbers) List~Dict~
        +get_statistics() Dict
    }

    class ProductionContractDataDAO {
        +query_by_总排号(总排号_list) List~Dict~
        +get_source_numbers_by_总排号(总排号_list) List~string~
    }

    class MaterialsToBeDeletedDAO {
        +insert_material(material_name, manager_name) bool
        +get_all_materials() List~Dict~
        +get_materials_by_manager(manager_name) List~Dict~
        +update_manager(material_name, old_manager, new_manager) bool
        +delete_material(material_name, manager_name) bool
        +get_statistics() Dict
    }

    class ConnectionFactory {
        <<factory>>
        +create_connection(db_type, config) BaseDatabaseConnection
        +create_from_config(database_config) BaseDatabaseConnection
    }

    class TableNameConverter {
        <<utility>>
        +to_mysql(table_name) string
        +to_sqlserver(table_name) string
        +convert_sql(sql, db_type) string
        +extract_table_names(sql) List~string~
    }

    BaseDatabaseConnection <|-- SQLServerConnection
    BaseDatabaseConnection <|-- MySQLConnection
    BaseDAO <|-- BIPUsersDAO
    BaseDAO <|-- DiscreteMaterialPlanDAO
    BaseDAO <|-- ProductionContractDataDAO
    BaseDAO <|-- MaterialsToBeDeletedDAO
    ConnectionFactory ..> BaseDatabaseConnection : creates
    BaseDAO ..> BaseDatabaseConnection : uses
    BaseDAO ..> TableNameConverter : uses

设计模式

1. 抽象工厂模式 (Abstract Factory)

   • ConnectionFactory 根据配置创建相应的数据库连接实例
   • 客户端代码无需关心具体实现类型

2. DAO 模式 (Data Access Object)

   • BaseDAO 提供统一的数据访问接口
   • 每个 DAO 类专注于特定表的数据操作
   • 封装 SQL 语句和数据库交互细节

3. 策略模式 (Strategy)

   • 不同的数据库连接类实现相同的接口
   • 运行时根据配置选择具体实现

4. 模板方法模式 (Template Method)

   • BaseDAO 定义通用的数据访问流程
   • 子类实现具体的业务逻辑

---

连接管理

连接工厂流程

flowchart TD
    A[Application calls get_connection] --> B[ConfigLoader.load]
    B --> C{Database Type?}
    C -->|SQL Server| D[ConnectionFactory.create_from_config]
    C -->|MySQL| D
    D --> E{db_type value}
    E -->|sqlserver| F[SQLServerConnection config]
    E -->|mysql| G[MySQLConnection config]
    F --> H[Connect to database]
    G --> H
    H --> I[Return connection instance]
    I --> J[Execute query/update]
    J --> K[Close connection]

配置加载流程

flowchart LR
    A[.env file] --> B[env_loader.py load_env_file]
    B --> C[ConfigLoader.load]
    C --> D[AppConfig.from_env]
    D --> E[DatabaseConfig.from_env]
    E --> F[DatabaseType from env]
    F --> G{DB_TYPE value}
    G -->|sqlserver| H[SQLServerConfig.from_env]
    G -->|mysql| I[MySQLConfig.from_env]
    H --> J[ConnectionFactory.create_from_config]
    I --> J

连接生命周期

sequenceDiagram
    participant App as Application
    participant CF as ConnectionFactory
    participant DB as DatabaseConnection
    participant SQL as SQL Database

    App->>CF: get_connection()
    CF->>CF: load database config
    CF->>DB: create_connection(db_type, config)
    DB->>SQL: connect()
    SQL-->>DB: connection object
    DB-->>CF: connection instance
    CF-->>App: return connection

    App->>DB: execute_query(sql, params)
    DB->>SQL: execute query
    SQL-->>DB: results
    DB-->>App: return results

    App->>DB: disconnect()
    DB->>SQL: close connection

---

数据访问层

DAO 操作序列图

sequenceDiagram
    participant App as Application
    participant DAO as BaseDAO
    participant CF as ConnectionFactory
    participant DB as DatabaseConnection
    participant SQL as SQL Database

    App->>DAO: with dao:
    DAO->>DAO: __enter__()
    DAO->>CF: get_connection()
    CF->>DB: create_connection()
    DB->>SQL: connect()
    DB-->>DAO: return connection

    DAO->>DAO: _convert_sql(sql)
    Note over DAO: TableNameConverter.convert_sql

    DAO->>DB: execute_query(sql, params)
    DB->>SQL: execute
    SQL-->>DB: results
    DB-->>DAO: return data

    DAO-->>App: return results

    App->>DAO: end with block
    DAO->>DAO: __exit__()
    DAO->>DB: disconnect()
    DB->>SQL: close connection

批处理流程

flowchart TD
    A[Batch Data] --> B{Database Type?}
    B -->|SQL Server| C[Calculate batch size]
    B -->|MySQL| D[Calculate batch size]

    C --> E[Max 72 records/batch]
    D --> F[Max 2000 records/batch]

    E --> G[Split into batches]
    F --> G

    G --> H[Execute first batch]
    H --> I{Has more data?}
    I -->|Yes| J[Next batch]
    J --> H
    I -->|No| K[Complete]

批处理限制说明：

• SQL Server: 最多 2100 个参数/查询。以 28 个字段的表为例，每批最多 72 条记录（2100 ÷ 28 ≈ 72）
• MySQL: 默认无硬限制，但建议每批不超过 2000 条记录以优化性能

错误处理流程

flowchart TD
    A[Execute Query/Update] --> B{Success?}
    B -->|Yes| C[Return Results]
    B -->|No| D[Catch Exception]
    D --> E{Is Update Operation?}
    E -->|Yes| F[Rollback Transaction]
    E -->|No| G[Skip Rollback]
    F --> H[Log Error]
    G --> H
    H --> I[Raise Exception to Caller]

---

数据库表结构

实体关系图

erDiagram
    BIPUsers ||--o{ DiscreteMaterialPlanData : creates
    BIPUsers ||--o{ MaterialsTypeToBeDeleted : manages
    BIPUsers ||--o{ MaterialsToBeDeleted : tracks
    ProductionContractData ||--o{ DiscreteMaterialPlanData : references

    BIPUsers {
        int ID PK
        string UserName
        string Password
        string UserType
        datetime CreateTime
    }

    DiscreteMaterialPlanData {
        int ID PK
        string Factory
        string MaterialStatus
        string PlanNumber
        string SourceNumber
        string MaterialType
        string ProductCode
        string ProductName
        string ProductUnit
        int ProductPlanQuantity
        string UseDepartment
        string Remark
        string Creator
        datetime CreateDate
        string Approver
        datetime ApproveDate
        int SequenceNumber
        string MaterialCode
        string MaterialName
        string Specification
        string Model
        string DrawingNumber
        string MaterialQuality
        int PlanQuantity
        string Unit
        datetime RequiredDate
        string Warehouse
        float UnitUsage
        int CumulativeOutputQuantity
        string BOMVersion
    }

    ProductionContractData {
        int ID PK
        string 总排号
        string 生产订单号
        int 序号
        string 订单号
        string 客户名称
        string 产品型号
    }

    MaterialsTypeToBeDeleted {
        int ID PK
        string MaterialName
        string ManagerName
    }

    MaterialsToBeDeleted {
        int ID PK
        string MaterialName
        string ManagerName
        string Reason
        datetime CreateDate
    }

表结构说明

1. BIPUsers (用户表)

存储系统用户信息和认证凭据。

字段名	类型	说明	约束
ID	int	用户 ID	主键
UserName	varchar	用户名	唯一
Password	varchar	密码	-
UserType	varchar	用户类型	'Admin', 'User', 'Guest'
CreateTime	datetime	创建时间	-

2. DiscreteMaterialPlanData (离散备料计划数据表)

存储生产备料计划详细信息。

字段名	类型	说明	约束
ID	int	记录 ID	主键
Factory	varchar	工厂	-
MaterialStatus	varchar	备料状态	-
PlanNumber	varchar	备料计划单号	-
SourceNumber	varchar	来源单号（生产订单号）	外键关联
MaterialType	varchar	备料类型	-
ProductCode	varchar	产品编码	-
ProductName	varchar	产品名称	-
ProductUnit	varchar	产品单位	-
ProductPlanQuantity	int	产品计划数量	默认 0
UseDepartment	varchar	用料部门	-
Remark	varchar	备注	-
Creator	varchar	制单人	关联 BIPUsers
CreateDate	datetime	制单日期	-
Approver	varchar	审批人	-
ApproveDate	datetime	审批日期	-
SequenceNumber	int	序号	默认 0
MaterialCode	varchar	材料编码	-
MaterialName	varchar	材料名称	-
Specification	varchar	规格	-
Model	varchar	型号	-
DrawingNumber	varchar	图号	-
MaterialQuality	varchar	物料材质	-
PlanQuantity	int	计划数量	默认 0
Unit	varchar	单位	-
RequiredDate	datetime	需用日期	-
Warehouse	varchar	发料仓库	-
UnitUsage	float	单位用量	默认 0.0
CumulativeOutputQuantity	int	累计出库数量	默认 0
BOMVersion	varchar	BOM 版本	-

3. ProductionContractData (生产合同数据表)

存储生产合同信息（26年压力表合同数据）。

字段名	类型	说明	约束
ID	int	记录 ID	主键
总排号	varchar	总排号	-
生产订单号	varchar	生产订单号	-
序号	int	序号	-
订单号	varchar	订单号	-
客户名称	varchar	客户名称	-
产品型号	varchar	产品型号	-

4. MaterialsTypeToBeDeleted (待删除物料类型表)

存储需要删除的物料及其管理员信息。

字段名	类型	说明	约束
ID	int	记录 ID	主键
MaterialName	varchar	物料名称	-
ManagerName	varchar	管理员名称	关联 BIPUsers

5. MaterialsToBeDeleted (待删除物料记录表)

存储待删除物料的详细记录。

字段名	类型	说明	约束
ID	int	记录 ID	主键
MaterialName	varchar	物料名称	-
ManagerName	varchar	管理员名称	关联 BIPUsers
Reason	varchar	删除原因	-
CreateDate	datetime	创建日期	-

---

SQL 方言处理

SQL 转换流程

flowchart TD
    A[SQL Server Query] --> B{Contains schema dot table format?}
    B -->|Yes| C[TableNameConverter.convert_sql]
    B -->|No| D[Skip conversion]
    C --> E{Target Database Type?}
    E -->|MySQL| F[Replace schema.table with schema_table]
    E -->|SQL Server| G[Keep original format]
    D --> G
    F --> H{Placeholder conversion}
    G --> H
    H --> I{Database Type?}
    I -->|MySQL| J[Replace question mark with percent s]
    I -->|SQL Server| K[Keep question mark placeholders]
    J --> L[Execute query]
    K --> L

表名转换规则

SQL Server 格式	MySQL 格式	示例
[dbo].[TableName]	dbo_TableName	[dbo].[BIPUsers] → dbo_BIPUsers
[schema].[table name]	schema_table name	[productionContractData].[26年压力表合同数据] → productionContractData_26年压力表合同数据
TableName	dbo_TableName	BIPUsers → dbo_BIPUsers

占位符差异

数据库	占位符	示例
SQL Server	?	WHERE UserName = ? AND Password = ?
MySQL	%s	WHERE UserName = %s AND Password = %s

列名格式差异

数据库	列名格式	示例
SQL Server	[ColumnName]	SELECT [ID], [UserName] FROM [dbo].[BIPUsers]
MySQL	ColumnName	SELECT ID, UserName FROM dbo_BIPUsers

---

数据流

查询执行流程

flowchart TD
    A[User calls DAO method] --> B[DAO enters context with 'with']
    B --> C[DAO __enter__]
    C --> D[Get connection from factory]
    D --> E[Database connect]
    E --> F[Convert SQL if needed]
    F --> G[Build placeholders]
    G --> H[Execute query]
    H --> I{Success?}
    I -->|Yes| J[Convert results to dicts]
    I -->|No| K[Raise exception]
    J --> L[Return data to user]
    K --> M[DAO __exit__ cleanup]
    L --> M
    M --> N[Database disconnect]

保存数据流程

flowchart TD
    A[User provides DataFrame] --> B[Check if empty]
    B -->|Empty| C[Return zero counts]
    B -->|Has data| D[Remove duplicates]
    D --> E[Extract unique plan numbers]
    E --> F[Delete existing records]
    F --> G[Split into batches]
    G --> H[Insert batch 1]
    H --> I{More batches?}
    I -->|Yes| J[Insert next batch]
    I -->|No| K[Return statistics]
    J --> H

权限过滤流程

flowchart TD
    A[User queries materials] --> B{User authenticated?}
    B -->|No| C[Return empty result]
    B -->|Yes| D{User type?}
    D -->|Admin| E[Return all materials]
    D -->|User| F[Get user's materials]
    D -->|Guest| G[Return limited materials]
    F --> H[Filter by ManagerName]
    H --> I[Return filtered result]
    G --> I
    E --> J[Return unfiltered result]

---

代码示例

连接使用示例

from db.connection import get_connection

# 使用上下文管理器自动管理连接
with get_connection() as db:
    results = db.execute_query(
        "SELECT * FROM [dbo].[BIPUsers] WHERE UserType = ?",
        ('Admin',)
    )
    for user in results:
        print(f"User: {user['UserName']}")

# 连接自动关闭

DAO 使用示例

from db.bip_users_dao import BIPUsersDAO

# 用户认证
dao = BIPUsersDAO()
user = dao.authenticate("admin", "password123")
if user:
    print(f"Authenticated: {user['username']}")

# 获取所有用户
with dao:
    users = dao.get_all_users()
    for user in users:
        print(f"{user['username']} - {user['user_type']}")

# 创建新用户
success = dao.create_user(
    username="newuser",
    password="pass123",
    user_type="User"
)

配置示例

.env 文件配置

# 数据库类型 (sqlserver 或 mysql)
DB_TYPE=mysql

# SQL Server 配置
DB_SERVER=192.168.110.114
DB_NAME=CompanyDB
DB_USERNAME=peng
DB_PASSWORD=your_password
DB_SQLSERVER_DRIVER=ODBC Driver 18 for SQL Server
DB_TRUST_SERVER_CERTIFICATE=yes

# MySQL 配置
DB_MYSQL_HOST=192.168.31.83
DB_MYSQL_PORT=3306
DB_MYSQL_CHARSET=utf8mb4

# ERP 配置
ERP_URL=https://68.11.34.30:8082/
ERP_USERNAME=your_username
ERP_PASSWORD=your_password
ERP_HEADLESS=false

代码中加载配置

from config.loader import ConfigLoader

# 加载配置
config = ConfigLoader.load()

# 访问数据库配置
db_config = config.database
print(f"Database type: {db_config.db_type}")
print(f"Database name: {db_config.database}")

# 访问 ERP 配置
erp_config = config.erp
print(f"ERP URL: {erp_config.url}")

---

批处理限制详解

SQL Server 参数限制

SQL Server 对每个 SQL 查询的参数数量有硬性限制：最多 2100 个参数。

计算示例

对于 DiscreteMaterialPlanData 表（28 个字段）：

最大记录数 = floor(2100 / 28) = 72 条记录/批次

实现代码

def _batch_insert(self, db, df: pd.DataFrame, batch_size: int = 72) -> int:
    """SQL Server: 最多 72 条记录/批次"""
    total_inserted = 0
    records = self._convert_df_to_records(df)

    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        for record in batch:
            db.execute_update(sql, record)
            total_inserted += 1

    return total_inserted

MySQL 批处理

MySQL 没有硬性参数限制，但建议每批不超过 2000 条记录。

计算示例

建议批次大小 = 2000 条记录/批次

实现代码

def _batch_insert(self, db, df: pd.DataFrame, batch_size: int = 2000) -> int:
    """MySQL: 最多 2000 条记录/批次"""
    total_inserted = 0
    records = self._convert_df_to_records(df)

    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        for record in batch:
            db.execute_update(sql, record)
            total_inserted += 1

    return total_inserted

IN 子句限制

对于包含 IN 子句的查询，也需要批处理：

def _delete_by_plan_numbers(self, db, plan_numbers: List[str]) -> int:
    """分批删除以避免参数限制"""
    batch_size = 1000  # 安全限制
    total_deleted = 0

    for i in range(0, len(plan_numbers), batch_size):
        batch = plan_numbers[i:i + batch_size]
        placeholders = ','.join(['?' for _ in batch])
        sql = f"DELETE FROM {table_name} WHERE PlanNumber IN ({placeholders})"
        deleted = db.execute_update(sql, tuple(batch))
        total_deleted += deleted

    return total_deleted

---

事务管理

自动提交和回滚

flowchart TD
    A[Start Transaction] --> B[Execute Query]
    B --> C{Success?}
    C -->|Yes| D[Commit Transaction]
    C -->|No| E[Rollback Transaction]
    D --> F[Return Results]
    E --> G[Log Error]
    G --> H[Raise Exception]

实现代码

SQL Server

def execute_update(self, sql: str, params: Optional[tuple] = None) -> int:
    cursor = self.connection.cursor()
    try:
        if params:
            cursor.execute(sql, params)
        else:
            cursor.execute(sql)

        self.connection.commit()  # 自动提交
        return cursor.rowcount
    except pyodbc.Error as e:
        self.connection.rollback()  # 自动回滚
        print(f"执行失败，已回滚: {e}")
        raise
    finally:
        cursor.close()

MySQL

def execute_update(self, sql: str, params: Optional[tuple] = None) -> int:
    cursor = self.connection.cursor()
    try:
        if params:
            cursor.execute(sql, params)
        else:
            cursor.execute(sql)

        self.connection.commit()  # 自动提交
        return cursor.rowcount
    except Error as e:
        self.connection.rollback()  # 自动回滚
        print(f"执行失败，已回滚: {e}")
        raise
    finally:
        cursor.close()

---

错误处理

常见错误类型

错误类型	原因	处理方式
连接失败	网络问题、凭据错误	记录日志，抛出异常
参数超限	超过 2100 参数限制	自动分批处理
表名格式错误	SQL 方言不匹配	自动转换表名
约束违反	重复键、外键约束	回滚事务，返回错误
超时	查询执行时间过长	增加超时时间或优化查询

错误处理最佳实践

def safe_database_operation():
    try:
        with get_connection() as db:
            # 执行数据库操作
            results = db.execute_query(sql, params)
            return results
    except pyodbc.Error as e:
        # 数据库特定错误
        print(f"Database error: {e}")
        # 记录到日志文件
        log_error(e)
        raise
    except Exception as e:
        # 通用错误
        print(f"Unexpected error: {e}")
        log_error(e)
        raise

---

性能优化建议

1. 使用批处理

对于大量数据插入/更新，始终使用批处理：

# 推荐：批处理
for i in range(0, len(records), batch_size):
    batch = records[i:i + batch_size]
    db.execute_batch(sql, batch)

# 避免：逐条插入
for record in records:
    db.execute_update(sql, record)

2. 使用索引

确保常用查询字段有索引：

CREATE INDEX idx_plan_number ON DiscreteMaterialPlanData(PlanNumber)
CREATE INDEX idx_source_number ON DiscreteMaterialPlanData(SourceNumber)
CREATE INDEX idx_manager_name ON MaterialsTypeToBeDeleted(ManagerName)

3. 使用连接池

对于频繁的数据库操作，考虑使用连接池：

from db.connection_pool import ConnectionPool

pool = ConnectionPool(max_connections=5)
with pool.get_connection() as db:
    results = db.execute_query(sql, params)

4. 优化查询

• 只选择需要的列
• 使用 WHERE 子句过滤数据
• 避免使用 SELECT *

# 推荐
sql = "SELECT ID, UserName FROM dbo_BIPUsers WHERE UserType = ?"

# 避免
sql = "SELECT * FROM dbo_BIPUsers"

---

安全考虑

1. 参数化查询

始终使用参数化查询防止 SQL 注入：

# 推荐：参数化查询
sql = "SELECT * FROM dbo_BIPUsers WHERE UserName = ?"
results = db.execute_query(sql, (username,))

# 避免：字符串拼接
sql = f"SELECT * FROM dbo_BIPUsers WHERE UserName = '{username}'"  # 危险！

2. 密码管理

• 不要在代码中硬编码密码
• 使用环境变量或配置文件
• 考虑使用密钥管理服务

3. 最小权限原则

数据库用户应该只有必要的权限：

-- 只授予必要的权限
GRANT SELECT, INSERT, UPDATE ON DiscreteMaterialPlanData TO app_user;
GRANT SELECT ON BIPUsers TO app_user;

4. 连接字符串安全

• 不要在日志中记录连接字符串
• 使用加密存储敏感信息
• 定期轮换凭据

---

维护指南

添加新 DAO 类

1. 创建新的 DAO 类继承 BaseDAO
2. 实现业务方法
3. 使用 _convert_sql() 处理表名
4. 使用 _get_placeholder() 处理参数

from db.base_dao import BaseDAO

class NewTableDAO(BaseDAO):
    def get_by_id(self, record_id: int) -> Dict:
        table_name = self._convert_sql('[dbo].[NewTable]')
        placeholder = self._get_placeholder()
        sql = f"SELECT * FROM {table_name} WHERE ID = {placeholder}"

        with get_connection() as db:
            results = db.execute_query(sql, (record_id,))
            return results[0] if results else None

添加新数据库支持

1. 创建新的连接类继承 BaseDatabaseConnection
2. 实现所有抽象方法
3. 在 ConnectionFactory 中注册
4. 更新 DatabaseType 枚举

# 1. 创建连接类
class PostgreSQLConnection(BaseDatabaseConnection):
    def connect(self):
        # 实现连接逻辑
        pass

    # 实现其他抽象方法...

# 2. 更新枚举
class DatabaseType(str, Enum):
    SQLSERVER = "sqlserver"
    MYSQL = "mysql"
    POSTGRESQL = "postgresql"

# 3. 更新工厂
def create_connection(db_type, config):
    if db_type == DatabaseType.POSTGRESQL:
        return PostgreSQLConnection(config)
    # 其他类型...

更新文档

当代码变更时，及时更新本文档：

• 更新类图和关系图
• 添加新的表结构说明
• 更新代码示例
• 记录新的限制和最佳实践

---

附录

相关文件

文件路径	说明
db/base_connection.py	数据库连接抽象基类
db/sqlserver_connection.py	SQL Server 连接实现
db/mysql_connection.py	MySQL 连接实现
db/base_dao.py	DAO 基类
db/connection_factory.py	连接工厂
db/table_name_converter.py	表名转换工具
db/bip_users_dao.py	用户 DAO
db/discrete_material_plan_dao.py	备料计划 DAO
db/production_contract_data_dao.py	合同数据 DAO
db/materials_to_be_deleted_dao.py	待删除物料 DAO
db/connection.py	连接获取辅助函数
config/schema.py	配置结构定义
config/env_loader.py	环境变量加载器
config/loader.py	配置加载器

依赖项

pyodbc>=4.0.0          # SQL Server 支持
mysql-connector-python>=8.0.0  # MySQL 支持
python-dotenv>=0.19.0   # 环境变量管理
pandas>=1.3.0          # 数据处理

参考资料

• SQL Server Documentation
• MySQL Documentation
• pyodbc Documentation
• MySQL Connector/Python
• DAO Pattern
• Abstract Factory Pattern

---

文档版本: 1.0 最后更新: 2026-02-09 维护者: Development Team