为何带复合索引的数据库表仍可重复导入相同记录？

Let's break down why you're still seeing duplicate records even after adding that composite index, and how to fix it.

The Core Issue: Regular vs. Unique Indexes #

Your current mbs_composite index is just a regular composite index—its only job is to speed up queries that filter on those combined columns. It does not enforce uniqueness of the column values.

Databases only block duplicate inserts when you define an index as unique=True. Without that flag, the database doesn't check for duplicate combinations of those fields during insertion.

Fix 1: Turn Your Composite Index Into a Unique Index #

Modify your index definition to include the unique=True parameter. This tells the database to enforce that no two rows have the same combination of the indexed columns:

Index(
    'mbs_composite',
    "tour_id_mbs",
    "tournament_id_mbs",
    "round_id_mbs",
    "p1_id_mbs",
    "p2_id_mbs",
    unique=True  # Add this line to enforce uniqueness
),

After making this change, you'll need to apply the database migration (e.g., using Alembic if you're using SQLAlchemy with migrations) to create the unique index in your database. Once active, any attempt to insert a duplicate combination will throw an integrity error, preventing duplicates.

Fix 2: Check for Existing Records Before Insertion (Optional but Recommended) #

Even with a unique index, trying to insert duplicates will throw an error. To make your code more graceful, you can check if the record already exists before adding it to the session:

matches = dal.session.query(
    TodayATP.TOUR,
    TodayATP.ROUND,
    TodayATP.ID1,
    TodayATP.ID2,
)
for match in matches:
    match_dict = {
        "tour_id_mbs": 0,
        "tournament_id_mbs": match[0],
        "round_id_mbs": match[1],
        "p1_id_mbs": match[2],
        "p2_id_mbs": match[3],
    }
    # Check if the record already exists
    existing_match = dal.session.query(MatchesBlgScheduled).filter(
        MatchesBlgScheduled.tour_id_mbs == match_dict["tour_id_mbs"],
        MatchesBlgScheduled.tournament_id_mbs == match_dict["tournament_id_mbs"],
        MatchesBlgScheduled.round_id_mbs == match_dict["round_id_mbs"],
        MatchesBlgScheduled.p1_id_mbs == match_dict["p1_id_mbs"],
        MatchesBlgScheduled.p2_id_mbs == match_dict["p2_id_mbs"]
    ).first()
    if not existing_match:
        dal.session.add(MatchesBlgScheduled(**match_dict))
dal.session.commit()

A Quick Note on tour_id_mbs #

You're setting tour_id_mbs to a fixed value of 0 in every insert. Make sure this is intentional—since it's part of your composite unique index, it will be included in the uniqueness check. If you ever need to change this value later, it won't break existing uniqueness constraints as long as the full combination stays unique.

内容的提问来源于stack exchange，提问作者Jossy