"duplicate key" error when adding a new ORM object with related (child) objects

Question

I have an association object defined using SQLAlchemy to represent a many-to-many relationship between 2 tables. The reason I am using the association object pattern is because the association table contains extra columns. I have a unique constraint on the name column in the data_type table. When I try to insert data into source_key, and create the relationships, it results in the error below. My question is, how would I say, "Get the ID if it exists and add to association table; otherwise, create the record in data_type, then add to association table"?

error

the-librarian-backend-1       | sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "ix_data_type_name"
the-librarian-backend-1       | DETAIL:  Key (name)=(str) already exists.
the-librarian-backend-1       |
the-librarian-backend-1       | [SQL: INSERT INTO data_type (name) VALUES (%(name)s) RETURNING data_type.id]
the-librarian-backend-1       | [parameters: ({'name': 'str'}, {'name': 'str'}, {'name': 'str'}, {'name': 'str'}, {'name': 'str'}, {'name': 'date'}, {'name': 'list'}, {'name': 'int'}  ... displaying 10 of 747 total bound parameter sets ...  {'name': 'date'}, {'name': 'str'})]

models

# source_key.py
class SourceKey(Base):
    __tablename__ = 'source_key'
    id = Column(Integer, primary_key=True, index=True)
    source_id = Column(Integer, ForeignKey('source.id'), nullable=False)
    key_id = Column(Integer, ForeignKey('key.id'), nullable=False)
    description = Column(Text)
    data_types = relationship("SourceKeyDataType", back_populates="source_keys")

# data_type.py
class DataType(Base):
    __tablename__ = 'data_type'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(Text, index=True, nullable=False, unique=True)
    source_keys = relationship("SourceKeyDataType", back_populates="data_types")

# Association Object
class SourceKeyDataType(Base):
    __tablename__ = 'source_key_data_type_assoc'
    source_key_id = Column(ForeignKey('source_key.id'), primary_key=True)
    data_type_id = Column(ForeignKey('data_type.id'), primary_key=True)
    count = Column(BigInteger)
    source_keys = relationship("SourceKey", back_populates="data_types")
    data_types = relationship("DataType", back_populates="source_keys")

code

source_keys = [
    {
      "key": {
        "name": "total"
      },
      "description": "the total cost of all items",
      "data_types": [
        {
          "name": "str",
          "count": 1904165
        }
      ]
    },
    {
      "key": {
        "name": "item_value"
      },
      "description": "the cost of a single item",
      "data_types": [
        {
          "name": "str",
          "count": 2079817
        }
      ]
    }
]

for source_key in source_keys:
    source_key_obj = {k: v for k, v in item.items() if isinstance(v, (str, int, bool, float))}
    source_key_db_obj = SourceKey(**source_key_obj)
    for dt in source_key.get("data_types") or []:
        a = SourceKeyDataType(is_inferred=item.get("is_inferred", False), count=item.get("count", 0))
        a.data_types = models.DataType(name=item["name"])
        source_key_db_obj.data_types.append(a)
    db.add(source_key_db_obj)
    db.commit()
    db.refresh(source_key_db_obj)

Answer 1

My question is, how would I say, "Get the ID if it exists and add to association table; otherwise, create the record in data_type, then add to association table"?

Your code needs to do exactly that. Let's look at a simplified example that uses an association table instead of an association object. To set up the test:

from sqlalchemy import Column, create_engine, ForeignKey, Integer, select, String, Table
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import declarative_base, relationship, Session

engine = create_engine("sqlite://")
Base = declarative_base()

post_tag = Table(
    "post_tag",
    Base.metadata,
    Column("post_id", Integer, ForeignKey("post.id"), primary_key=True),
    Column("tag_id", Integer, ForeignKey("tag.id"), primary_key=True),
)

class Post(Base):
    __tablename__ = "post"
    id = Column(Integer, primary_key=True)
    title = Column(String)
    tags = relationship("Tag", secondary=post_tag)

class Tag(Base):
    __tablename__ = "tag"
    id = Column(Integer, primary_key=True)
    name = Column(String, unique=True)

Base.metadata.create_all(engine)

# add test data into empty tables
with Session(engine) as sess:
    sess.add(
        Post(
            title="getting unique constraint violation",
            tags=[Tag(name="SQLAlchemy")],
        )
    )
    sess.commit()

First let's try adding a new Post the simplistic way:

# 1st try: adding a new post and blindly creating a new Tag object
with Session(engine) as sess:
    sess.add(
        Post(
            title="some other issue",
            tags=[Tag(name="SQLAlchemy")],
        )
    )
    try:
        sess.commit()
    except IntegrityError:
        print("1st try: An IntegrityError has occurred")
        # this error gets printed

Now let's check to see if the Tag already exists:

# 2nd try: check for existing tag first
with Session(engine) as sess:
    sqla_tag = sess.scalars(
        select(Tag).where(Tag.name == "SQLAlchemy")
    ).first()
    if not sqla_tag:
        # Tag object does not already exist, so create it
        sqla_tag = Tag(name="SQLAlchemy")
    sess.add(
        Post(
            title="some other issue",
            tags=[sqla_tag],
        )
    )
    sess.commit()
    print("2nd try: Success.")
    # no error this time

That's the most straightforward solution. A more advanced technique is to use an association proxy to automate the process, but some users find them difficult to work with.