Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

SQLAlchemy - Merge/extend existing metadata?

Is it possible to merge a _BoundDeclarativeMeta instance with another?

For example:

from sqlalchemy.ext.declarative import declarative_base

Base1 = declarative_base()
Base2 = declarative_base()

# Something like this?
CombinedBase = Base1.merge(Base2)

In reality I would be subclassing orm models from each of them before merging. The problem is that I have an independent package that needs certain tables to work. Yet it is going to be used by another independent package.

like image 426
Tim Martin Avatar asked Apr 29 '15 21:04

Tim Martin


2 Answers

For my use case this worked:

from sqlalchemy import MetaData

combined_meta_data = MetaData()

for declarative_base in [Base1, Base2]:
    for (table_name, table) in declarative_base.metadata.tables.items():
        combined_meta_data._add_table(table_name, table.schema, table)

Or, even better, automatically merge all declarative bases:

import gc
from sqlalchemy import MetaData

combined_meta_data = MetaData()

for declarative_base in ([obj for obj in gc.get_objects() if isinstance(obj, DeclarativeMeta)]):
    for (table_name, table) in declarative_base.metadata.tables.items():
        combined_meta_data._add_table(table_name, table.schema, table)

That meta data can then be used for example to create a diff to the current database:

from sqlalchemy import create_engine
from alembic.migration import MigrationContext
from alembic.autogenerate import compare_metadata
import pprint

engine = create_engine(...)
migration_context = MigrationContext.configure(engine.connect())

diff = compare_metadata(migration_context, combined_meta_data)
pprint.pprint(diff)
like image 180
Martin Loetzsch Avatar answered Oct 27 '22 07:10

Martin Loetzsch


sqlalchemy.Table.to_metadata appears to be a non-internal API that can copy a Table from one metadata object to another.

from sqlalchemy import MetaData
from sqlalchemy.ext.declarative import declarative_base
    
def merge_metadata(*original_metadata) -> MetaData:
    merged = MetaData()

    for original_metadatum in original_metadata:
        for table in original_metadatum.tables.values():
            table.to_metadata(merged)
    
    return merged

Base1 = declarative_base()
Base2 = declarative_base()
print(merge_metadata(Base1.metadata, Base2.metadata).tables)
like image 44
Ben Avatar answered Oct 27 '22 08:10

Ben