sqlalchemy проверить на дубликат перед вставкой

Но зачем? По исключению и поймёшь это.

Отлови исключение и обработай, чо как не питонист

просто думал, что вдруг существует цивилизованный способ проверки на дубликат, как в списке, например elem in list1.

фигня какая то - не ловиться исключение, ни большое, ни маленькое:

from sqlalchemy.exc import IntegrityError

def insert(myTable: MyTable) -> None:
    with sessionScope() as session:
        try:
            session.add(myTable)
        except IntegrityError as e:
            print('pass duplicate')
        except Exception as e:
            print(e)
            raise

# ...
    myTable = MyTable(name='x', date='2021-04-30', count=22)
    insert(myTable)

IntegrityError: (psycopg2.IntegrityError) duplicate key value violates unique constraint "uniq_idx_1"
DETAIL:  Key (name, date)=(x, 2021-04-30) already exists.

Это не цивилизованный способ, а лишний запрос, к тому же бесполезный без блокировки. Обработка ошибки — самый оптимальный и единственный корректный способ.

А если отлавливать не session.add, а with sessionScope()?

psycopg2.IntegrityError

Внимательней.

А почему алхимия не перехватывает его не выбрасывает sqlalchemy.exc.IntegrityError? Неудобно однако.

то работает )))

но почему?

понятно psycopg2.IntegrityError

на практике, в контекстном менеджере вот так перехватывает

from sqlalchemy.exc import IntegrityError

def sessionScope():
    try:
        session = Session(bind=engine)
        yield session
        session.commit()
    except IntegrityError:
        print('pass duplicate')
        session.rollback()
    except:
        session.rollback()
        raise

def insert(myTable: MyTable) -> None:
    with sessionScope() as session:
        session.add(myTable)

except:

никогда так не делай. в крайнем случае except Exception:. в твоём же случае session.rollback() лучше делать в finally:. а ещё в алхимии должен быть контекстный менеджер (engine.begin() или лучше connection.begin()), который обрабатывает транзакцию, тебе не нужно его реализовывать

Спасибо! Мне показалось, что в случае exception я должен откатиться, а в finaly я сделал session.close() … буду разбираться

в случае exception я должен откатиться, а в finaly я сделал session.close()

да, верно. тут моя ошибка

Не, я чот тупанул. У него ни коммита, ни flush. Оно на выходе из контектс-менеджера срабатывает.

Так сделай коммит сразу после вызова add.

Да, судя по всему так и есть.

Я сутал с поведением при работе с asyncpg, там отдельная UniqueViolationError из драйвера не перехватывается.

Алхимия далекро не такая консистентная как принято считать. Зачастую приходится аргументы непосредственно в инстанс драйвера пробрасывать, ошибки ловить и руками какие-то бд-специфичные фичи костылять.

Select for update сначала Потом проверка Insert

Так будет работать

Если запись с уникальным вторичным ключем уже существует, то обновлять не нужно. И по сути, я выбираю между: 1) сделать запрос и проверить наличие записи (select только приходит на ум) и потом вставить 2) попробовать сразу вставить и, если запись есть - обработать исключение

Или другими словами:

1) не допускать вставки

def insert(myTable: MyTable) -> None:
    with sessionScope() as session:
        row = session.query(MyTable) \
                     .filter(MyTable.name==myTable.name) \
                     .filter(MyTable.date==myTable.date) \
                     .all()
        if len(row) == 0:
            session.add(myTable)

2) по Питонски в контекстный манагер добавить пропуск исключения

@contextmanager
def sessionScope():
    try:
        session = Session(bind=engine)
        yield session
        session.commit()
    except IntegrityError as e:
        print('pass')
        session.rollback()
    # ...

def insert(myTable: MyTable) -> None:
    with sessionScope() as session:
        session.add(myTable)

IntegrityError кажется может возникать и по другим причинам ( ну или добавите новый уник индекс, надо будет код как то править)

А для селекта как раз и надо делать for update иначе может быть гонка