codekingpro/portable-devtools

1##########################################################################2#3# pgAdmin 4 - PostgreSQL Tools4#5# Copyright (C) 2013 - 2024, The pgAdmin Development Team6# This software is released under the PostgreSQL Licence7#8##########################################################################9 10from werkzeug.exceptions import HTTPException11from werkzeug.http import HTTP_STATUS_CODES12from flask_babel import gettext as _13from flask import request14 15from pgadmin.utils.ajax import service_unavailable, gone, internal_server_error16 17SERVICE_UNAVAILABLE = 'Service Unavailable'18 19 20class ConnectionLost(HTTPException):21    """22    Exception23    """24 25    def __init__(self, _server_id, _database_name, _conn_id):26        self.sid = _server_id27        self.db = _database_name28        self.conn_id = _conn_id29        HTTPException.__init__(self)30 31    @property32    def name(self):33        return HTTP_STATUS_CODES.get(503, SERVICE_UNAVAILABLE)34 35    def get_response(self, environ=None):36        return service_unavailable(37            _("Connection to the server has been lost."),38            info="CONNECTION_LOST",39            data={40                'sid': self.sid,41                'database': self.db,42                'conn_id': self.conn_id43            }44        )45 46    def __str__(self):47        return "Connection (id #{2}) lost for the server (#{0}) on " \48               "database ({1})".format(self.sid, self.db, self.conn_id)49 50    def __repr__(self):51        return "Connection (id #{2}) lost for the server (#{0}) on " \52               "database ({1})".format(self.sid, self.db, self.conn_id)53 54 55class SSHTunnelConnectionLost(HTTPException):56    """57    Exception when connection to SSH tunnel is lost58    """59 60    def __init__(self, _tunnel_host):61        self.tunnel_host = _tunnel_host62        HTTPException.__init__(self)63 64    @property65    def name(self):66        return HTTP_STATUS_CODES.get(503, SERVICE_UNAVAILABLE)67 68    def get_response(self, environ=None):69        return service_unavailable(70            _("Connection to the SSH Tunnel for host '{0}' has been lost. "71              "Reconnect to the database server.").format(self.tunnel_host),72            info="SSH_TUNNEL_CONNECTION_LOST",73            data={74                'tunnel_host': self.tunnel_host75            }76        )77 78    def __str__(self):79        return "Connection to the SSH Tunnel for host '{0}' has been lost. " \80               "Reconnect to the database server".format(self.tunnel_host)81 82    def __repr__(self):83        return "Connection to the SSH Tunnel for host '{0}' has been lost. " \84               "Reconnect to the database server".format(self.tunnel_host)85 86 87class CryptKeyMissing(HTTPException):88    """89    Exception90    """91    CRYPT_KEY_MISSING = "Crypt key is missing."92 93    def __init__(self):94        HTTPException.__init__(self)95 96    @property97    def name(self):98        return HTTP_STATUS_CODES.get(503, SERVICE_UNAVAILABLE)99 100    def get_response(self, environ=None):101        return service_unavailable(102            _(self.CRYPT_KEY_MISSING),103            info="CRYPTKEY_MISSING",104        )105 106    def __str__(self):107        return self.CRYPT_KEY_MISSING108 109    def __repr__(self):110        return self.CRYPT_KEY_MISSING111 112 113class ObjectGone(HTTPException):114    """115    Exception116    """117 118    def __init__(self, error_msg):119        self.error_msg = error_msg120        HTTPException.__init__(self)121 122    @property123    def name(self):124        return HTTP_STATUS_CODES.get(410, 'Gone')125 126    def get_response(self, environ=None):127        return gone(self.error_msg)128 129    def __str__(self):130        return self.error_msg131 132    def __repr__(self):133        return self.error_msg134 135 136class ExecuteError(HTTPException):137    """138    ExecuteError139    """140 141    def __init__(self, error_msg):142        self.error_msg = error_msg143        HTTPException.__init__(self)144 145    @property146    def name(self):147        return HTTP_STATUS_CODES.get(500, 'Internal server error')148 149    def get_response(self, environ=None):150        return internal_server_error(self.error_msg)151 152    def __str__(self):153        return self.error_msg154 155    def __repr__(self):156        return self.error_msg157