I'm tyring to mirror the schema of a remote Microsft SQL database into a local SQLite database.
I'm using ORM and the automap extension to mirror the remote database schema.
My code looks like this:
from sqlalchemy import create_engine, select, MetaData, text, Integer
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session
eng_str = rf"mssql+pymssql://{user_domain}\{username}:{password}@{hostip}/{dbname}"
engine_remote = create_engine(eng_str, echo=False)
dbfp = Path("localdb.sqlite3")
engine_local = create_engine(f"sqlite:///{dbfp}", echo=False)
Base = automap_base()
Base.prepare(autoload_with=engine_remote)
# We have to remove collation when mirroring a Microsoft SQL server into SQLite
# See: https://stackoverflow.com/a/59328211/1719931
for table in Base.metadata.sorted_tables:
for col in table.c:
if getattr(col.type, 'collation', None) is not None:
col.type.collation = None
Base.metadata.create_all(engine_local)
which generates the following error:
---------------------------------------------------------------------------
AttributeError Traceback (most recent call last)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\visitors.py:137, in Visitable._generate_compiler_dispatch.<locals>._compiler_dispatch(self, visitor, **kw)
136 try:
--> 137 meth = getter(visitor)
138 except AttributeError as err:
AttributeError: 'SQLiteTypeCompiler' object has no attribute 'visit_TINYINT'
The above exception was the direct cause of the following exception:
UnsupportedCompilationError Traceback (most recent call last)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:6656, in DDLCompiler.visit_create_table(self, create, **kw)
6655 try:
-> 6656 processed = self.process(
6657 create_column, first_pk=column.primary_key and not first_pk
6658 )
6659 if processed is not None:
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:915, in Compiled.process(self, obj, **kwargs)
914 def process(self, obj: Visitable, **kwargs: Any) -> str:
--> 915 return obj._compiler_dispatch(self, **kwargs)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\visitors.py:141, in Visitable._generate_compiler_dispatch.<locals>._compiler_dispatch(self, visitor, **kw)
140 else:
--> 141 return meth(self, **kw)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:6687, in DDLCompiler.visit_create_column(self, create, first_pk, **kw)
6685 return None
-> 6687 text = self.get_column_specification(column, first_pk=first_pk)
6688 const = " ".join(
6689 self.process(constraint) for constraint in column.constraints
6690 )
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\dialects\sqlite\base.py:1591, in SQLiteDDLCompiler.get_column_specification(self, column, **kwargs)
1590 def get_column_specification(self, column, **kwargs):
-> 1591 coltype = self.dialect.type_compiler_instance.process(
1592 column.type, type_expression=column
1593 )
1594 colspec = self.preparer.format_column(column) + " " + coltype
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:960, in TypeCompiler.process(self, type_, **kw)
959 type_ = type_._variant_mapping[self.dialect.name]
--> 960 return type_._compiler_dispatch(self, **kw)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\visitors.py:139, in Visitable._generate_compiler_dispatch.<locals>._compiler_dispatch(self, visitor, **kw)
138 except AttributeError as err:
--> 139 return visitor.visit_unsupported_compilation(self, err, **kw) # type: ignore # noqa: E501
140 else:
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:965, in TypeCompiler.visit_unsupported_compilation(self, element, err, **kw)
962 def visit_unsupported_compilation(
963 self, element: Any, err: Exception, **kw: Any
964 ) -> NoReturn:
--> 965 raise exc.UnsupportedCompilationError(self, element) from err
UnsupportedCompilationError: Compiler <sqlalchemy.dialects.sqlite.base.SQLiteTypeCompiler object at 0x000002D8741DB770> can't render element of type TINYINT (Background on this error at: https://sqlalche.me/e/20/l7de)
The above exception was the direct cause of the following exception:
CompileError Traceback (most recent call last)
Cell In[20], line 1
----> 1 Base.metadata.create_all(engine_local)
File HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\ddl.py:928, in SchemaGenerator.visit_metadata(self, metadata)
926 for table, fkcs in collection:
927 if table is not None:
--> 928 self.traverse_single(
929 table,
930 create_ok=True,
931 include_foreign_key_constraints=fkcs,
932 _is_metadata_operation=True,
933 )
934 else:
935 for fkc in fkcs:
File HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\ddl.py:966, in SchemaGenerator.visit_table(self, table, create_ok, include_foreign_key_constraints, _is_metadata_operation)
957 if not self.dialect.supports_alter:
958 # e.g., don't omit any foreign key constraints
959 include_foreign_key_constraints = None
961 CreateTable(
962 table,
963 include_foreign_key_constraints=(
964 include_foreign_key_constraints
965 ),
--> 966 )._invoke_with(self.connection)
968 if hasattr(table, "indexes"):
969 for index in table.indexes:
File HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\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 HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\engine\base.py:1524, in Connection._execute_ddl(self, ddl, distilled_parameters, execution_options)
1520 schema_translate_map = exec_opts.get("schema_translate_map", None)
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,
(...)
1533 compiled,
1534 )
1535 if self._has_events or self.engine._has_events:
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\elements.py:309, in CompilerElement.compile(self, bind, dialect, **kw)
304 url = util.preloaded.engine_url
305 dialect = url.URL.create(
306 self.stringify_dialect
307 ).get_dialect()()
--> 309 return self._compiler(dialect, **kw)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\ddl.py:69, in BaseDDLElement._compiler(self, dialect, **kw)
65 def _compiler(self, dialect, **kw):
66 """Return a compiler appropriate for this ClauseElement, given a
67 Dialect."""
---> 69 return dialect.ddl_compiler(dialect, self, **kw)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:870, in Compiled.__init__(self, dialect, statement, schema_translate_map, render_schema_translate, compile_kwargs)
868 assert isinstance(statement, Executable)
869 self.execution_options = statement._execution_options
--> 870 self.string = self.process(self.statement, **compile_kwargs)
872 if render_schema_translate:
873 self.string = self.preparer._render_schema_translates(
874 self.string, schema_translate_map
875 )
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:915, in Compiled.process(self, obj, **kwargs)
914 def process(self, obj: Visitable, **kwargs: Any) -> str:
--> 915 return obj._compiler_dispatch(self, **kwargs)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\visitors.py:141, in Visitable._generate_compiler_dispatch.<locals>._compiler_dispatch(self, visitor, **kw)
139 return visitor.visit_unsupported_compilation(self, err, **kw) # type: ignore # noqa: E501
140 else:
--> 141 return meth(self, **kw)
File HOME\AppData\Local\Programs\Python\Python313\Lib\site-packages\sqlalchemy\sql\compiler.py:6666, in DDLCompiler.visit_create_table(self, create, **kw)
6664 first_pk = True
6665 except exc.CompileError as ce:
-> 6666 raise exc.CompileError(
6667 "(in table '%s', column '%s'): %s"
6668 % (table.description, column.name, ce.args[0])
6669 ) from ce
6671 const = self.create_table_constraints(
6672 table,
6673 _include_foreign_key_constraints=create.include_foreign_key_constraints, # noqa
6674 )
6675 if const:
CompileError: (in table 'TABLE_NAME', column 'COLUMN_NAME'): Compiler <sqlalchemy.dialects.sqlite.base.SQLiteTypeCompiler object at 0x000002D8741DB770> can't render element of type TINYINT
I would like to fix the problem by running the following code before creating the metadata:
for table in Base.metadata.sorted_tables:
for col in table.c:
if "TINYINT" in str(col.type):
print(f"Column {col} is of type {col.type}. Changing to Integer")
col.type = Integer
Is this the right/best way to solve it?
This seems to work:
# 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 dfinitions: 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 Core doucumentation 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']}`")
# reflect the tables
Base.prepare(autoload_with=engine_remote)