0

I'm tyring to mirror the schema of a remote Microsft SQL Server 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?

Improve this question

2 Answers 2

Reset to default
0

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)
Improve this answer
Sign up to request clarification or add additional context in comments.

Comments

0

Startgin from SQLAlchemy 2.0.42, reflection from Microsoft SQL Server has been reworked "to be based on the sys.columns table rather than information_schema.columns view."

It now works natively and you don't need neither to remove collation nor to convert dialect specific column types to SQLAlchemy agnostic types.

Just do:

engine_remote = create_engine(rf"mssql+pymssql://{user_domain}\{username}:{password}@{hostip}:{port}/{dbname}", echo=False)
engine_local = create_engine(f"sqlite:///{local_db_fp}", echo=False)
Base = automap_base()
Base.prepare(autoload_with=engine_remote)
Base.metadata.create_all(engine_local)
Improve this answer

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.