[Python] Mirror SQL Server to DuckDB local database with SQLAlchemy: constraint not implemented

I'm trying to mirror a remote SQL Server database into a local DuckDB database.

Here is my code:

# Connect to remote db
username = "USERNAME"
hostip = "IPADD"
port = "PORT"
dbname = "DBNAME"
user_domain = "USER_DOMAIN"
# See: https://stackoverflow.com/a/58858572/1719931
eng_str = rf"mssql+pymssql://{user_domain}\{username}:{password}@{hostip}/{dbname}"
# Establish connection
engine_remote = create_engine(eng_str, echo=False)
# Test that connection to remote database works
with Session(engine_remote) as session:
print(session.execute(text("SELECT 'hello world'")).fetchall())

# Local db
dbfp = Path("localdb.duckdb")
engine_local = create_engine(f"duckdb:///{dbfp}", echo=False)

# To reflect the metadata of a database with ORM we need the "Automap" extension
# Automap is an extension to the sqlalchemy.ext.declarative system which automatically generates
# mapped classes and relationships from a database schema,
# typically though not necessarily one which is reflected.
# See: https://docs.sqlalchemy.org/en/20/orm/extensions/automap.html
Base = automap_base()

# See ORM documentation on intercepting column definitions: https://docs.sqlalchemy.org/en/20/orm/extensions/automap.html#intercepting-column-definitions
@event.listens_for(Base.metadata, "column_reflect")
def genericize_datatypes(inspector, tablename, column_dict):
# Convert dialect specific column types to SQLAlchemy agnostic types
# See: https://stackoverflow.com/questions/79496414/convert-tinyint-to-int-when-mirroring-microsoft-sql-server-to-local-sqlite-with
# See Core documentation on reflecting with database-agnostic types: https://docs.sqlalchemy.org/en/20/core/reflection.html#reflecting-with-database-agnostic-types
old_type = column_dict['type']
column_dict["type"] = column_dict["type"].as_generic()
# We have to remove collation when mirroring a Microsoft SQL server into SQLite
# See: https://stackoverflow.com/a/59328211/1719931
if getattr(column_dict["type"], "collation", None) is not None:
column_dict["type"].collation = None
# Print debug info
if not isinstance(column_dict['type'], type(old_type)):
print(f"Genericizing `{column_dict['name']}` of type `{str(old_type)}` into `{column_dict['type']}`")

# Load Base with remote DB metadata
Base.prepare(autoload_with=engine_remote)

Base.metadata.create_all(engine_local)


But I'm getting this error:

---------------------------------------------------------------------------
NotImplementedException Traceback (most recent call last)
File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:1964, in Connection._exec_single_context(self, dialect, context, statement, parameters)
1963 if not evt_handled:
-> 1964 self.dialect.do_execute(
1965 cursor, str_statement, effective_parameters, context
1966 )
1968 if self._has_events or self.engine._has_events:

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\default.py:942, in DefaultDialect.do_execute(self, cursor, statement, parameters, context)
941 def do_execute(self, cursor, statement, parameters, context=None):
--> 942 cursor.execute(statement, parameters)

File PYTHONPATH\Lib\site-packages\duckdb_engine\__init__.py:150, in CursorWrapper.execute(self, statement, parameters, context)
149 else:
--> 150 self.__c.execute(statement, parameters)
151 except RuntimeError as e:

NotImplementedException: Not implemented Error: Constraint not implemented!

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

NotSupportedError Traceback (most recent call last)
Cell In[25], line 1
----> 1 Base.metadata.create_all(engine_local)

File PYTHONPATH\Lib\site-packages\sqlalchemy\sql\schema.py:5907, in MetaData.create_all(self, bind, tables, checkfirst)
5883 def create_all(
5884 self,
5885 bind: _CreateDropBind,
5886 tables: Optional[_typing_Sequence[Table]] = None,
5887 checkfirst: bool = True,
5888 ) -> None:
5889 """Create all tables stored in this metadata.
5890
5891 Conditional by default, will not attempt to recreate tables already
(...) 5905
5906 """
-> 5907 bind._run_ddl_visitor(
5908 ddl.SchemaGenerator, self, checkfirst=checkfirst, tables=tables
5909 )

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:3249, in Engine._run_ddl_visitor(self, visitorcallable, element, **kwargs)
3242 def _run_ddl_visitor(
3243 self,
3244 visitorcallable: Type[Union[SchemaGenerator, SchemaDropper]],
3245 element: SchemaItem,
3246 **kwargs: Any,
3247 ) -> None:
3248 with self.begin() as conn:
-> 3249 conn._run_ddl_visitor(visitorcallable, element, **kwargs)

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:2456, in Connection._run_ddl_visitor(self, visitorcallable, element, **kwargs)
2444 def _run_ddl_visitor(
2445 self,
2446 visitorcallable: Type[Union[SchemaGenerator, SchemaDropper]],
2447 element: SchemaItem,
2448 **kwargs: Any,
2449 ) -> None:
2450 """run a DDL visitor.
2451
2452 This method is only here so that the MockConnection can change the
2453 options given to the visitor so that "checkfirst" is skipped.
2454
2455 """
-> 2456 visitorcallable(self.dialect, self, **kwargs).traverse_single(element)

File PYTHONPATH\Lib\site-packages\sqlalchemy\sql\visitors.py:664, in ExternalTraversal.traverse_single(self, obj, **kw)
662 meth = getattr(v, "visit_%s" % obj.__visit_name__, None)
663 if meth:
--> 664 return meth(obj, **kw)

File PYTHONPATH\Lib\site-packages\sqlalchemy\sql\ddl.py:978, in SchemaGenerator.visit_metadata(self, metadata)
976 for table, fkcs in collection:
977 if table is not None:
--> 978 self.traverse_single(
979 table,
980 create_ok=True,
981 include_foreign_key_constraints=fkcs,
982 _is_metadata_operation=True,
983 )
984 else:
985 for fkc in fkcs:

File PYTHONPATH\Lib\site-packages\sqlalchemy\sql\visitors.py:664, in ExternalTraversal.traverse_single(self, obj, **kw)
662 meth = getattr(v, "visit_%s" % obj.__visit_name__, None)
663 if meth:
--> 664 return meth(obj, **kw)

File PYTHONPATH\Lib\site-packages\sqlalchemy\sql\ddl.py:1016, in SchemaGenerator.visit_table(self, table, create_ok, include_foreign_key_constraints, _is_metadata_operation)
1007 if not self.dialect.supports_alter:
1008 # e.g., don't omit any foreign key constraints
1009 include_foreign_key_constraints = None
1011 CreateTable(
1012 table,
1013 include_foreign_key_constraints=(
1014 include_foreign_key_constraints
1015 ),
-> 1016 )._invoke_with(self.connection)
1018 if hasattr(table, "indexes"):
1019 for index in table.indexes:

File PYTHONPATH\Lib\site-packages\sqlalchemy\sql\ddl.py:314, in ExecutableDDLElement._invoke_with(self, bind)
312 def _invoke_with(self, bind):
313 if self._should_execute(self.target, bind):
--> 314 return bind.execute(self)

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:1416, in Connection.execute(self, statement, parameters, execution_options)
1414 raise exc.ObjectNotExecutableError(statement) from err
1415 else:
-> 1416 return meth(
1417 self,
1418 distilled_parameters,
1419 execution_options or NO_OPTIONS,
1420 )

File PYTHONPATH\Lib\site-packages\sqlalchemy\sql\ddl.py:180, in ExecutableDDLElement._execute_on_connection(self, connection, distilled_params, execution_options)
177 def _execute_on_connection(
178 self, connection, distilled_params, execution_options
179 ):
--> 180 return connection._execute_ddl(
181 self, distilled_params, execution_options
182 )

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:1527, in Connection._execute_ddl(self, ddl, distilled_parameters, execution_options)
1522 dialect = self.dialect
1524 compiled = ddl.compile(
1525 dialect=dialect, schema_translate_map=schema_translate_map
1526 )
-> 1527 ret = self._execute_context(
1528 dialect,
1529 dialect.execution_ctx_cls._init_ddl,
1530 compiled,
1531 None,
1532 exec_opts,
1533 compiled,
1534 )
1535 if self._has_events or self.engine._has_events:
1536 self.dispatch.after_execute(
1537 self,
1538 ddl,
(...) 1542 ret,
1543 )

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:1843, in Connection._execute_context(self, dialect, constructor, statement, parameters, execution_options, *args, **kw)
1841 return self._exec_insertmany_context(dialect, context)
1842 else:
-> 1843 return self._exec_single_context(
1844 dialect, context, statement, parameters
1845 )

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:1983, in Connection._exec_single_context(self, dialect, context, statement, parameters)
1980 result = context._setup_result_proxy()
1982 except BaseException as e:
-> 1983 self._handle_dbapi_exception(
1984 e, str_statement, effective_parameters, cursor, context
1985 )
1987 return result

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:2352, in Connection._handle_dbapi_exception(self, e, statement, parameters, cursor, context, is_sub_exec)
2350 elif should_wrap:
2351 assert sqlalchemy_exception is not None
-> 2352 raise sqlalchemy_exception.with_traceback(exc_info[2]) from e
2353 else:
2354 assert exc_info[1] is not None

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\base.py:1964, in Connection._exec_single_context(self, dialect, context, statement, parameters)
1962 break
1963 if not evt_handled:
-> 1964 self.dialect.do_execute(
1965 cursor, str_statement, effective_parameters, context
1966 )
1968 if self._has_events or self.engine._has_events:
1969 self.dispatch.after_cursor_execute(
1970 self,
1971 cursor,
(...) 1975 context.executemany,
1976 )

File PYTHONPATH\Lib\site-packages\sqlalchemy\engine\default.py:942, in DefaultDialect.do_execute(self, cursor, statement, parameters, context)
941 def do_execute(self, cursor, statement, parameters, context=None):
--> 942 cursor.execute(statement, parameters)

File PYTHONPATH\Lib\site-packages\duckdb_engine\__init__.py:150, in CursorWrapper.execute(self, statement, parameters, context)
148 self.__c.execute(statement)
149 else:
--> 150 self.__c.execute(statement, parameters)
151 except RuntimeError as e:
152 if e.args[0].startswith("Not implemented Error"):

NotSupportedError: (duckdb.duckdb.NotImplementedException) Not implemented Error: Constraint not implemented!
[SQL:
CREATE TABLE sysdiagrams (
name VARCHAR(128) NOT NULL,
principal_id INTEGER NOT NULL,
diagram_id INTEGER GENERATED BY DEFAULT AS IDENTITY (INCREMENT BY 1 START WITH 1),
version INTEGER,
definition BYTEA,
CONSTRAINT "PK__sysdiagr__C2B05B613750874A" PRIMARY KEY (diagram_id)
)

]
(Background on this error at: https://sqlalche.me/e/20/tw8g)


It looks like the primary key constraint is not implemented in DuckDB?

I used to use SQLite and it worked, but now I'm trying to switch to DuckDB.

Continue reading...