python - SQLAlchemy mssql: DropCreate Table on the Fly - Stack Overflow

admin管理员组

文章数量:1357620

I've created a SQLAlchemy (2.0.39), Python (3.10.11) definition on the fly:

    dynamic_class = type(table_name, (Base,), attributes)

This is working, now I want to drop any existing table and recreate it using the current definition (and where the problems kick in)

    log.info(f"Dropping Table: mygrc.{table_name}")
    dynamic_class.__table__.drop(dbEngine, checkfirst=True)

    log.info(f"Creating Table: mygrc.{table_name}")
    Base.metadata.create_all(dbEngine, tables=[dynamic_class.__table__])

Both the drop and create statements give me errors:

No results. Previous SQL was not a query.

If I try to pass the query direct to execute, it doesn't error but it doesn't execute the command either (even with a commit)

    with dbEngine.connect() as connection:
        create_table_statement = str(dynamic_class.__table__pile(dbEngine))
        connection.execute(text(create_table_statement))

create_table_statement gives me a valid create statement.

Any ideas?

Track trace:

Traceback (most recent call last):
  File "c:\Users\username\Documents\Coding\azure-pipeline\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1980, in _exec_single_context
    result = context._setup_result_proxy()
  File "c:\Users\username\Documents\Coding\azure-pipeline\.venv\lib\site-packages\sqlalchemy\engine\default.py", line 1838, in _setup_result_proxy
    strategy = _cursor.BufferedRowCursorFetchStrategy(
  File "c:\Users\username\Documents\Coding\azure-pipeline\.venv\lib\site-packages\sqlalchemy\engine\cursor.py", line 1188, in __init__
    self._rowbuffer = collections.deque(dbapi_cursor.fetchmany(1))
pyodbc.ProgrammingError: No results.  Previous SQL was not a query.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\username\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 196, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "C:\Users\username\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 86, in _run_code
    exec(code, run_globals)
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\__main__.py", line 71, in <module>
    cli.main()
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy/..\debugpy\server\cli.py", line 501, in main
    run()
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy/..\debugpy\server\cli.py", line 351, in run_file
    runpy.run_path(target, run_name="__main__")
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\_vendored\pydevd\_pydevd_bundle\pydevd_runpy.py", line 310, in run_path
    return _run_module_code(code, init_globals, run_name, pkg_name=pkg_name, script_name=fname)
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\_vendored\pydevd\_pydevd_bundle\pydevd_runpy.py", line 127, in _run_module_code
    _run_code(code, mod_globals, init_globals, mod_name, mod_spec, pkg_name, script_name)
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\_vendored\pydevd\_pydevd_bundle\pydevd_runpy.py", line 118, in _run_code
    exec(code, run_globals)
  File "C:\Users\username\Documents\Coding\azure-pipeline\mygrcreport.py", line 94, in <module>
    mygrc_3ps_inherent_risk, primary_key = mygrcreport_db.dynamic_table(schema, "mygrc_3ps_inherent_risk", dbEngine)
  File "\mygrcreport_db.py", line 81, in dynamic_table
    dynamic_class.__table__.drop(dbEngine, checkfirst=True)
  File "\.venv\lib\site-packages\sqlalchemy\sql\schema.py", line 1301, in drop
    bind._run_ddl_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 3249, in _run_ddl_visitor
    conn._run_ddl_visitor(visitorcallable, element, **kwargs)
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 2456, in _run_ddl_visitor
    visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
  File "\.venv\lib\site-packages\sqlalchemy\sql\visitors.py", line 664, in traverse_single
    return meth(obj, **kw)
  File "\.venv\lib\site-packages\sqlalchemy\sql\ddl.py", line 1203, in visit_table
    DropTable(table)._invoke_with(self.connection)
  File "\.venv\lib\site-packages\sqlalchemy\sql\ddl.py", line 314, in _invoke_with
    return bind.execute(self)
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1416, in execute
    return meth(
  File "\.venv\lib\site-packages\sqlalchemy\sql\ddl.py", line 180, in _execute_on_connection
    return connection._execute_ddl(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1527, in _execute_ddl
    ret = self._execute_context(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1843, in _execute_context
    return self._exec_single_context(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1983, in _exec_single_context
    self._handle_dbapi_exception(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 2352, in _handle_dbapi_exception
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1980, in _exec_single_context
    result = context._setup_result_proxy()
  File "\.venv\lib\site-packages\sqlalchemy\engine\default.py", line 1838, in _setup_result_proxy
    strategy = _cursor.BufferedRowCursorFetchStrategy(
  File "\.venv\lib\site-packages\sqlalchemy\engine\cursor.py", line 1188, in __init__
    self._rowbuffer = collections.deque(dbapi_cursor.fetchmany(1))
sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) No results.  Previous SQL was not a query.
[SQL:
DROP TABLE mygrc.mygrc_3ps_inherent_risk]

I set the code to:

logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
log.info(f"Dropping Table: mygrc.{table_name}")
with dbEngine.connect().execution_options(isolation_level='AUTOCOMMIT') as connection:
    connection.execute(text(f'DROP TABLE IF EXISTS mygrc.{table_name}'))

It still didn't work, it looks like it's still trying to do a rollback even with auto commit enabled, no errors no exceptions thrown and notice the raw sql is blank, same thing happens with the create and my user can run the command without issue (and the table exists), if I issue the create statement manually to execute it does the same:

025-03-28 10:07:56,802 - INFO - Dropping Table: mygrc.mygrc_3ps_inherent_risk
2025-03-28 10:07:58,279 INFO sqlalchemy.engine.Engine SELECT CAST(SERVERPROPERTY('ProductVersion') AS VARCHAR)
2025-03-28 10:07:58,279 - INFO - SELECT CAST(SERVERPROPERTY('ProductVersion') AS VARCHAR)
2025-03-28 10:07:58,279 INFO sqlalchemy.engine.Engine [raw sql] ()
2025-03-28 10:07:58,279 - INFO - [raw sql] ()
2025-03-28 10:07:58,416 INFO sqlalchemy.engine.Engine SELECT schema_name()
2025-03-28 10:07:58,416 - INFO - SELECT schema_name()
2025-03-28 10:07:58,416 INFO sqlalchemy.engine.Engine [generated in 0.00073s] ()
2025-03-28 10:07:58,416 - INFO - [generated in 0.00073s] ()
2025-03-28 10:07:58,822 INFO sqlalchemy.engine.Engine SELECT CAST('test max support' AS NVARCHAR(max))
2025-03-28 10:07:58,822 - INFO - SELECT CAST('test max support' AS NVARCHAR(max))
2025-03-28 10:07:58,823 INFO sqlalchemy.engine.Engine [generated in 0.00052s] ()
2025-03-28 10:07:58,823 - INFO - [generated in 0.00052s] ()
2025-03-28 10:07:58,958 INFO sqlalchemy.engine.Engine SELECT 1 FROM fn_listextendedproperty(default, default, default, default, default, default, default)
2025-03-28 10:07:58,958 - INFO - SELECT 1 FROM fn_listextendedproperty(default, default, default, default, default, default, default)  
2025-03-28 10:07:58,959 INFO sqlalchemy.engine.Engine [generated in 0.00053s] ()
2025-03-28 10:07:58,959 - INFO - [generated in 0.00053s] ()
2025-03-28 10:08:35,716 INFO sqlalchemy.engine.Engine BEGIN (implicit; DBAPI should not BEGIN due to autocommit mode)
2025-03-28 10:08:35,716 - INFO - BEGIN (implicit; DBAPI should not BEGIN due to autocommit mode)
2025-03-28 10:08:35,716 INFO sqlalchemy.engine.Engine DROP TABLE IF EXISTS mygrc.mygrc_3ps_inherent_risk
2025-03-28 10:08:35,716 - INFO - DROP TABLE IF EXISTS mygrc.mygrc_3ps_inherent_risk
2025-03-28 10:08:35,717 INFO sqlalchemy.engine.Engine [generated in 0.00132s] ()
2025-03-28 10:08:35,717 - INFO - [generated in 0.00132s] ()
2025-03-28 10:08:43,022 INFO sqlalchemy.engine.Engine ROLLBACK using DBAPI connection.rollback(), DBAPI should ignore due to autocommit mode
2025-03-28 10:08:43,022 - INFO - ROLLBACK using DBAPI connection.rollback(), DBAPI should ignore due to autocommit mode

I've created a SQLAlchemy (2.0.39), Python (3.10.11) definition on the fly:

    dynamic_class = type(table_name, (Base,), attributes)

This is working, now I want to drop any existing table and recreate it using the current definition (and where the problems kick in)

    log.info(f"Dropping Table: mygrc.{table_name}")
    dynamic_class.__table__.drop(dbEngine, checkfirst=True)

    log.info(f"Creating Table: mygrc.{table_name}")
    Base.metadata.create_all(dbEngine, tables=[dynamic_class.__table__])

Both the drop and create statements give me errors:

No results. Previous SQL was not a query.

If I try to pass the query direct to execute, it doesn't error but it doesn't execute the command either (even with a commit)

    with dbEngine.connect() as connection:
        create_table_statement = str(dynamic_class.__table__pile(dbEngine))
        connection.execute(text(create_table_statement))

create_table_statement gives me a valid create statement.

Any ideas?

Track trace:

Traceback (most recent call last):
  File "c:\Users\username\Documents\Coding\azure-pipeline\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1980, in _exec_single_context
    result = context._setup_result_proxy()
  File "c:\Users\username\Documents\Coding\azure-pipeline\.venv\lib\site-packages\sqlalchemy\engine\default.py", line 1838, in _setup_result_proxy
    strategy = _cursor.BufferedRowCursorFetchStrategy(
  File "c:\Users\username\Documents\Coding\azure-pipeline\.venv\lib\site-packages\sqlalchemy\engine\cursor.py", line 1188, in __init__
    self._rowbuffer = collections.deque(dbapi_cursor.fetchmany(1))
pyodbc.ProgrammingError: No results.  Previous SQL was not a query.

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\username\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 196, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "C:\Users\username\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 86, in _run_code
    exec(code, run_globals)
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\__main__.py", line 71, in <module>
    cli.main()
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy/..\debugpy\server\cli.py", line 501, in main
    run()
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy/..\debugpy\server\cli.py", line 351, in run_file
    runpy.run_path(target, run_name="__main__")
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\_vendored\pydevd\_pydevd_bundle\pydevd_runpy.py", line 310, in run_path
    return _run_module_code(code, init_globals, run_name, pkg_name=pkg_name, script_name=fname)
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\_vendored\pydevd\_pydevd_bundle\pydevd_runpy.py", line 127, in _run_module_code
    _run_code(code, mod_globals, init_globals, mod_name, mod_spec, pkg_name, script_name)
  File "c:\Users\username\.vscode\extensions\ms-python.debugpy-2025.4.1-win32-x64\bundled\libs\debugpy\_vendored\pydevd\_pydevd_bundle\pydevd_runpy.py", line 118, in _run_code
    exec(code, run_globals)
  File "C:\Users\username\Documents\Coding\azure-pipeline\mygrcreport.py", line 94, in <module>
    mygrc_3ps_inherent_risk, primary_key = mygrcreport_db.dynamic_table(schema, "mygrc_3ps_inherent_risk", dbEngine)
  File "\mygrcreport_db.py", line 81, in dynamic_table
    dynamic_class.__table__.drop(dbEngine, checkfirst=True)
  File "\.venv\lib\site-packages\sqlalchemy\sql\schema.py", line 1301, in drop
    bind._run_ddl_visitor(ddl.SchemaDropper, self, checkfirst=checkfirst)
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 3249, in _run_ddl_visitor
    conn._run_ddl_visitor(visitorcallable, element, **kwargs)
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 2456, in _run_ddl_visitor
    visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
  File "\.venv\lib\site-packages\sqlalchemy\sql\visitors.py", line 664, in traverse_single
    return meth(obj, **kw)
  File "\.venv\lib\site-packages\sqlalchemy\sql\ddl.py", line 1203, in visit_table
    DropTable(table)._invoke_with(self.connection)
  File "\.venv\lib\site-packages\sqlalchemy\sql\ddl.py", line 314, in _invoke_with
    return bind.execute(self)
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1416, in execute
    return meth(
  File "\.venv\lib\site-packages\sqlalchemy\sql\ddl.py", line 180, in _execute_on_connection
    return connection._execute_ddl(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1527, in _execute_ddl
    ret = self._execute_context(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1843, in _execute_context
    return self._exec_single_context(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1983, in _exec_single_context
    self._handle_dbapi_exception(
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 2352, in _handle_dbapi_exception
    raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
  File "\.venv\lib\site-packages\sqlalchemy\engine\base.py", line 1980, in _exec_single_context
    result = context._setup_result_proxy()
  File "\.venv\lib\site-packages\sqlalchemy\engine\default.py", line 1838, in _setup_result_proxy
    strategy = _cursor.BufferedRowCursorFetchStrategy(
  File "\.venv\lib\site-packages\sqlalchemy\engine\cursor.py", line 1188, in __init__
    self._rowbuffer = collections.deque(dbapi_cursor.fetchmany(1))
sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) No results.  Previous SQL was not a query.
[SQL:
DROP TABLE mygrc.mygrc_3ps_inherent_risk]

I set the code to:

logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
log.info(f"Dropping Table: mygrc.{table_name}")
with dbEngine.connect().execution_options(isolation_level='AUTOCOMMIT') as connection:
    connection.execute(text(f'DROP TABLE IF EXISTS mygrc.{table_name}'))

It still didn't work, it looks like it's still trying to do a rollback even with auto commit enabled, no errors no exceptions thrown and notice the raw sql is blank, same thing happens with the create and my user can run the command without issue (and the table exists), if I issue the create statement manually to execute it does the same:

025-03-28 10:07:56,802 - INFO - Dropping Table: mygrc.mygrc_3ps_inherent_risk
2025-03-28 10:07:58,279 INFO sqlalchemy.engine.Engine SELECT CAST(SERVERPROPERTY('ProductVersion') AS VARCHAR)
2025-03-28 10:07:58,279 - INFO - SELECT CAST(SERVERPROPERTY('ProductVersion') AS VARCHAR)
2025-03-28 10:07:58,279 INFO sqlalchemy.engine.Engine [raw sql] ()
2025-03-28 10:07:58,279 - INFO - [raw sql] ()
2025-03-28 10:07:58,416 INFO sqlalchemy.engine.Engine SELECT schema_name()
2025-03-28 10:07:58,416 - INFO - SELECT schema_name()
2025-03-28 10:07:58,416 INFO sqlalchemy.engine.Engine [generated in 0.00073s] ()
2025-03-28 10:07:58,416 - INFO - [generated in 0.00073s] ()
2025-03-28 10:07:58,822 INFO sqlalchemy.engine.Engine SELECT CAST('test max support' AS NVARCHAR(max))
2025-03-28 10:07:58,822 - INFO - SELECT CAST('test max support' AS NVARCHAR(max))
2025-03-28 10:07:58,823 INFO sqlalchemy.engine.Engine [generated in 0.00052s] ()
2025-03-28 10:07:58,823 - INFO - [generated in 0.00052s] ()
2025-03-28 10:07:58,958 INFO sqlalchemy.engine.Engine SELECT 1 FROM fn_listextendedproperty(default, default, default, default, default, default, default)
2025-03-28 10:07:58,958 - INFO - SELECT 1 FROM fn_listextendedproperty(default, default, default, default, default, default, default)  
2025-03-28 10:07:58,959 INFO sqlalchemy.engine.Engine [generated in 0.00053s] ()
2025-03-28 10:07:58,959 - INFO - [generated in 0.00053s] ()
2025-03-28 10:08:35,716 INFO sqlalchemy.engine.Engine BEGIN (implicit; DBAPI should not BEGIN due to autocommit mode)
2025-03-28 10:08:35,716 - INFO - BEGIN (implicit; DBAPI should not BEGIN due to autocommit mode)
2025-03-28 10:08:35,716 INFO sqlalchemy.engine.Engine DROP TABLE IF EXISTS mygrc.mygrc_3ps_inherent_risk
2025-03-28 10:08:35,716 - INFO - DROP TABLE IF EXISTS mygrc.mygrc_3ps_inherent_risk
2025-03-28 10:08:35,717 INFO sqlalchemy.engine.Engine [generated in 0.00132s] ()
2025-03-28 10:08:35,717 - INFO - [generated in 0.00132s] ()
2025-03-28 10:08:43,022 INFO sqlalchemy.engine.Engine ROLLBACK using DBAPI connection.rollback(), DBAPI should ignore due to autocommit mode
2025-03-28 10:08:43,022 - INFO - ROLLBACK using DBAPI connection.rollback(), DBAPI should ignore due to autocommit mode

• python
• sql-server
• sqlalchemy

Share Improve this question Follow edited Mar 28 at 10:14 trevrobwhite asked Mar 27 at 17:04 trevrobwhitetrevrobwhite 46311 gold badge88 silver badges2222 bronze badges 6

• This seems to work for me, modulo I don't have access to MSSQL so I used SQLite. Can you share a traceback? – snakecharmerb Commented Mar 27 at 17:37
• Added above, I couldn't include in the comments, thank-you – trevrobwhite Commented Mar 27 at 20:48
• I suspect the actual error is that table may not exist yet, then you can't drop it. If you look at the stack trace, the error comes after handling a dbapi exception. Could be wrong though! – siggemannen Commented Mar 27 at 21:14
• The table exists but then plan B when I issue connection.execute why does it just step over it steps this meh code. It’s really strange. – trevrobwhite Commented Mar 27 at 21:32
• Can you enable logging and share the SQLA logs prior to the error (please edit them into the question body). I would like to know more about the "Previous SQL was not a query." Are any of the answers / situations here applicable? – snakecharmerb Commented Mar 28 at 6:42

 |  Show 1 more comment

1 Answer 1

Sorted by: Reset to default 0

I finally managed to get it to work, it was a lot more hassle than I was expecting so I'm open to improvements, sharing to save someone else from the pain, I increased the logging and could see it was always rolling back:

logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)

For some reason it didn't like me issuing the drop and create statements on SQLAlchemy engine, this caused a rollback every time but without error or exception which made me think it wasn't doing anything until I turned on the logging above.

Changing engine to a session it worked, the create and drop functions should have also worked but they always gave the error "No results. Previous SQL was not a query."

I share my full function below in case it's useful for anyone, pick out any useful parts.

def dynamic_table(schema, table_name, dbEngine=False, dbSchema="mygrc"):
    """
    Dynamically creates a SQLAlchemy ORM class based on the provided schema and optionally manages the database table.
                       Each field should have a "name" key for the column name and an optional "typename" key for the data type.
        dbEngine (Engine, optional): A SQLAlchemy database engine. If provided, the function will manage the table in the database.
        dbSchema (str, optional): The schema name for the table. Defaults to "mygrc".
        tuple: A tuple containing:
            - type: The dynamically created SQLAlchemy ORM class.
            - str: The name of the primary key column.
    Raises:
        KeyError: If a required key is missing in the schema dictionary.
        Exception: If there are issues with database operations when `dbEngine` is provided.
    Notes:
        - The function maps common data types (e.g., "text", "int", "boolean") to SQLAlchemy column types.
        - The first field in the schema is treated as the primary key.
        - If `dbEngine` is provided, the function will drop the existing table (if any) and create a new one.
    """
    from sqlalchemy.engine import Engine  # Add this import at the top if not already present
    if dbEngine and not isinstance(dbEngine, Engine):
        raise TypeError("dbEngine must be an instance of sqlalchemy.engine.Engine")

    type_mapping = {
        "text": NVARCHAR(None),
        "int": Integer,
        "boolean": Boolean,
        "datetime": DateTime,
        "date": Date,
        "float": Float,
        "uniqueidentifier": UNIQUEIDENTIFIER,
    }

    # Define attributes for the dynamic class
    attributes = {
        "__tablename__": table_name,
        "__table_args__": {"schema": dbSchema},
    }

    count = 0

    for field_id, field_info in schema.items():
        count += 1
        field_name = field_info["name"]
        field_alias = field_info["name"].lower().replace(' ', '_')
        field_type = field_info.get("typename", "text").lower()
        column_type = type_mapping.get(field_type, String)  # Default to String if type is unknown
        
        # First column is primary
        if count == 1:
            attributes[field_alias] = mapped_column(field_name, column_type, primary_key=True, autoincrement=False)
            primary_key = field_alias
        else:
            attributes[field_alias] = mapped_column(field_name, column_type)
        

    # Dynamically create the class
    dynamic_class = type(table_name, (Base,), attributes)

    if dbEngine:
        dbSession = Session(dbEngine)
        logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
        log.info(f"Dropping Table: mygrc.{table_name}")
        dbSession.execute(text(f'DROP TABLE IF EXISTS {dbSchema}.{table_name}'))
        dbSessionmit()

        log.info(f"Creating Table: {dbSchema}.{table_name}")
        create_table_statement = str(CreateTable(dynamic_class.__table__)pile(dbEngine))
        dbSession.execute(text(create_table_statement))
        dbSessionmit()

        dbSession.close()

    return dynamic_class, primary_key

本文标签： pythonSQLAlchemy mssql DropCreate Table on the FlyStack Overflow