SQLAlchemy 2.0中create_all()未创建表的问题求助

你的代码存在几个关键问题导致表创建失败，以下是具体分析和修复：

核心问题1：错误处理Engine与Connection #

你在connect方法中将self.db.connect()返回的Connection实例赋值给了self.engine，但Base.metadata.create_all更适合直接使用Engine实例操作。同时，Connection默认处于事务中，DDL操作可能未自动提交（部分数据库如PostgreSQL需要显式提交），导致表创建后回滚。

核心问题2：冗余的表存在性判断 #

Base.metadata.create_all会自动检查目标表是否存在，无需手动循环判断，手动判断还可能因事务上下文问题出现误判（比如事务未提交时，判断表不存在，实际创建后又因回滚消失）。

次要问题：closed字段非空约束 #

closed字段未设置nullable=True，也没有默认值，创建表时会触发数据库的非空约束错误，只是你的代码未捕获这个异常，导致日志误报创建成功。

修复后的完整代码 #

import logging
import logging.handlers

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(name)-5s] [%(levelname)-3s] %(message)s'
)

from typing import List, Dict
import util.utility as util

import sqlalchemy
from sqlalchemy import DateTime, func, exc
from sqlalchemy.engine import Engine
from sqlalchemy.orm import relationship, sessionmaker, Mapped, DeclarativeBase, mapped_column
from datetime import datetime

config: Dict = util.read_config("../static/configurations/custom.yaml")

class Base(DeclarativeBase):
    pass

class Attribute_Details(Base):
    __tablename__ = "attribute_details"
    id: Mapped[int] = mapped_column(primary_key=True)
    description: Mapped[str] = mapped_column()
    opened: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
    # 允许closed字段为空，或根据需求设置默认值
    closed: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    report_attr: Mapped["Report_Attributes"] = relationship(back_populates="details")

class Database:
    engine: Engine
    session_maker: sessionmaker
    __connected: bool = False

    def __init__(self, con_str: str):
        self.logger = logging.getLogger(Database.__name__)
        self.logger.info("Connecting to DB: {}".format(con_str))
        # 创建Engine实例，直接赋值给self.engine
        self.engine = sqlalchemy.create_engine(con_str, echo=False)
        # 直接绑定Engine到sessionmaker
        self.session_maker = sessionmaker(bind=self.engine)

    def connect(self) -> bool:
        """验证数据库连接可用性"""
        try:
            # 用connect()验证连接，不需要保留Connection实例
            with self.engine.connect() as conn:
                conn.execute(sqlalchemy.text("SELECT 1"))
            self.logger.info("DB connection established")
            return True
        except exc.SQLAlchemyError as e:
            self.logger.error(f"DB connection failed: {str(e)}")
            return False

    def create_tables(self):
        """创建所有未存在的表"""
        try:
            # create_all自动检查并创建不存在的表，无需循环
            Base.metadata.create_all(self.engine)
            # 遍历所有表，打印创建状态
            for table in Base.metadata.tables.values():
                if self.engine.dialect.has_table(self.engine, table.name):
                    self.logger.info(f"DB Table {table.name} exists or was created")
        except exc.SQLAlchemyError as e:
            self.logger.error(f"Failed to create tables: {str(e)}")

db = Database(config['db_con_test'])
if db.connect():
    db.create_tables()

关键修改点说明 #

1. Engine与Connection分离：__init__中直接将create_engine返回的Engine赋值给self.engine，connect方法仅用于验证连接可用性，不保留Connection实例。
2. 简化表创建逻辑：移除循环判断，直接调用Base.metadata.create_all(self.engine)，由SQLAlchemy自动处理表存在性检查。
3. 修复closed字段约束：添加nullable=True并将类型改为Mapped[datetime | None]，符合SQLAlchemy 2.0的类型注解规范，避免非空约束错误。
4. 增加异常捕获：在create_tables中添加异常捕获，能准确打印表创建失败的原因，避免日志误报。

内容的提问来源于stack exchange，提问作者po.pe