SQL 화학에서 삽입할 때 중복된 기본 키 처리(선언 스타일)

SQL 화학에서 삽입할 때 중복된 기본 키 처리(선언 스타일)

내 애플리케이션은 범위가 지정된 세션과 SQLAlchemy 선언 스타일을 사용하고 있습니다.웹 앱이며 은 이은웹앱많은며이 DB의입다에 됩니다.Celery작업 일정표

일반적으로 개체를 삽입하기로 결정할 때 내 코드는 다음과 같은 작업을 수행할 수 있습니다.

from schema import Session
from schema.models import Bike

pk = 123 # primary key
bike = Session.query(Bike).filter_by(bike_id=pk).first()
if not bike: # no bike in DB
    new_bike = Bike(pk, "shiny", "bike")
    Session.add(new_bike)
    Session.commit()

가 되는 것은 이 작업자에 입니다.Bike와 함께id=123다른 하나가 존재를 확인하는 동안. 두 키를 에서 SQLChemy의 SQLChemy를 .IntegrityError.

이 요. ㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠㅠSession.commit()선택사항:

'''schema/__init__.py'''
from sqlalchemy.orm import scoped_session, sessionmaker
Session = scoped_session(sessionmaker())

def commit(ignore=False):
    try:
        Session.commit()
    except IntegrityError as e:
        reason = e.message
        logger.warning(reason)

        if not ignore:
            raise e

        if "Duplicate entry" in reason:
            logger.info("%s already in table." % e.params[0])
            Session.rollback()

그리고 내가 가진 모든 곳에서Session.commit나는 지금 가지고 있습니다.schema.commit(ignore=True)행이 다시 삽입되지 않아도 상관없는 경우.

제가 보기에 이것은 끈 검사 때문에 매우 부서지기 쉬운 것처럼 보입니다.▁an,에로▁when▁just만약▁an▁as▁f고yi참.IntegrityError상승했습니다. 다음과 같이 보입니다.

(IntegrityError) (1062, "Duplicate entry '123' for key 'PRIMARY'")

그래서 물론 제가 삽입한 주요 열쇠는 다음과 같습니다.Duplicate entry is a cool thing그러면 내가 놓칠 수도 있을 것 같아요IntegrityError기본 키가 중복되었기 때문이 아닙니다.

문자열 등으로 문장을 작성하기 시작하는 것과 달리 제가 사용하고 있는 깨끗한 SQL 화학 접근 방식을 유지하는 더 나은 접근 방식이 있습니까?

Db는 MySQL입니다(단, 유닛 테스트의 경우 SQLite를 사용하는 것이 좋으며 새로운 접근 방식으로 이러한 기능을 방해하고 싶지 않습니다).

건배!

사용하는 경우session.merge(bike)에 session.add(bike)그러면 기본 키 오류가 발생하지 않습니다.bike필요에 따라 검색 및 업데이트되거나 생성됩니다.

은 모든 것을 .IntegrityError동일한 방법으로 트랜잭션을 롤백하고 선택적으로 다시 시도합니다.일부 데이터베이스는 사용자가 나중에 그 이상의 작업을 수행할 수 없도록 합니다.IntegrityError충돌하는 두 트랜잭션의 시작 부분에서 테이블의 잠금을 획득하거나 데이터베이스에서 허용하는 경우 보다 세밀한 잠금을 획득할 수도 있습니다.

with트랜잭션을 명시적으로 시작하고 자동으로 커밋(또는 예외에 대해 롤백)하는 문장:

from schema import Session
from schema.models import Bike

session = Session()
with session.begin():
    pk = 123 # primary key
    bike = session.query(Bike).filter_by(bike_id=pk).first()
    if not bike: # no bike in DB
        new_bike = Bike(pk, "shiny", "bike")
        session.add(new_bike)

에 에.session.add(obj)당신은 아래에 언급된 코드를 사용해야 합니다. 이것은 훨씬 깨끗할 것이고 당신이 언급한 것처럼 사용자 지정 커밋 기능을 사용할 필요가 없습니다.그러나 중복 키뿐만 아니라 다른 키의 충돌도 무시됩니다.

mysql:

 self.session.execute(insert(self.table, values=values, prefixes=['IGNORE']))

사철석의

self.session.execute(insert(self.table, values=values, prefixes=['OR IGNORE']))

여기서 기본 키는 어떤 면에서는 자연스러운 것이라고 생각합니다. 그렇기 때문에 일반적인 자동 증가 기술에 의존할 수 없습니다.그래서 문제가 삽입해야 하는 고유한 열 중 하나라고 가정해 보겠습니다. 이 열은 더 일반적입니다.

"실패 시 부분적으로 롤백 삽입 시도"를 원하는 경우 SAVEPOINT를 사용합니다. SAVEPOINT는 SQLLchemy가 begin_nested(다음 롤백) 또는 commit()인 경우 해당 SAVEPOINT에만 적용되며 더 큰 범위의 작업은 수행되지 않습니다.

그러나 전체적으로 여기서의 패턴은 정말 피해야 할 하나일 뿐입니다.당신이 여기서 정말 하고 싶은 것은 세 가지 중 하나입니다. 1.삽입해야 하는 동일한 키를 처리하는 동시 작업을 실행하지 마십시오. 2. 작업 중인 동시 키에 대해 작업을 어떻게든 동기화하고 3. 공통 서비스를 사용하여 작업별로 공유된 이러한 특정 유형의 새 레코드를 생성합니다(또는 작업이 실행되기 전에 모두 설정되었는지 확인).

생각해보면 #2는 고립도가 높은 어떤 경우에도 일어납니다.두 개의 사후 세션을 시작합니다.세션 1:

test=> create table foo(id integer primary key);
NOTICE:  CREATE TABLE / PRIMARY KEY will create implicit index "foo_pkey" for table "foo"
CREATE TABLE
test=> begin;
BEGIN
test=> insert into foo (id) values (1);

세션 2:

test=> begin;
BEGIN
test=> insert into foo(id) values(1);

PK #1이 있는 행이 잠겨 있는 경우 세션 2 블록이 표시됩니다.MySQL이 이 작업을 수행할 수 있을 정도로 똑똑한지는 모르겠지만, 그것이 올바른 동작입니다.OTOH가 다른 PK를 삽입하려는 경우:

^CCancel request sent
ERROR:  canceling statement due to user request
test=> rollback;
ROLLBACK
test=> begin;
BEGIN
test=> insert into foo(id) values(2);
INSERT 0 1
test=> \q

차단하지 않고 잘 진행됩니다.

중요한 것은 이러한 종류의 PK/UQ 경합을 수행하는 경우 셀러리 작업이 어쨌든 자체적으로 직렬화되거나 적어도 직렬화되어야 한다는 것입니다.

다음 코드를 사용하면 이 문제를 해결하는 것뿐만 아니라 원하는 모든 것을 할 수 있어야 합니다.

class SessionWrapper(Session):
    def commit(self, ignore=True):
        try:
            super(SessionWrapper, self).commit()
        except IntegrityError as e:
            if not ignore:
                raise e
            message = e.args[0]
            if "Duplicate entry" in message:
                logging.info("Error while executing %s.\n%s.", e.statement, message)
        finally:
            super(SessionWrapper, self).close()


def session(self, auto_commit=False):
    session_factory = sessionmaker(class_=SessionWrapper, bind=self.engine, autocommit=auto_commit)
    return scoped_session(session_factory)

Session = session()
s1 = Session()

p = Test(test="xxx", id=1)
s1.add(p)
s1.commit()
s1.close()

롤백하고 하나씩 다시 시도하기만 하면 됩니다.

try:
    self._session.bulk_insert_mappings(mapper, items)
    self._session.commit()
except IntegrityError:
    self._session.rollback()
    logger.info("bulk inserting rows failed, fallback to insert one-by-one")
    for item in items:
        try:
            self._session.execute(insert(mapper).values(**item))
            self._session.commit()
        except SQLAlchemyError as e:
            logger.error("Error inserting item: %s for %s", item, e)

언급URL : https://stackoverflow.com/questions/10322514/dealing-with-duplicate-primary-keys-on-insert-in-sqlalchemy-declarative-style