如何在 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

我使用 try-except 块来捕获它并且它可以工作。但是当我运行 pytests 时不会引发异常。

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)

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

我怎样才能捕捉到这个错误？

Answer 1

OperationalError 是错误的错误类。见Exceptions：

异常 psycopg2.OperationalError

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

您想要的广泛例外是：

异常 psycopg2.DatabaseError¶

与数据库相关的错误引发异常。它是错误的子类。

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

所以：

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

Answer 2

问题出在异常和日志记录中。有错误。我进行下一个修复：

    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

而且没关系