I have app for working with database that uses psycopg2. When I try to insert a column with repeating name I got this error:

>           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

I use try-except block to catch it and it work. But exception doesn't raise when I run 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

Test:

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)

There was a similar question here, but this solutions doesn't help to solve problem.

How I can catch this error?

CodePudding user response：

OperationalError is the wrong error class. See Exceptions:

exception psycopg2.OperationalError

Exception raised for errors that are related to the database’s operation and not necessarily under the control of the programmer, e.g. an unexpected disconnect occurs, the data source name is not found, a transaction could not be processed, a memory allocation error occurred during processing, etc. It is a subclass of DatabaseError.

The broad exception you want is:

exception psycopg2.DatabaseError¶

Exception raised for errors that are related to the database. It is a subclass of Error.

If you want finer grain exceptions you need to look at Errors. For instance the psycopg2.errors.UniqueViolation that shows up in your error message.

So:

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

CodePudding user response：

Problem is in exceptions and logging.There was mistake. I make next fix:

    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

And it`s okay