Python django.db 模块，ProgrammingError() 实例源码

def check_duplicate_emails(app_configs=None, **kwargs):
    from accounts.utils import get_duplicate_emails
    errors = []
    try:
        if len(get_duplicate_emails()):
            errors.append(
                checks.Warning(
                    _("There are user accounts with duplicate emails. This "
                      "will not be allowed in Pootle 2.8."),
                    hint=_("Try using 'pootle find_duplicate_emails', and "
                           "then update user emails with 'pootle "
                           "update_user_email username email'. You might also "
                           "want to consider using pootle merge_user or "
                           "purge_user commands"),
                    id="pootle.W017"
                )
            )
    except (OperationalError, ProgrammingError):
        # no accounts set up - most likely in a test
        pass
    return errors

def check_users(app_configs=None, **kwargs):
    from django.contrib.auth import get_user_model

    errors = []

    User = get_user_model()
    try:
        admin_user = User.objects.get(username='admin')
    except (User.DoesNotExist, OperationalError, ProgrammingError):
        pass
    else:
        if admin_user.check_password('admin'):
            errors.append(checks.Warning(
                _("The default 'admin' user still has a password set to "
                  "'admin'."),
                hint=_("Remove the 'admin' user or change its password."),
                id="pootle.W016",
            ))

    return errors

def check_revision(app_configs=None, **kwargs):
    from pootle.core.models import Revision
    from pootle_store.models import Unit

    errors = []
    revision = Revision.get()
    try:
        max_revision = Unit.max_revision()
    except (OperationalError, ProgrammingError):
        return errors
    if revision is None or revision < max_revision:
        errors.append(checks.Critical(
            _("Revision is missing or has an incorrect value."),
            hint=_("Run `revision --restore` to reset the revision counter."),
            id="pootle.C016",
        ))

    return errors

def fetch_column(self, database_name, table_name, column_name):
        # prepare sql string
        sql = 'SHOW FULL COLUMNS FROM %(database)s.%(table)s WHERE `Field` = %(column)s' % {
            'database': self.escape_identifier(database_name),
            'table': self.escape_identifier(table_name),
            'column': self.escape_string(column_name)
        }

        # execute query
        try:
            row = self.fetchone(sql)
        except ProgrammingError as e:
            logger.error('Could not fetch %s.%s.%s (%s)' % (database_name, table_name, column_name, e))
            return {}
        else:
            return {
                'name': row[0],
                'datatype': row[1],
                'indexed': bool(row[4])
            }

def prepare_database(self):
        """Ran to prepare the configured database.

        This is where we enable the `hstore` extension
        if it wasn't enabled yet."""

        super().prepare_database()
        with self.cursor() as cursor:
            try:
                cursor.execute('CREATE EXTENSION IF NOT EXISTS hstore')
            except ProgrammingError:  # permission denied
                logger.warning(
                    'Failed to create "hstore" extension. '
                    'Tables with hstore columns may fail to migrate. '
                    'If hstore is needed, make sure you are connected '
                    'to the database as a superuser '
                    'or add the extension manually.',
                    exc_info=True)

def get_queryset(self):
        try:
            return OCDCandidacyProxy.objects.filter(party__in=[
                OCDPartyProxy.objects.unknown()
            ])
        except (OCDPartyProxy.DoesNotExist, ProgrammingError):
            return OCDCandidacyProxy.objects.none()

def get_app_patterns():
    try:
        return _get_app_patterns()
    except (OperationalError, ProgrammingError):
        # ignore if DB is not ready
        # Starting with Django 1.9 this code gets called even when creating
        # or running migrations. So in many cases the DB will not be ready yet.
        return []

def handle(self, *args, **opts):
        for _, shard in settings.SHARDS.items():
            cursor = shard.connection._nodb_connection.cursor()
            try:
                cursor.execute('CREATE DATABASE %s' % shard.database)
                self.stdout.write(self.style.MIGRATE_HEADING(
                    'Database [%s] created.' % shard.database
                ))
            except ProgrammingError as e:
                self.stderr.write(str(e))
        self.stdout.write(self.style.MIGRATE_HEADING('Done.\n'))

def fetch_columns(self, database_name, table_name):
        # prepare sql string
        sql = 'SHOW FULL COLUMNS FROM %(database)s.%(table)s;' % {
            'database': self.escape_identifier(database_name),
            'table': self.escape_identifier(table_name)
        }

        # execute query
        try:
            rows = self.fetchall(sql)
        except ProgrammingError as e:
            logger.error('Could not fetch from %s.%s (%s)' % (database_name, table_name, e))
            return []
        else:
            column_metadata = []
            for row in rows:
                datatype, arraysize = self.convert_datatype(row[1])

                column_metadata.append({
                    'name': row[0],
                    'datatype': datatype,
                    'arraysize': arraysize,
                    'indexed': bool(row[4])
                })

            return column_metadata

def test_other_db_error_not_handled_by_us(self):
        with self.assertRaises(ProgrammingError):
            self.client.get('/sql/?sql=foo')

def test_missing_table_when_schema_set(self):
        with self.assertRaises(ProgrammingError):
            self.client.get('/sql/?sql=SELECT+1+FROM+foo.bar')

def __call__(self, request):
        try:
            return self.process_request(request) or self.get_response(request)
        except (TemplateSchemaActivation, ProgrammingError) as exception:
            return self.process_exception(request, exception)

def process_exception(self, request, exception):
        """
        In the case a request returned a DatabaseError, and there was no
        schema set on ``request.session``, then look and see if the error
        that was provided by the database may indicate that we should have
        been looking inside a schema.

        In the case we had a :class:`TemplateSchemaActivation` exception,
        then we want to remove that key from the session.
        """
        if isinstance(exception, ProgrammingError) and not request.session.get('schema'):
            if re.search('relation ".*" does not exist', exception.args[0]):
                # Should we return an error, or redirect? When should we
                # do one or the other? For an API, we would want an error
                # but for a regular user, a redirect may be better.
                if request.is_ajax():
                    return HttpResponse(
                        _('You must select a schema to access that resource'),
                        status=400
                    )
                # Can we see if there is already a pending message for this
                # request that has the same content as us?
                messages.error(
                    request,
                    _("You must select a schema to access that resource"),
                    fail_silently=True
                )
                return HttpResponseRedirect('..')
        # I'm not sure we ever really hit this one, but it's worth keeping
        # here just in case we've missed something. I guess it could occur
        # if a view manually attempted to activate the template schema.
        if isinstance(exception, TemplateSchemaActivation):
            request.session.pop('schema', None)
            return HttpResponseForbidden(_('You may not select that schema'))

        raise exception

def process_exception(self, request, exception):

        # Don't catch exceptions when in debug mode
        if settings.DEBUG:
            return

        # Determine the type of exception
        if isinstance(exception, ProgrammingError):
            template_name = 'exceptions/programming_error.html'
        elif isinstance(exception, ImportError):
            template_name = 'exceptions/import_error.html'
        elif (
            sys.version_info[0] >= 3 and isinstance(exception, PermissionError)
        ) or (
            isinstance(exception, OSError) and exception.errno == 13
        ):
            template_name = 'exceptions/permission_error.html'
        else:
            template_name = '500.html'

        # Return an error message
        type_, error, traceback = sys.exc_info()
        return render(request, template_name, {
            'exception': str(type_),
            'error': error,
        }, status=500)