我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用django.db.utils.OperationalError()。
def ready(self): from django.core.management import call_command from morango.models import InstanceIDModel from morango.certificates import ScopeDefinition from .signals import add_to_deleted_models # noqa: F401 # NOTE: Warning: https://docs.djangoproject.com/en/1.10/ref/applications/#django.apps.AppConfig.ready # its recommended not to execute queries in this method, but we are producing the same result after the first call, so its OK # call this on app load up to get most recent system config settings try: InstanceIDModel.get_or_create_current_instance() if not ScopeDefinition.objects.filter(): call_command("loaddata", "scopedefinitions") # we catch this error in case the database has not been migrated, b/c we can't query it until its been created except (OperationalError, ProgrammingError): pass # add models to be synced by profile add_syncable_models()
def remove_field(self, model, field): if isinstance(field, GeometryField) and field.spatial_index: qn = self.connection.ops.quote_name sql = self.sql_drop_spatial_index % { 'index': qn(self._create_spatial_index_name(model, field)), 'table': qn(model._meta.db_table), } try: self.execute(sql) except OperationalError: logger.error( "Couldn't remove spatial index: %s (may be expected " "if your storage engine doesn't support them)." % sql ) super(MySQLGISSchemaEditor, self).remove_field(model, field)
def remove_field(self, model, field): if isinstance(field, GeometryField) and field.spatial_index: qn = self.connection.ops.quote_name sql = self.sql_drop_spatial_index % { 'index': qn(self._create_spatial_index_name(model, field)), 'table': qn(model._meta.db_table), } try: self.execute(sql) except OperationalError: logger.error( "Couldn't remove spatial index: %s (may be expected " "if your storage engine doesn't support them).", sql ) super(MySQLGISSchemaEditor, self).remove_field(model, field)
def check_database_connected(app_configs, **kwargs): """ A Django check to see if connecting to the configured default database backend succeeds. """ errors = [] try: connection.ensure_connection() except OperationalError as e: msg = 'Could not connect to database: {!s}'.format(e) errors.append(checks.Error(msg, id=ERROR_CANNOT_CONNECT_DATABASE)) except ImproperlyConfigured as e: msg = 'Datbase misconfigured: "{!s}"'.format(e) errors.append(checks.Error(msg, id=ERROR_MISCONFIGURED_DATABASE)) else: if not connection.is_usable(): errors.append(checks.Error('Database connection is not usable', id=ERROR_UNUSABLE_DATABASE)) return errors
def ready(self): from pretalx.event.models import Event from pretalx.common.tasks import regenerate_css from django.db import connection, utils if Event._meta.db_table not in connection.introspection.table_names(): # commands like `compilemessages` execute ready(), but do not # require a database to be present. Bail out early, if the Event # table has not been created yet. return try: for event in Event.objects.all(): regenerate_css.apply_async(args=(event.pk,)) except (utils.OperationalError, utils.ProgrammingError): pass
def handle(self, *args, **options): changed = set() self.stdout.write("Checking...") for db in settings.DATABASES.keys(): try: executor = MigrationExecutor(connections[db]) except OperationalError: sys.exit("Unable to check migrations: cannot connect to database\n") autodetector = MigrationAutodetector( executor.loader.project_state(), ProjectState.from_apps(apps), ) changed.update(autodetector.changes(graph=executor.loader.graph).keys()) changed -= set(options['ignore']) if changed: sys.exit( "Apps with model changes but no corresponding migration file: %(changed)s\n" % { 'changed': list(changed) }) else: sys.stdout.write("All migration files present\n")
def create_spatial_indexes(self): for sql in self.geometry_sql: try: self.execute(sql) except OperationalError: logger.error( "Cannot create SPATIAL INDEX %s. Only MyISAM and (as of " "MySQL 5.7.5) InnoDB support them." % sql ) self.geometry_sql = []
def reconnect_db(logfunc, default=None, retry_seconds=None): ''' Catch Operational errors from the database and try to reconnect. ''' def _decorator(func): def _wrapper(*args, **kwargs): while True: try: return func(*args, **kwargs) except OperationalError as e: logfunc(e) close_old_connections() if retry_seconds is None: return default else: time.sleep(retry_seconds) _wrapper.__doc__ = func.__doc__ return _wrapper return _decorator
def create_profile(sender, **kwargs): user = kwargs["instance"] if kwargs["created"]: default_plan = Plan.objects.get(pk=Plan.DEFAULT_PK) up = Profile(user=user, plan=default_plan) up.save() try: for tg in TalkGroupAccess.objects.filter(default_group=True): up.talkgroup_access.add(tg) except OperationalError: pass try: new_user_email = SiteOption.objects.get(name='SEND_ADMIN_EMAIL_ON_NEW_USER') if new_user_email.value_boolean_or_string() == True: send_mail( 'New {} User {}'.format(settings.SITE_TITLE, user.username), 'New User {} {} Username {} Email {} just registered'.format(user.first_name, user.last_name, user.username, user.email), settings.SERVER_EMAIL, [ mail for name, mail in settings.ADMINS], fail_silently=False, ) except (SiteOption.DoesNotExist, OperationalError): pass
def reset_connection_on_interface_error(func): """Decorates function to reset the current thread's Django database connection on exceptions that appear to come from connection resets. """ def _reset(*args, **kwargs): try: return func(*args, **kwargs) except (InterfaceError, OperationalError, DjangoInterfaceError, DjangoOperationalError) as error: thread = threading.current_thread() _logger.warning("it appears this thread's database connection was " "dropped, resetting it now - you may see further " "errors about this until the situation is fully " "resolved for all threads " "(this thread is '%s', error was '%s')", thread.name, error) django.db.connection.connection = None raise ResetDBConnectionError("The database connection was reset", error) return wraps(func)(_reset)
def generate_response(msg): """ indices_list = [] try: indices_list = QueryResults.objects.get(keyname = sno) except OperationalError: pass """ if 'search' in msg: q = ''.join([ix for ix in msg.split('search', 1)[1]]) #result = get_data(q) #display the data resulting from a query search return get_data(q) elif int(msg) in range(1,100): #fetch the link of that video link = get_video(int(msg)) return link else: return "Use search 'search-term' to look for the video you want" #return msg
def handle(self, *args, **options): try: connection = connections[options['database']] except ConnectionDoesNotExist: raise CommandError('Database "%s" does not exist in settings' % options['database']) for i in range(0, options['retries']): try: connection.cursor() except OperationalError: i += 1 self.stdout.write('{} / {}: Waiting for database...'.format(i, options['retries'])) sleep(options['sleep_time']) else: self.stdout.write(self.style.SUCCESS('Successfully connected to database')) return raise CommandError('Number of retries reached, exiting')
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 check_migrations(): from django.db.migrations.autodetector import MigrationAutodetector from django.db.migrations.executor import MigrationExecutor from django.db.migrations.state import ProjectState changed = set() print("Checking {} migrations...".format(APP_NAME)) for db in settings.DATABASES.keys(): try: executor = MigrationExecutor(connections[db]) except OperationalError as e: sys.exit( "Unable to check migrations due to database: {}".format(e) ) autodetector = MigrationAutodetector( executor.loader.project_state(), ProjectState.from_apps(apps), ) changed.update( autodetector.changes(graph=executor.loader.graph).keys() ) if changed and APP_NAME in changed: sys.exit( "A migration file is missing. Please run " "`python makemigrations.py` to generate it." ) else: print("All migration files present.")
def get_for_model(self, model, for_concrete_model=True): """ Returns the ContentType object for a given model, creating the ContentType if necessary. Lookups are cached so that subsequent lookups for the same model don't hit the database. """ opts = self._get_opts(model, for_concrete_model) try: return self._get_from_cache(opts) except KeyError: pass # The ContentType entry was not found in the cache, therefore we # proceed to load or create it. try: try: # We start with get() and not get_or_create() in order to use # the db_for_read (see #20401). ct = self.get(app_label=opts.app_label, model=opts.model_name) except self.model.DoesNotExist: # Not found in the database; we proceed to create it. This time we # use get_or_create to take care of any race conditions. ct, created = self.get_or_create( app_label=opts.app_label, model=opts.model_name, ) except (OperationalError, ProgrammingError, IntegrityError): # It's possible to migrate a single app before contenttypes, # as it's not a required initial dependency (it's contrib!) # Have a nice error for this. raise RuntimeError( "Error creating new content types. Please make sure contenttypes " "is migrated before trying to migrate apps individually." ) self._add_to_cache(self.db, ct) return ct
def create_spatial_indexes(self): for sql in self.geometry_sql: try: self.execute(sql) except OperationalError: logger.error( "Cannot create SPATIAL INDEX %s. Only MyISAM and (as of " "MySQL 5.7.5) InnoDB support them.", sql ) self.geometry_sql = []
def ready(self): from .models import Notification from herald import registry self.module.autodiscover() try: # add any new notifications to database. for index, klass in enumerate(registry._registry): notification, created = Notification.objects.get_or_create( notification_class=klass.get_class_path(), defaults={ 'verbose_name': klass.get_verbose_name(), 'can_disable': klass.can_disable, } ) if not created: notification.verbose_name = klass.get_verbose_name() notification.can_disable = klass.can_disable notification.save() except OperationalError: # if the table is not created yet, just keep going. pass except ProgrammingError: # if the database is not created yet, keep going (ie: during testing) pass
def check_migrations_applied(app_configs, **kwargs): """ A Django check to see if all migrations have been applied correctly. """ from django.db.migrations.loader import MigrationLoader errors = [] # Load migrations from disk/DB try: loader = MigrationLoader(connection, ignore_no_migrations=True) except (ImproperlyConfigured, ProgrammingError, OperationalError): msg = "Can't connect to database to check migrations" return [checks.Info(msg, id=INFO_CANT_CHECK_MIGRATIONS)] if app_configs: app_labels = [app.label for app in app_configs] else: app_labels = loader.migrated_apps for node, migration in loader.graph.nodes.items(): if migration.app_label not in app_labels: continue if node not in loader.applied_migrations: msg = 'Unapplied migration {}'.format(migration) # NB: This *must* be a Warning, not an Error, because Errors # prevent migrations from being run. errors.append(checks.Warning(msg, id=WARNING_UNAPPLIED_MIGRATION)) return errors
