在 SQLAlchemy 中检测唯一约束的违反

问题描述 投票：0回答：1

因为以下代码示例中的 

Country.name
有 
unique=True
，并且创建了两个同名的国家/地区，所以我预计 
session.commit()
会失败。

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


Base = declarative_base()


class Country(Base):
    __tablename__ = 'countries'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(64), unique=True, nullable=False, index=True)


engine = create_engine('sqlite:///countries_example.db')
Base.metadata.create_all(engine)
Base.metadata.bind = engine
DBSession = sessionmaker(bind=engine)
session = DBSession()

malta1 = Country(name='Malta')
malta2 = Country(name='Malta')
session.add(malta1)
session.add(malta2)
session.commit()

但是提交进展顺利。如何检测是否违反了唯一约束？

python orm sqlalchemy
1个回答
0
投票

当我按照编写的方式运行你的代码时，它确实失败了：

    Traceback (most recent call last):
      File "so4.py", line 26, in <module>
        session.commit()
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 906, in commit
        self.transaction.commit()
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 461, in commit
        self._prepare_impl()
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 441, in _prepare_impl
        self.session.flush()
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2177, in flush
        self._flush(objects)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2297, in _flush
        transaction.rollback(_capture_exception=True)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py", line 66, in __exit__
        compat.reraise(exc_type, exc_value, exc_tb)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 187, in reraise
        raise value
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/session.py", line 2261, in _flush
        flush_context.execute()
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 389, in execute
        rec.execute(self)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/unitofwork.py", line 548, in execute
        uow
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 181, in save_obj
        mapper, table, insert)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/orm/persistence.py", line 835, in _emit_insert_statements
        execute(statement, params)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 945, in execute
        return meth(self, multiparams, params)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 263, in _execute_on_connection
        return connection._execute_clauseelement(self, multiparams, params)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1053, in _execute_clauseelement
        compiled_sql, distilled_params
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1189, in _execute_context
        context)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1402, in _handle_dbapi_exception
        exc_info
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 203, in raise_from_cause
        reraise(type(exception), exception, tb=exc_tb, cause=cause)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 186, in reraise
        raise value.with_traceback(tb)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1182, in _execute_context
        context)
      File "/Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 470, in do_execute
        cursor.execute(statement, parameters)
    sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: countries.name [SQL: 'INSERT INTO countries (name) VALUES (?)'] [parameters: ('Malta',)]

唯一约束是表的一部分，不一定是模型的一部分。当我在没有它的情况下首先运行数据库创建，然后运行您的代码时，我没有遇到任何问题。

对于更复杂的表，您可能需要使用像 alembic 这样的迁移工具，但是为此，您可以自己完成（您可以在一个脚本中完成，但我只做了两个以保持简单，抱歉如果那不是Pythonic：

    from sqlalchemy import Column, Integer, String
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    
    
    Base = declarative_base()
    
    
    class Country(Base):
        __tablename__ = 'countries'
        id = Column(Integer, primary_key=True, autoincrement=True)
        name = Column(String(64), unique=True, nullable=False, index=True)
    
    class TempCountry(Base):
        __tablename__ = 'temp_countries'
        id = Column(Integer, primary_key=True, autoincrement=True)
        name = Column(String(64), nullable=False, index=True)
    
    engine = create_engine('sqlite:///countries_example.db')
    Base.metadata.create_all(engine)
    Base.metadata.bind = engine
    DBSession = sessionmaker(bind=engine)
    session = DBSession()
    
    for country in session.query(Country).all():
        if len(session.query(TempCountry).filter_by(name=country.name).all()) == 0:
            temp_country = TempCountry(id=country.id, name=country.name)
            session.add(temp_country)
            session.commit()
    
    Country.__table__.drop(engine)
    session.commit()

随后

    from sqlalchemy import Column, Integer, String
    from sqlalchemy.ext.declarative import declarative_base
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    
    
    Base = declarative_base()
    
    
    class Country(Base):
        __tablename__ = 'countries'
        id = Column(Integer, primary_key=True, autoincrement=True)
        name = Column(String(64), unique=True, nullable=False, index=True)
    
    class TempCountry(Base):
        __tablename__ = 'temp_countries'
        id = Column(Integer, primary_key=True, autoincrement=True)
        name = Column(String(64), nullable=False, index=True)
    
    engine = create_engine('sqlite:///countries_example.db')
    Base.metadata.create_all(engine)
    Base.metadata.bind = engine
    DBSession = sessionmaker(bind=engine)
    session = DBSession()
    
    for temp_country in session.query(TempCountry).all():
        country = Country(id=country.id, name=country.name)
        session.add(country)
    
    session.commit()

    TempCountry.__table__.drop(engine)
    session.commit()

您不必传递 ID，但如果您出于某种原因需要保留它，则可以。另外，请记住在删除表之前进行提交，否则它将无法正常工作。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.