sqlalchemy.exc OperationalError Example Code

# errors.py

from __future__ import absolute_import, unicode_literals

import traceback
from uuid import uuid4

from flask import g, jsonify, render_template, request, session
from itsdangerous import BadData
from sqlalchemy.exc import OperationalError
from werkzeug.exceptions import Forbidden, HTTPException

from indico.core.errors import NoReportError
from indico.legacy.common.cache import GenericCache
from indico.web.util import get_request_info
from indico.web.views import WPError


def render_error(exc, title, message, code, standalone=False):
    _save_error(exc, title, message)
    if _need_json_response():
        return _jsonify_error(exc, title, message, code)
    elif standalone:
        return render_template('standalone_error.html', error_message=title, error_description=message), code
    else:
        try:
            return WPError(title, message).getHTML(), code
        except OperationalError:
            return render_error(exc, title, message, code, standalone=True)


def load_error_data(uuid):
    return GenericCache('errors').get(uuid)


def _save_error(exc, title, message):
    if 'saved_error_uuid' in g:
        return
    if not _is_error_reportable(exc):
        return
    g.saved_error_uuid = uuid = unicode(uuid4())
    tb = traceback.format_exc()
    data = {'title': title,
            'message': message,
            'request_info': get_request_info(),
            'traceback': tb,
            'sentry_event_id': g.get('sentry_event_id')}
    GenericCache('errors').set(uuid, data, 7200)


def _need_json_response():
    return request.is_xhr or request.is_json


## ... source file continues with no further OperationalError examples...

# database.py
import itertools
import os
from collections.abc import Mapping, Sequence
from copy import copy

import sqlalchemy as sa
from sqlalchemy.engine.url import make_url
from sqlalchemy.exc import OperationalError, ProgrammingError

from ..utils import starts_with
from .orm import quote


def escape_like(string, escape_char='*'):
    return (
        string
        .replace(escape_char, escape_char * 2)
        .replace('%', escape_char + '%')
        .replace('_', escape_char + '_')
    )


def json_sql(value, scalars_to_json=True):
    scalar_convert = sa.text
    if scalars_to_json:
        def scalar_convert(a):
            return sa.func.to_json(sa.text(a))

    if isinstance(value, Mapping):
        return sa.func.json_build_object(
            *(
                json_sql(v, scalars_to_json=False)


## ... source file abbreviated to get to OperationalError examples ...


        text = "SELECT 1 FROM pg_database WHERE datname='%s'" % database
        return bool(get_scalar_result(engine, text))

    elif engine.dialect.name == 'mysql':
        text = ("SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA "
                "WHERE SCHEMA_NAME = '%s'" % database)
        return bool(get_scalar_result(engine, text))

    elif engine.dialect.name == 'sqlite':
        if database:
            return database == ':memory:' or sqlite_file_exists(database)
        else:
            return True

    else:
        engine.dispose()
        engine = None
        text = 'SELECT 1'
        try:
            url.database = database
            engine = sa.create_engine(url)
            result = engine.execute(text)
            result.close()
            return True

        except (ProgrammingError, OperationalError):
            return False
        finally:
            if engine is not None:
                engine.dispose()


def create_database(url, encoding='utf8', template=None):

    url = copy(make_url(url))

    database = url.database

    if url.drivername.startswith('postgres'):
        url.database = 'postgres'
    elif url.drivername.startswith('mssql'):
        url.database = 'master'
    elif not url.drivername.startswith('sqlite'):
        url.database = None

    if url.drivername == 'mssql+pyodbc':
        engine = sa.create_engine(url, connect_args={'autocommit': True})
    elif url.drivername == 'postgresql+pg8000':
        engine = sa.create_engine(url, isolation_level='AUTOCOMMIT')
    else:


## ... source file continues with no further OperationalError examples...