我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用django.db.router.get_migratable_models()。
def sequence_list(self): "Returns a list of information about all DB sequences for all models in all apps." from django.apps import apps from django.db import models, router sequence_list = [] for app_config in apps.get_app_configs(): for model in router.get_migratable_models(app_config, self.connection.alias): if not model._meta.managed: continue if model._meta.swapped: continue for f in model._meta.local_fields: if isinstance(f, models.AutoField): sequence_list.append({'table': model._meta.db_table, 'column': f.column}) break # Only one AutoField is allowed per model, so don't bother continuing. for f in model._meta.local_many_to_many: # If this is an m2m using an intermediate table, # we don't need to reset the sequence. if f.rel.through is None: sequence_list.append({'table': f.m2m_db_table(), 'column': None}) return sequence_list
def sequence_list(self): "Returns a list of information about all DB sequences for all models in all apps." from django.apps import apps from django.db import models, router sequence_list = [] for app_config in apps.get_app_configs(): for model in router.get_migratable_models(app_config, self.connection.alias): if not model._meta.managed: continue if model._meta.swapped: continue for f in model._meta.local_fields: if isinstance(f, models.AutoField): sequence_list.append({'table': model._meta.db_table, 'column': f.column}) break # Only one AutoField is allowed per model, so don't bother continuing. for f in model._meta.local_many_to_many: # If this is an m2m using an intermediate table, # we don't need to reset the sequence. if f.remote_field.through is None: sequence_list.append({'table': f.m2m_db_table(), 'column': None}) return sequence_list
def django_table_names(self, only_existing=False, include_views=True): """ Returns a list of all table names that have associated Django models and are in INSTALLED_APPS. If only_existing is True, the resulting list will only include the tables that actually exist in the database. """ from django.apps import apps from django.db import router tables = set() for app_config in apps.get_app_configs(): for model in router.get_migratable_models(app_config, self.connection.alias): if not model._meta.managed: continue tables.add(model._meta.db_table) tables.update(f.m2m_db_table() for f in model._meta.local_many_to_many) tables = list(tables) if only_existing: existing_tables = self.table_names(include_views=include_views) tables = [ t for t in tables if self.table_name_converter(t) in existing_tables ] return tables
def installed_models(self, tables): "Returns a set of all models represented by the provided list of table names." from django.apps import apps from django.db import router all_models = [] for app_config in apps.get_app_configs(): all_models.extend(router.get_migratable_models(app_config, self.connection.alias)) tables = list(map(self.table_name_converter, tables)) return { m for m in all_models if self.table_name_converter(m._meta.db_table) in tables }
def emit_post_migrate(verbosity, interactive, database): # Emit the post migrate signal. This allows individual applications to # respond as if the database had been migrated from scratch. all_models = [] for app_config in apps.get_app_configs(): all_models.extend(router.get_migratable_models(app_config, database, include_auto_created=True)) emit_post_migrate_signal(set(all_models), verbosity, interactive, database)
def sql_custom(app_config, style, connection): "Returns a list of the custom table modifying SQL statements for the given app." check_for_migrations(app_config, connection) output = [] app_models = router.get_migratable_models(app_config, connection.alias) for model in app_models: output.extend(custom_sql_for_model(model, style, connection)) return output
def sql_indexes(app_config, style, connection): "Returns a list of the CREATE INDEX SQL statements for all models in the given app." check_for_migrations(app_config, connection) output = [] for model in router.get_migratable_models(app_config, connection.alias, include_auto_created=True): output.extend(connection.creation.sql_indexes_for_model(model, style)) return output
def django_table_names(self, only_existing=False, include_views=True): """ Returns a list of all table names that have associated Django models and are in INSTALLED_APPS. If only_existing is True, the resulting list will only include the tables that actually exist in the database. """ from django.apps import apps from django.db import router tables = set() for app_config in apps.get_app_configs(): for model in router.get_migratable_models(app_config, self.connection.alias): if not model._meta.managed: continue tables.add(model._meta.db_table) tables.update( f.m2m_db_table() for f in model._meta.local_many_to_many if f.remote_field.through._meta.managed ) tables = list(tables) if only_existing: existing_tables = self.table_names(include_views=include_views) tables = [ t for t in tables if self.table_name_converter(t) in existing_tables ] return tables
def sql_create(app_config, style, connection): "Returns a list of the CREATE TABLE SQL statements for the given app." if connection.settings_dict['ENGINE'] == 'django.db.backends.dummy': # This must be the "dummy" database backend, which means the user # hasn't set ENGINE for the database. raise CommandError( "Django doesn't know which syntax to use for your SQL statements," "\nbecause you haven't properly specified the ENGINE setting for " "the database.\n" "see: https://docs.djangoproject.com/en/%s/ref/settings/" "#databases" % get_docs_version() ) # Get installed models, so we generate REFERENCES right. # We trim models from the current app so that the sqlreset command does not # generate invalid SQL (leaving models out of known_models is harmless, so # we can be conservative). final_output = [] output = [] for model in router.get_migratable_models( app_config, connection.alias, include_auto_created=True): with connection.schema_editor() as editor: output += sql_create_model(editor, model) final_output.extend(editor.deferred_sql) editor.deferred_sql = [] return output + final_output
def sync_apps(self, connection, app_labels): "Runs the old syncdb-style operation on a list of app_labels." cursor = connection.cursor() try: # Get a list of already installed *models* so that references work right. tables = connection.introspection.table_names(cursor) created_models = set() # Build the manifest of apps and models that are to be synchronized all_models = [ (app_config.label, router.get_migratable_models(app_config, connection.alias, include_auto_created=False)) for app_config in apps.get_app_configs() if app_config.models_module is not None and app_config.label in app_labels ] def model_installed(model): opts = model._meta converter = connection.introspection.table_name_converter # Note that if a model is unmanaged we short-circuit and never try to install it return not ((converter(opts.db_table) in tables) or (opts.auto_created and converter(opts.auto_created._meta.db_table) in tables)) manifest = OrderedDict( (app_name, list(filter(model_installed, model_list))) for app_name, model_list in all_models ) # Create the tables for each model if self.verbosity >= 1: self.stdout.write(" Creating tables...\n") with transaction.atomic(using=connection.alias, savepoint=connection.features.can_rollback_ddl): deferred_sql = [] for app_name, model_list in manifest.items(): for model in model_list: if not model._meta.can_migrate(connection): continue if self.verbosity >= 3: self.stdout.write( " Processing %s.%s model\n" % (app_name, model._meta.object_name) ) with connection.schema_editor() as editor: if self.verbosity >= 1: self.stdout.write(" Creating table %s\n" % model._meta.db_table) editor.create_model(model) deferred_sql.extend(editor.deferred_sql) editor.deferred_sql = [] created_models.add(model) if self.verbosity >= 1: self.stdout.write(" Running deferred SQL...\n") for statement in deferred_sql: cursor.execute(statement) finally: cursor.close() return created_models
def sync_apps(self, connection, app_labels): "Runs the old syncdb-style operation on a list of app_labels." cursor = connection.cursor() try: # Get a list of already installed *models* so that references work right. tables = connection.introspection.table_names(cursor) created_models = set() # Build the manifest of apps and models that are to be synchronized all_models = [ (app_config.label, router.get_migratable_models(app_config, connection.alias, include_auto_created=False)) for app_config in apps.get_app_configs() if app_config.models_module is not None and app_config.label in app_labels ] def model_installed(model): opts = model._meta converter = connection.introspection.table_name_converter # Note that if a model is unmanaged we short-circuit and never try to install it return not ( (converter(opts.db_table) in tables) or (opts.auto_created and converter(opts.auto_created._meta.db_table) in tables) ) manifest = OrderedDict( (app_name, list(filter(model_installed, model_list))) for app_name, model_list in all_models ) # Create the tables for each model if self.verbosity >= 1: self.stdout.write(" Creating tables...\n") with transaction.atomic(using=connection.alias, savepoint=connection.features.can_rollback_ddl): deferred_sql = [] for app_name, model_list in manifest.items(): for model in model_list: if not model._meta.can_migrate(connection): continue if self.verbosity >= 3: self.stdout.write( " Processing %s.%s model\n" % (app_name, model._meta.object_name) ) with connection.schema_editor() as editor: if self.verbosity >= 1: self.stdout.write(" Creating table %s\n" % model._meta.db_table) editor.create_model(model) deferred_sql.extend(editor.deferred_sql) editor.deferred_sql = [] created_models.add(model) if self.verbosity >= 1: self.stdout.write(" Running deferred SQL...\n") for statement in deferred_sql: cursor.execute(statement) finally: cursor.close() return created_models
def sync_apps(self, connection, app_labels): """Run the old syncdb-style operation on a list of app_labels.""" with connection.cursor() as cursor: tables = connection.introspection.table_names(cursor) # Build the manifest of apps and models that are to be synchronized. all_models = [ ( app_config.label, router.get_migratable_models(app_config, connection.alias, include_auto_created=False), ) for app_config in apps.get_app_configs() if app_config.models_module is not None and app_config.label in app_labels ] def model_installed(model): opts = model._meta converter = connection.introspection.table_name_converter return not ( (converter(opts.db_table) in tables) or (opts.auto_created and converter(opts.auto_created._meta.db_table) in tables) ) manifest = OrderedDict( (app_name, list(filter(model_installed, model_list))) for app_name, model_list in all_models ) # Create the tables for each model if self.verbosity >= 1: self.stdout.write(" Creating tables...\n") with connection.schema_editor() as editor: for app_name, model_list in manifest.items(): for model in model_list: # Never install unmanaged models, etc. if not model._meta.can_migrate(connection): continue if self.verbosity >= 3: self.stdout.write( " Processing %s.%s model\n" % (app_name, model._meta.object_name) ) if self.verbosity >= 1: self.stdout.write(" Creating table %s\n" % model._meta.db_table) editor.create_model(model) # Deferred SQL is executed when exiting the editor's context. if self.verbosity >= 1: self.stdout.write(" Running deferred SQL...\n")
def sql_create(app_config, style, connection): "Returns a list of the CREATE TABLE SQL statements for the given app." check_for_migrations(app_config, connection) if connection.settings_dict['ENGINE'] == 'django.db.backends.dummy': # This must be the "dummy" database backend, which means the user # hasn't set ENGINE for the database. raise CommandError( "Django doesn't know which syntax to use for your SQL statements,\n" "because you haven't properly specified the ENGINE setting for the database.\n" "see: https://docs.djangoproject.com/en/%s/ref/settings/#databases" % get_docs_version() ) # Get installed models, so we generate REFERENCES right. # We trim models from the current app so that the sqlreset command does not # generate invalid SQL (leaving models out of known_models is harmless, so # we can be conservative). app_models = list(app_config.get_models(include_auto_created=True)) final_output = [] tables = connection.introspection.table_names() known_models = set(model for model in connection.introspection.installed_models(tables) if model not in app_models) pending_references = {} for model in router.get_migratable_models(app_config, connection.alias, include_auto_created=True): output, references = connection.creation.sql_create_model(model, style, known_models) final_output.extend(output) for refto, refs in references.items(): pending_references.setdefault(refto, []).extend(refs) if refto in known_models: final_output.extend(connection.creation.sql_for_pending_references(refto, style, pending_references)) final_output.extend(connection.creation.sql_for_pending_references(model, style, pending_references)) # Keep track of the fact that we've created the table for this model. known_models.add(model) # Handle references to tables that are from other apps # but don't exist physically. not_installed_models = set(pending_references.keys()) if not_installed_models: alter_sql = [] for model in not_installed_models: alter_sql.extend('-- ' + sql for sql in connection.creation.sql_for_pending_references(model, style, pending_references)) if alter_sql: final_output.append('-- The following references should be added but depend on non-existent tables:') final_output.extend(alter_sql) return final_output
def sql_delete(app_config, style, connection, close_connection=True): "Returns a list of the DROP TABLE SQL statements for the given app." check_for_migrations(app_config, connection) # This should work even if a connection isn't available try: cursor = connection.cursor() except Exception: cursor = None try: # Figure out which tables already exist if cursor: table_names = connection.introspection.table_names(cursor) else: table_names = [] output = [] # Output DROP TABLE statements for standard application tables. to_delete = set() references_to_delete = {} app_models = router.get_migratable_models(app_config, connection.alias, include_auto_created=True) for model in app_models: if cursor and connection.introspection.table_name_converter(model._meta.db_table) in table_names: # The table exists, so it needs to be dropped opts = model._meta for f in opts.local_fields: if f.rel and f.rel.to not in to_delete: references_to_delete.setdefault(f.rel.to, []).append((model, f)) to_delete.add(model) for model in app_models: if connection.introspection.table_name_converter(model._meta.db_table) in table_names: output.extend(connection.creation.sql_destroy_model(model, references_to_delete, style)) finally: # Close database connection explicitly, in case this output is being piped # directly into a database client, to avoid locking issues. if cursor and close_connection: cursor.close() connection.close() if not output: output.append('-- App creates no tables in the database. Nothing to do.') return output[::-1] # Reverse it, to deal with table dependencies.
