我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用django.db.utils.ProgrammingError()。
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 _execute_wrapper(self, method, query, args): """Wrapper around execute() and executemany()""" try: return method(query, args) except (PyMysqlPool.mysql.connector.ProgrammingError) as err: six.reraise(utils.ProgrammingError, utils.ProgrammingError(err.msg), sys.exc_info()[2]) except (PyMysqlPool.mysql.connector.IntegrityError) as err: six.reraise(utils.IntegrityError, utils.IntegrityError(err.msg), sys.exc_info()[2]) except PyMysqlPool.mysql.connector.OperationalError as err: # Map some error codes to IntegrityError, since they seem to be # misclassified and Django would prefer the more logical place. if err.args[0] in self.codes_for_integrityerror: six.reraise(utils.IntegrityError, utils.IntegrityError(err.msg), sys.exc_info()[2]) else: six.reraise(utils.DatabaseError, utils.DatabaseError(err.msg), sys.exc_info()[2]) except PyMysqlPool.mysql.connector.DatabaseError as err: six.reraise(utils.DatabaseError, utils.DatabaseError(err.msg), sys.exc_info()[2])
def _execute_wrapper(self, method, query, args): """Wrapper around execute() and executemany()""" try: return method(query, args) except (mysql.connector.ProgrammingError) as err: six.reraise(utils.ProgrammingError, utils.ProgrammingError(err.msg), sys.exc_info()[2]) except (mysql.connector.IntegrityError) as err: six.reraise(utils.IntegrityError, utils.IntegrityError(err.msg), sys.exc_info()[2]) except mysql.connector.OperationalError as err: # Map some error codes to IntegrityError, since they seem to be # misclassified and Django would prefer the more logical place. if err.args[0] in self.codes_for_integrityerror: six.reraise(utils.IntegrityError, utils.IntegrityError(err.msg), sys.exc_info()[2]) else: six.reraise(utils.DatabaseError, utils.DatabaseError(err.msg), sys.exc_info()[2]) except mysql.connector.DatabaseError as err: six.reraise(utils.DatabaseError, utils.DatabaseError(err.msg), sys.exc_info()[2])
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 spatial_version(self): """Determine the version of the PostGIS library.""" # Trying to get the PostGIS version because the function # signatures will depend on the version used. The cost # here is a database query to determine the version, which # can be mitigated by setting `POSTGIS_VERSION` with a 3-tuple # comprising user-supplied values for the major, minor, and # subminor revision of PostGIS. if hasattr(settings, 'POSTGIS_VERSION'): version = settings.POSTGIS_VERSION else: try: vtup = self.postgis_version_tuple() except ProgrammingError: raise ImproperlyConfigured( 'Cannot determine PostGIS version for database "%s". ' 'GeoDjango requires at least PostGIS version 1.5. ' 'Was the database created from a spatial database ' 'template?' % self.connection.settings_dict['NAME'] ) version = vtup[1:] return version
def handle(self, *args, **options): print ">>> Initializing your database" try: management.call_command('syncdb') management.call_command('migrate') try: # do we need cache table? cache.get('', None) except ProgrammingError: # yes we do management.call_command('createcachetable', 'vaultier_cache') # public static files management.call_command('collectstatic', interactive=False) except OperationalError as e: msg = ">>> Your DB is not configured correctly: {}" print msg.format(e.message) else: if options.get('no_statistics'): task_statistics_collector.delay() print (">>> DB is initialized, you can now try to run Vaultier " "using 'vaultier runserver'")
def get_current_version_from_table(): """ Return the current version of the database, from sql_version table. Return None if the table does not exist (schema not inited). """ with connection.cursor() as cursor: try: cursor.execute("SELECT version_num FROM sql_version;") except ProgrammingError: # table does not exist ? return None rows = cursor.fetchall() versions = [row[0] for row in rows if is_version(row[0])] if not versions: return None # sort versions versions.sort(key=StrictVersion) # return the last one return versions[-1]
def get_current_version_from_comment(): """ Return the current version of the database, from django_site comment. Return None if the table django_site does not exist (schema not inited). """ with connection.cursor() as cursor: try: cursor.execute( "SELECT obj_description('django_site'::regclass, 'pg_class');") except ProgrammingError: # table does not exist ? return None row = cursor.fetchone() comment = row[0] # no comment if comment is None: raise DBException('No comment found on django_site.') # parse comment if 'version ' not in comment: raise DBException("No version found in django_site's comment.") return comment.replace('version ', '').strip()
def forwards(self, orm): """Change IPMI nodes with default power_driver to LAN 2.0.""" ipmi_nodes = orm['maasserver.Node'].objects.filter(power_type='ipmi') for node in ipmi_nodes: if node.power_parameters.get('power_driver') == IPMI_DRIVER.DEFAULT: node.power_parameters['power_driver'] = IPMI_DRIVER.LAN_2_0 try: node.save() except ProgrammingError: # We catch django.db.utils.ProgrammingError here because of # the failure described on bug 1302156. Although we didn't # manage to recreate the failure in a controlled environment, # this try/except statement will prevent the migration from # crashing if the failure reappears. logger.exception( "Failed to apply migration 0072_remove_ipmi_autodetect. " "See bug 1302156 " "(https://bugs.launchpad.net/maas/+bug/1302156).")
def _drop_view(self, cursor): try: cursor.execute(self._drop_view_sql()) except ProgrammingError: cursor.execute(self._drop_table_sql())
def _drop_view(self, cursor): try: cursor.execute(self._drop_materialized_view_sql()) except ProgrammingError: try: cursor.execute(self._drop_view_sql()) except ProgrammingError: cursor.execute(self._drop_table_sql())
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 spatial_version(self): """Determine the version of the PostGIS library.""" # Trying to get the PostGIS version because the function # signatures will depend on the version used. The cost # here is a database query to determine the version, which # can be mitigated by setting `POSTGIS_VERSION` with a 3-tuple # comprising user-supplied values for the major, minor, and # subminor revision of PostGIS. if hasattr(settings, 'POSTGIS_VERSION'): version = settings.POSTGIS_VERSION else: # Run a basic query to check the status of the connection so we're # sure we only raise the error below if the problem comes from # PostGIS and not from PostgreSQL itself (see #24862). self._get_postgis_func('version') try: vtup = self.postgis_version_tuple() except ProgrammingError: raise ImproperlyConfigured( 'Cannot determine PostGIS version for database "%s" ' 'using command "SELECT postgis_lib_version()". ' 'GeoDjango requires at least PostGIS version 2.0. ' 'Was the database created from a spatial database ' 'template?' % self.connection.settings_dict['NAME'] ) version = vtup[1:] return version
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 sizeAttr(): try: return min(40, Language.objects.count()) except ProgrammingError: # If the database has not tables this would break otherwise return 40
def spatial_version(self): """Determine the version of the PostGIS library.""" # Trying to get the PostGIS version because the function # signatures will depend on the version used. The cost # here is a database query to determine the version, which # can be mitigated by setting `POSTGIS_VERSION` with a 3-tuple # comprising user-supplied values for the major, minor, and # subminor revision of PostGIS. if hasattr(settings, 'POSTGIS_VERSION'): version = settings.POSTGIS_VERSION else: # Run a basic query to check the status of the connection so we're # sure we only raise the error below if the problem comes from # PostGIS and not from PostgreSQL itself (see #24862). self._get_postgis_func('version') try: vtup = self.postgis_version_tuple() except ProgrammingError: raise ImproperlyConfigured( 'Cannot determine PostGIS version for database "%s" ' 'using command "SELECT postgis_lib_version()". ' 'GeoDjango requires at least PostGIS version 2.1. ' 'Was the database created from a spatial database ' 'template?' % self.connection.settings_dict['NAME'] ) version = vtup[1:] return version
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
def test_filter_without_shard_key(self): with self.assertRaises(ProgrammingError): list(OneModel.objects.all())
def db(self): if self._db: return self._db if self._horizontal_key is None: raise ProgrammingError("Missing horizontal key field's filter") self._add_hints(horizontal_key=self._horizontal_key) return super(HorizontalQuerySet, self).db
def ready(self): """ """ from .models import Stream if not _IN_TESTING_MODE: try: Stream.objects.close_all() close_old_connections() except ProgrammingError: # if migrations have not been run pass
def drop_table(self): # drop the corresponding database table, but fail silently try: get_adapter().database.drop_table(self.database_name, self.table_name) except ProgrammingError: pass
