如何在Python中捕获psycopg2.errors.UniqueViolation错误？

Question

我有一个应用程序来处理使用psycopg2的数据库。当我尝试插入一个具有重复名称的列时，我得到了以下错误：

>           self.cursor.execute(str(query))
E           psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "deposits_area_name_key"
E           DETAIL:  Key (area_name)=(test-name) already exists.

dbase-api-server/dbase_api_server/dbase.py:110: UniqueViolation

我用试-除了阻止抓住它和它的工作。但是当我运行pytest时，异常不会产生。

import logging

from psycopg2 import connect as connect_to_db
from psycopg2._psycopg import connection as postgres_connection
from psycopg2._psycopg import cursor as db_cursor
from psycopg2.errors import UniqueViolation
from pypika import Query, Table

from dbase_api_server.containers import PostgresConnectionParams


class StorageDBase:
    def __init__(self, params: PostgresConnectionParams):
        try:
            self.__connection = connect_to_db(
                host=params.host,
                port=params.port,
                user=params.user,
                password=params.password,
                database=params.database
            )
        except OperationalError as err:
            logging.error(err, 'problems with database operation')
            raise

    @property
    def connection(self) -> postgres_connection:
        return self.__connection

    @property
    def cursor(self) -> db_cursor:
        return self.__connection.cursor()

    def is_success_commit(self) -> bool:
        self.connection.commit()

    def add_deposit_info(self, area_name: str) -> bool:
        table = Table('deposits')
        query = str(Query.into(table).columns('area_name').insert(area_name))
        try:
            self.cursor.execute(query)
            return self.is_success_commit()
        except UniqueViolation as error:
            logging.error(
                error,
                f'deposit with name {area_name} already exists'
            )
            return False

测试：

from hamcrest import assert_that, equal_to, is_

from dbase_api_server.dbase import StorageDBase


class TestStorageDBase:
    def test_add_deposit_repeating_name(self, up_test_dbase):
        area_name = 'test-name'
        is_first_added = up_test_dbase.add_deposit_info(area_name)
        assert_that(actual_or_assertion=is_first_added, matcher=is_(True))

        is_second_added = up_test_dbase.add_deposit_info(area_name)
        assert_that(actual_or_assertion=is_second_added, matcher=is_(False))

        query_count = f'SELECT COUNT(1) FROM deposits WHERE area_name=\'{area_name}\''
        cursor = up_test_dbase.cursor
        cursor.execute(query_count)
        records_count = cursor.fetchone()[0]
        assert_that(actual_or_assertion=records_count, matcher=equal_to(1))
        self.remove_records_from_deposits(dbase_adapter=up_test_dbase)

有一个类似的问题这里，但这个解决方案无助于解决问题。

我怎么能抓到这个错误？

Answer 1

OperationalError是错误的错误类。请参阅例外情况

异常心理2.因与数据库操作有关且不一定受程序员控制的错误而引发的操作错误，例如出现意外断开、找不到数据源名称、无法处理事务、处理过程中发生内存分配错误等。它是DatabaseError的子类。

你想要的最广泛的例外是：

如果与数据库相关的错误引发，则会引发异常。它是错误的子类。

如果您想要更细的粒度异常，则需要查看错误。例如，出现在错误消息中的psycopg2.errors.UniqueViolation。

所以：

try:
    <some query>
except psycopg2.errors.UniqueViolation:
    <do something>

Answer 2

问题是在异常和logging.There是错误的。我做了下一个修正：

    def add_deposit_info(self, area_name: str) -> bool:
        table = Table('deposits')
        query = str(Query.into(table).columns('area_name').insert(area_name))
        try:
            self.cursor.execute(query)
            return self.is_success_commit()
        except UniqueViolation:
            logging.error('field(s) has non unique value')
            return False

也没什么