我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用django.db.DEFAULT_DB_ALIAS。
def test_find_unsafe_migrations(self): conn = connections[DEFAULT_DB_ALIAS] result = migration_utils.find_unsafe_migrations(conn) assert len(result) == 0 self.add_django_app('tests.unsafe_migrations') result = migration_utils.find_unsafe_migrations(conn) assert len(result) == 2 unsafemodel_field_added = result[0] assert unsafemodel_field_added.app_name == 'unsafe_migrations' assert unsafemodel_field_added.migration_name == '0002_unsafemodel_field_added' assert len(unsafemodel_field_added.offending_operations) == 1 unsafemodel_kitchen_sink = result[1] assert unsafemodel_kitchen_sink.app_name == 'unsafe_migrations' assert unsafemodel_kitchen_sink.migration_name == '0003_unsafemodel_kitchen_sink' assert len(unsafemodel_kitchen_sink.offending_operations) == 5
def add_arguments(self, parser): parser.add_argument('app_label', nargs='?', help='App label of an application to synchronize the state.') parser.add_argument('migration_name', nargs='?', help=( 'Database state will be brought to the state after that ' 'migration. Use the name "zero" to unapply all migrations.' ), ) parser.add_argument('--noinput', '--no-input', action='store_false', dest='interactive', default=True, help='Tells Django to NOT prompt the user for input of any kind.') parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database to synchronize. ' 'Defaults to the "default" database.') parser.add_argument('--fake', action='store_true', dest='fake', default=False, help='Mark migrations as run without actually running them.') parser.add_argument('--fake-initial', action='store_true', dest='fake_initial', default=False, help='Detect if tables already exist and fake-apply initial migrations if so. Make sure ' 'that the current database schema matches your initial migration before using this ' 'flag. Django will only check for an existing table name.') parser.add_argument('--list', '-l', action='store_true', dest='list', default=False, help='Show a list of all known migrations and which are applied.') parser.add_argument('--run-syncdb', action='store_true', dest='run_syncdb', help='Creates tables for apps without migrations.')
def check_migrations(self): """ Checks to see if the set of migrations on disk matches the migrations in the database. Prints a warning if they don't match. """ try: executor = MigrationExecutor(connections[DEFAULT_DB_ALIAS]) except ImproperlyConfigured: # No databases are configured (or the dummy one) return except MigrationSchemaMissing: self.stdout.write(self.style.NOTICE( "\nNot checking migrations as it is not possible to access/create the django_migrations table." )) return plan = executor.migration_plan(executor.loader.graph.leaf_nodes()) if plan: self.stdout.write(self.style.NOTICE( "\nYou have unapplied migrations; your app may not work properly until they are applied." )) self.stdout.write(self.style.NOTICE("Run 'python manage.py migrate' to apply them.\n")) # Kept for backward compatibility
def add_arguments(self, parser): parser.add_argument('--%s' % self.UserModel.USERNAME_FIELD, dest=self.UserModel.USERNAME_FIELD, default=None, help='Specifies the login for the superuser.') parser.add_argument('--noinput', '--no-input', action='store_false', dest='interactive', default=True, help=('Tells Django to NOT prompt the user for input of any kind. ' 'You must use --%s with --noinput, along with an option for ' 'any other required field. Superusers created with --noinput will ' ' not be able to log in until they\'re given a valid password.' % self.UserModel.USERNAME_FIELD)) parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Specifies the database to use. Default is "default".') for field in self.UserModel.REQUIRED_FIELDS: parser.add_argument('--%s' % field, dest=field, default=None, help='Specifies the %s for the superuser.' % field)
def _nodb_connection(self): nodb_connection = super(DatabaseWrapper, self)._nodb_connection try: nodb_connection.ensure_connection() except (DatabaseError, WrappedDatabaseError): warnings.warn( "Normally Django will use a connection to the 'postgres' database " "to avoid running initialization queries against the production " "database when it's not needed (for example, when running tests). " "Django was unable to create a connection to the 'postgres' database " "and will use the default database instead.", RuntimeWarning ) settings_dict = self.settings_dict.copy() settings_dict['NAME'] = settings.DATABASES[DEFAULT_DB_ALIAS]['NAME'] nodb_connection = self.__class__( self.settings_dict.copy(), alias=self.alias, allow_thread_sharing=False) return nodb_connection
def is_nullable(self, field): """ A helper to check if the given field should be treated as nullable. Some backends treat '' as null and Django treats such fields as nullable for those backends. In such situations field.null can be False even if we should treat the field as nullable. """ # We need to use DEFAULT_DB_ALIAS here, as QuerySet does not have # (nor should it have) knowledge of which connection is going to be # used. The proper fix would be to defer all decisions where # is_nullable() is needed to the compiler stage, but that is not easy # to do currently. if ((connections[DEFAULT_DB_ALIAS].features.interprets_empty_strings_as_nulls) and field.empty_strings_allowed): return True else: return field.null
def add_arguments(self, parser): parser.add_argument( 'app_label', nargs='*', help='App labels of applications to limit the output to.', ) parser.add_argument( '--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database to synchronize. Defaults to the "default" database.', ) formats = parser.add_mutually_exclusive_group() formats.add_argument( '--list', '-l', action='store_const', dest='format', const='list', help='Shows a list of all migrations and which are applied.', ) formats.add_argument( '--plan', '-p', action='store_const', dest='format', const='plan', help=( 'Shows all migrations in the order they will be applied. ' 'With a verbosity level of 2 or above all direct migration dependencies ' 'and reverse dependencies (run_before) will be included.' ) ) parser.set_defaults(format='list')
def reset_sequences(*models): """ After loading data the sequences must be reset in the database if the primary keys are manually specified. This is handled automatically by django for fixtures. Much of this is modeled after django.core.management.commands.loaddata. """ # connection = connections[DEFAULT_DB_ALIAS] # cursor = connection.cursor() # sequence_sql = connection.ops.sequence_reset_sql(no_style(), models) # if sequence_sql: # for line in sequence_sql: # cursor.execute(line) # # transaction.commit_unless_managed() # cursor.close() pass
def inject_pre_migration_operations(plan=None, apps=global_apps, using=DEFAULT_DB_ALIAS, **kwargs): """ Insert a `LTreeExtension` operation before every planned `CreateModel` operation. """ if plan is None: return for migration, backward in plan: for index, operation in enumerate(migration.operations): if isinstance(operation, migrations.CreateModel): for name, field in operation.fields: if isinstance(field, LTreeField): migration.operations.insert(index, LTreeExtension()) return
def inject_post_migration_operations(plan=None, apps=global_apps, using=DEFAULT_DB_ALIAS, **kwargs): if plan is None: return for migration, backward in plan: for index, operation in reversed(list(enumerate(migration.operations))): if isinstance(operation, migrations.CreateModel): model = apps.get_model(migration.app_label, operation.name) post_migrate_mpathnode(model)
def add_arguments(self, parser): parser.add_argument('app_label', nargs='?', help='App label of an application to synchronize the state.') parser.add_argument('migration_name', nargs='?', help=( 'Database state will be brought to the state after that ' 'migration. Use the name "zero" to unapply all migrations.' ), ) parser.add_argument('--noinput', action='store_false', dest='interactive', default=True, help='Tells Django to NOT prompt the user for input of any kind.') parser.add_argument('--no-initial-data', action='store_false', dest='load_initial_data', default=True, help='Tells Django not to load any initial data after database synchronization.') parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database to synchronize. ' 'Defaults to the "default" database.') parser.add_argument('--fake', action='store_true', dest='fake', default=False, help='Mark migrations as run without actually running them.') parser.add_argument('--fake-initial', action='store_true', dest='fake_initial', default=False, help='Detect if tables already exist and fake-apply initial migrations if so. Make sure ' 'that the current database schema matches your initial migration before using this ' 'flag. Django will only check for an existing table name.') parser.add_argument('--list', '-l', action='store_true', dest='list', default=False, help='Show a list of all known migrations and which are applied.')
def add_arguments(self, parser): parser.add_argument('--%s' % self.UserModel.USERNAME_FIELD, dest=self.UserModel.USERNAME_FIELD, default=None, help='Specifies the login for the superuser.') parser.add_argument('--noinput', action='store_false', dest='interactive', default=True, help=('Tells Django to NOT prompt the user for input of any kind. ' 'You must use --%s with --noinput, along with an option for ' 'any other required field. Superusers created with --noinput will ' ' not be able to log in until they\'re given a valid password.' % self.UserModel.USERNAME_FIELD)) parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Specifies the database to use. Default is "default".') for field in self.UserModel.REQUIRED_FIELDS: parser.add_argument('--%s' % field, dest=field, default=None, help='Specifies the %s for the superuser.' % field)
def add_arguments(self, parser): parser.add_argument('app_label', nargs='?', help='App label of an application to synchronize the state.') parser.add_argument('migration_name', nargs='?', help=( 'Database state will be brought to the state after that ' 'migration. Use the name "zero" to unapply all migrations.' ), ) parser.add_argument('--noinput', action='store_false', dest='interactive', default=True, help='Tells Django to NOT prompt the user for input of any kind.') parser.add_argument('--no-initial-data', action='store_false', dest='load_initial_data', default=True, help='Tells Django not to load any initial data after database synchronization.') parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database to synchronize. ' 'Defaults to the "default" database.') parser.add_argument('--fake', action='store_true', dest='fake', default=False, help='Mark migrations as run without actually running them') parser.add_argument('--fake-initial', action='store_true', dest='fake_initial', default=False, help='Detect if tables already exist and fake-apply initial migrations if so. Make sure ' 'that the current database schema matches your initial migration before using this ' 'flag. Django will only check for an existing table name.') parser.add_argument('--list', '-l', action='store_true', dest='list', default=False, help='Show a list of all known migrations and which are applied')
def wait_for_db(max_attempts=15, seconds_between_attempts=1): # type: (int, int) -> None ''' Some manage.py commands interact with the database, and we want them to be directly callable from `docker-compose run`. However, because docker may start the database container at the same time as it runs `manage.py`, we potentially face a race condition, and the manage.py command may attempt to connect to a database that isn't yet ready for connections. To alleviate this, we'll just wait for the database before calling the manage.py command. ''' from django.db import DEFAULT_DB_ALIAS, connections from django.db.utils import OperationalError connection = connections[DEFAULT_DB_ALIAS] attempts = 0 while True: try: connection.ensure_connection() break except OperationalError as e: if attempts >= max_attempts: raise e attempts += 1 time.sleep(seconds_between_attempts) info("Attempting to connect to database.") info("Connection to database established.")
def test_find_migration_conflicts(self): conn = connections[DEFAULT_DB_ALIAS] self.add_django_app('tests.conflicting_migrations') result = migration_utils.find_unsafe_migrations(conn) assert result[0].app_name == 'conflicting_migrations' assert result[0].migration_names == { '0002_unsafemodel_some_other_changes', '0002_unsafemodel_first_changes' }
def __init__(self, stream_or_string, **options): super(Deserializer, self).__init__(stream_or_string, **options) self.event_stream = pulldom.parse(self.stream, self._make_parser()) self.db = options.pop('using', DEFAULT_DB_ALIAS) self.ignore = options.pop('ignorenonexistent', False)
def add_arguments(self, parser): super(Command, self).add_arguments(parser) parser.add_argument('--database', default=DEFAULT_DB_ALIAS, help='Nominates a database to print the SQL for. Defaults to the ' '"default" database.')
def add_arguments(self, parser): parser.add_argument('app_labels', nargs='*', help='App labels of applications to limit the output to.') parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database to synchronize. Defaults to the "default" database.') formats = parser.add_mutually_exclusive_group() formats.add_argument('--list', '-l', action='store_const', dest='format', const='list', help='Shows a list of all migrations and which are applied.') formats.add_argument('--plan', '-p', action='store_const', dest='format', const='plan', help='Shows all migrations in the order they will be applied.') parser.set_defaults(format='list')
def add_arguments(self, parser): parser.add_argument('app_label', help='App label of the application containing the migration.') parser.add_argument('migration_name', help='Migration name to print the SQL for.') parser.add_argument('--database', default=DEFAULT_DB_ALIAS, help='Nominates a database to create SQL for. Defaults to the ' '"default" database.') parser.add_argument('--backwards', action='store_true', dest='backwards', default=False, help='Creates SQL to unapply the migration, rather than to apply it')
def add_arguments(self, parser): parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database onto which to ' 'open a shell. Defaults to the "default" database.')
def add_arguments(self, parser): parser.add_argument('args', metavar='app_label[.ModelName]', nargs='*', help='Restricts dumped data to the specified app_label or app_label.ModelName.') parser.add_argument('--format', default='json', dest='format', help='Specifies the output serialization format for fixtures.') parser.add_argument('--indent', default=None, dest='indent', type=int, help='Specifies the indent level to use when pretty-printing output.') parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a specific database to dump fixtures from. ' 'Defaults to the "default" database.') parser.add_argument('-e', '--exclude', dest='exclude', action='append', default=[], help='An app_label or app_label.ModelName to exclude ' '(use multiple --exclude to exclude multiple apps/models).') parser.add_argument('--natural-foreign', action='store_true', dest='use_natural_foreign_keys', default=False, help='Use natural foreign keys if they are available.') parser.add_argument('--natural-primary', action='store_true', dest='use_natural_primary_keys', default=False, help='Use natural primary keys if they are available.') parser.add_argument('-a', '--all', action='store_true', dest='use_base_manager', default=False, help="Use Django's base manager to dump all models stored in the database, " "including those that would otherwise be filtered or modified by a custom manager.") parser.add_argument('--pks', dest='primary_keys', help="Only dump objects with given primary keys. " "Accepts a comma separated list of keys. " "This option will only work when you specify one model.") parser.add_argument('-o', '--output', default=None, dest='output', help='Specifies file to which the output is written.')
def add_arguments(self, parser): parser.add_argument('username', nargs='?', help='Username to change password for; by default, it\'s the current username.') parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Specifies the database to use. Default is "default".')
def bulk_related_objects(self, objs, using=DEFAULT_DB_ALIAS): """ Return all objects related to ``objs`` via this ``GenericRelation``. """ return self.remote_field.model._base_manager.db_manager(using).filter(**{ "%s__pk" % self.content_type_field_name: ContentType.objects.db_manager(using).get_for_model( self.model, for_concrete_model=self.for_concrete_model).pk, "%s__in" % self.object_id_field_name: [obj.pk for obj in objs] })
def create_default_site(app_config, verbosity=2, interactive=True, using=DEFAULT_DB_ALIAS, **kwargs): try: Site = apps.get_model('sites', 'Site') except LookupError: return if not router.allow_migrate_model(using, Site): return if not Site.objects.using(using).exists(): # The default settings set SITE_ID = 1, and some tests in Django's test # suite rely on this value. However, if database sequences are reused # (e.g. in the test suite after flush/syncdb), it isn't guaranteed that # the next id will be 1, so we coerce it. See #15573 and #16353. This # can also crop up outside of tests - see #15346. if verbosity >= 2: print("Creating example.com Site object") Site(pk=getattr(settings, 'SITE_ID', 1), domain="example.com", name="example.com").save(using=using) # We set an explicit pk instead of relying on auto-incrementation, # so we need to reset the database sequence. See #17415. sequence_sql = connections[using].ops.sequence_reset_sql(no_style(), [Site]) if sequence_sql: if verbosity >= 2: print("Resetting sequence") with connections[using].cursor() as cursor: for command in sequence_sql: cursor.execute(command)
def sql_with_params(self): """ Returns the query as an SQL string and the parameters that will be substituted into the query. """ return self.get_compiler(DEFAULT_DB_ALIAS).as_sql()
def prepare_lookup_value(self, value, lookups, can_reuse, allow_joins=True): # Default lookup if none given is exact. used_joins = [] if len(lookups) == 0: lookups = ['exact'] # Interpret '__exact=None' as the sql 'is NULL'; otherwise, reject all # uses of None as a query value. if value is None: if lookups[-1] not in ('exact', 'iexact'): raise ValueError("Cannot use None as a query value") lookups[-1] = 'isnull' value = True elif hasattr(value, 'resolve_expression'): pre_joins = self.alias_refcount.copy() value = value.resolve_expression(self, reuse=can_reuse, allow_joins=allow_joins) used_joins = [k for k, v in self.alias_refcount.items() if v > pre_joins.get(k, 0)] # Subqueries need to use a different set of aliases than the # outer query. Call bump_prefix to change aliases of the inner # query (the value). if hasattr(value, 'query') and hasattr(value.query, 'bump_prefix'): value = value._clone() value.query.bump_prefix(self) if hasattr(value, 'bump_prefix'): value = value.clone() value.bump_prefix(self) # For Oracle '' is equivalent to null. The check needs to be done # at this stage because join promotion can't be done at compiler # stage. Using DEFAULT_DB_ALIAS isn't nice, but it is the best we # can do here. Similar thing is done in is_nullable(), too. if (connections[DEFAULT_DB_ALIAS].features.interprets_empty_strings_as_nulls and lookups[-1] == 'exact' and value == ''): value = True lookups[-1] = 'isnull' return value, lookups, used_joins
def assertNumQueries(self, num, func=None, *args, **kwargs): using = kwargs.pop("using", DEFAULT_DB_ALIAS) conn = connections[using] context = _AssertNumQueriesContext(self, num, conn) if func is None: return context with context: func(*args, **kwargs)
def check_migrations(self): """ Print a warning if the set of migrations on disk don't match the migrations in the database. """ from django.db.migrations.executor import MigrationExecutor try: executor = MigrationExecutor(connections[DEFAULT_DB_ALIAS]) except ImproperlyConfigured: # No databases are configured (or the dummy one) return except MigrationSchemaMissing: self.stdout.write(self.style.NOTICE( "\nNot checking migrations as it is not possible to access/create the django_migrations table." )) return plan = executor.migration_plan(executor.loader.graph.leaf_nodes()) if plan: apps_waiting_migration = sorted(set(migration.app_label for migration, backwards in plan)) self.stdout.write( self.style.NOTICE( "\nYou have %(unpplied_migration_count)s unapplied migration(s). " "Your project may not work properly until you apply the " "migrations for app(s): %(apps_waiting_migration)s." % { "unpplied_migration_count": len(plan), "apps_waiting_migration": ", ".join(apps_waiting_migration), } ) ) self.stdout.write(self.style.NOTICE("Run 'python manage.py migrate' to apply them.\n"))
def add_arguments(self, parser): super(Command, self).add_arguments(parser) parser.add_argument( '--database', default=DEFAULT_DB_ALIAS, help='Nominates a database to print the SQL for. Defaults to the "default" database.', )
def add_arguments(self, parser): parser.add_argument( '--noinput', '--no-input', action='store_false', dest='interactive', default=True, help='Tells Django to NOT prompt the user for input of any kind.', ) parser.add_argument( '--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database to flush. Defaults to the "default" database.', )
def add_arguments(self, parser): parser.add_argument( 'app_label', nargs='?', help='App label of an application to synchronize the state.', ) parser.add_argument( 'migration_name', nargs='?', help='Database state will be brought to the state after that ' 'migration. Use the name "zero" to unapply all migrations.', ) parser.add_argument( '--noinput', '--no-input', action='store_false', dest='interactive', default=True, help='Tells Django to NOT prompt the user for input of any kind.', ) parser.add_argument( '--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database to synchronize. Defaults to the "default" database.', ) parser.add_argument( '--fake', action='store_true', dest='fake', default=False, help='Mark migrations as run without actually running them.', ) parser.add_argument( '--fake-initial', action='store_true', dest='fake_initial', default=False, help='Detect if tables already exist and fake-apply initial migrations if so. Make sure ' 'that the current database schema matches your initial migration before using this ' 'flag. Django will only check for an existing table name.', ) parser.add_argument( '--run-syncdb', action='store_true', dest='run_syncdb', help='Creates tables for apps without migrations.', )
def add_arguments(self, parser): parser.add_argument('app_label', help='App label of the application containing the migration.') parser.add_argument('migration_name', help='Migration name to print the SQL for.') parser.add_argument( '--database', default=DEFAULT_DB_ALIAS, help='Nominates a database to create SQL for. Defaults to the "default" database.', ) parser.add_argument( '--backwards', action='store_true', dest='backwards', default=False, help='Creates SQL to unapply the migration, rather than to apply it', )
def add_arguments(self, parser): parser.add_argument( '--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Nominates a database onto which to open a shell. Defaults to the "default" database.', )
def add_arguments(self, parser): parser.add_argument( '--%s' % self.UserModel.USERNAME_FIELD, dest=self.UserModel.USERNAME_FIELD, default=None, help='Specifies the login for the superuser.', ) parser.add_argument( '--noinput', '--no-input', action='store_false', dest='interactive', default=True, help=( 'Tells Django to NOT prompt the user for input of any kind. ' 'You must use --%s with --noinput, along with an option for ' 'any other required field. Superusers created with --noinput will ' 'not be able to log in until they\'re given a valid password.' % self.UserModel.USERNAME_FIELD ), ) parser.add_argument( '--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Specifies the database to use. Default is "default".', ) for field in self.UserModel.REQUIRED_FIELDS: parser.add_argument( '--%s' % field, dest=field, default=None, help='Specifies the %s for the superuser.' % field, )
def add_arguments(self, parser): parser.add_argument( 'username', nargs='?', help='Username to change password for; by default, it\'s the current username.', ) parser.add_argument( '--database', action='store', dest='database', default=DEFAULT_DB_ALIAS, help='Specifies the database to use. Default is "default".', )
def inject_rename_contenttypes_operations(plan=None, apps=global_apps, using=DEFAULT_DB_ALIAS, **kwargs): """ Insert a `RenameContentType` operation after every planned `RenameModel` operation. """ if plan is None: return # Determine whether or not the ContentType model is available. try: ContentType = apps.get_model('contenttypes', 'ContentType') except LookupError: available = False else: if not router.allow_migrate_model(using, ContentType): return available = True for migration, backward in plan: if ((migration.app_label, migration.name) == ('contenttypes', '0001_initial')): # There's no point in going forward if the initial contenttypes # migration is unapplied as the ContentType model will be # unavailable from this point. if backward: break else: available = True continue # The ContentType model is not available yet. if not available: continue inserts = [] for index, operation in enumerate(migration.operations): if isinstance(operation, migrations.RenameModel): operation = RenameContentType( migration.app_label, operation.old_name_lower, operation.new_name_lower ) inserts.append((index + 1, operation)) for inserted, (index, operation) in enumerate(inserts): migration.operations.insert(inserted + index, operation)
