def aiopg_exception_handling()

in services/data/db_utils.py [0:0]

• 35 lines of code
• 9 McCabe index (conditional complexity)

def aiopg_exception_handling(exception):
    err_msg = str(exception)
    body = {"err_msg": err_msg}
    if isinstance(exception, asyncio.TimeoutError):
        body = {
            "err_msg": {
                "type": "timeout error",
            }
        }
    elif isinstance(exception, psycopg2.Error):
        # this means that this is a psycopg2 exception
        # since this is of type `psycopg2.Error` we can use https://www.psycopg.org/docs/module.html#psycopg2.Error
        body = {
            "err_msg": {
                "pgerror": exception.pgerror,
                "pgcode": exception.pgcode,
                "diag": None
                if exception.diag is None
                else {
                    "message_primary": exception.diag.message_primary,
                    "severity": exception.diag.severity,
                },
            }
        }

    if isinstance(exception, psycopg2.IntegrityError):
        if "duplicate key" in err_msg:
            return DBResponse(response_code=409, body=json.dumps(body))
        elif "foreign key" in err_msg:
            return DBResponse(response_code=404, body=json.dumps(body))
        else:
            return DBResponse(response_code=500, body=json.dumps(body))
    elif isinstance(exception, psycopg2.errors.UniqueViolation):
        return DBResponse(response_code=409, body=json.dumps(body))
    elif isinstance(exception, IndexError):
        return DBResponse(response_code=404, body={})
    else:
        return DBResponse(response_code=500, body=json.dumps(body))