Блог

Главная - Вопросы по программированию - SQLAlchemy — Добавление ограничения ForeignKeyConstraint в таблицу «многие ко многим», основанную на другой связи

SQLAlchemy — Добавление ограничения ForeignKeyConstraint в таблицу «многие ко многим», основанную на другой связи

#python #sql #sqlalchemy

Вопрос:

Простите меня, если на этот вопрос был дан ответ в другом месте. Я искал ТАК и не смог перевести, казалось бы, соответствующие вопросы и ответы на мой сценарий.

Я работаю над интересным личным проектом, в котором у меня есть 4 основные схемы (на данный момент за исключением отношений).:

  • Личность (имя, биография)
  • Эпизод (название, сюжет)
  • Клип (URL, метка времени)
  • Изображение (url)

Ограничения (Основа взаимоотношений):

  1. Персонаж может отображаться в нескольких эпизодах, а также в нескольких клипах и изображениях из этих эпизодов (но может быть не во всех клипах/изображениях, связанных с эпизодом).
  2. Эпизод может содержать несколько персонажей, клипов и изображений.
  3. Изображение/Клип может быть связано только с одним эпизодом, но может быть связано с несколькими персонажами.
  4. Если Персонаж уже назначен эпизоду(эпизодам), то любой клип/изображение, назначенное персонажу, может быть только из одного из этих эпизодов или (если новый) должен быть способен связать с клипом/изображением только один из эпизодов, в которых появился персонаж.
  5. Если Эпизоду уже присвоен персонаж(ы), то любой клип/изображение, назначенные эпизоду, должны быть связаны с одним из этих персонажей или (если новый) должны быть способны иметь только одного или нескольких персонажей из эпизода, связанного с клипом/изображением.

Я разработал структуру базы данных следующим образом: Схема БД

Это создает следующий sql:

 DROP TABLE IF EXISTS episodes;
DROP TABLE IF EXISTS personas;
DROP TABLE IF EXISTS personas_episodes;
DROP TABLE IF EXISTS clips;
DROP TABLE IF EXISTS personas_clips;
DROP TABLE IF EXISTS images;
DROP TABLE IF EXISTS personas_images;


CREATE TABLE episodes (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(120) NOT NULL UNIQUE,
plot TEXT,
tmdb_id VARCHAR(10) NOT NULL,
tvdb_id VARCHAR(10) NOT NULL,
imdb_id VARCHAR(10) NOT NULL);

CREATE TABLE personas (
id INT NOT NULL PRIMARY KEY,
name VARCHAR(30) NOT NULL,
bio TEXT NOT NULL);

CREATE TABLE personas_episodes (
persona_id INT NOT NULL,
episode_id INT NOT NULL,
PRIMARY KEY (persona_id,episode_id),
FOREIGN KEY(persona_id) REFERENCES personas(id),
FOREIGN KEY(episode_id) REFERENCES episodes(id));

CREATE TABLE clips (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(100) NOT NULL,
timestamp VARCHAR(7) NOT NULL,
link VARCHAR(100) NOT NULL,
episode_id INT NOT NULL,
FOREIGN KEY(episode_id) REFERENCES episodes(id));

CREATE TABLE personas_clips (
clip_id INT NOT NULL,
persona_id INT NOT NULL,
PRIMARY KEY (clip_id,persona_id),
FOREIGN KEY(clip_id) REFERENCES clips(id),
FOREIGN KEY(persona_id) REFERENCES personas(id));

CREATE TABLE images (
id INT NOT NULL PRIMARY KEY,
link VARCHAR(120) NOT NULL UNIQUE,
path VARCHAR(120) NOT NULL UNIQUE,
episode_id INT NOT NULL,
FOREIGN KEY(episode_id) REFERENCES episodes(id));

CREATE TABLE personas_images (
persona_id INT NOT NULL,
image_id INT NOT NULL,
PRIMARY KEY (persona_id,image_id),
FOREIGN KEY(persona_id) REFERENCES personas(id),
FOREIGN KEY(image_id) REFERENCES images(id));

И я попытался создать ту же схему в моделях SQLAchemy (имея в виду SQLite для тестирования, PostgreSQL для производства) примерно так:

 # db is a configured Flask-SQLAlchemy instance
from app import db
# Alias common SQLAlchemy names
Column = db.Column
relationship = db.relationship


class PkModel(Model):
    """Base model class that adds a 'primary key' column named ``id``."""
 
    __abstract__ = True
    id = Column(db.Integer, primary_key=True)
 
 
def reference_col(
    tablename, nullable=False, pk_name="id", foreign_key_kwargs=None, column_kwargs=None
):
    """Column that adds primary key foreign key reference.
 
    Usage: ::
 
        category_id = reference_col('category')
        category = relationship('Category', backref='categories')
    """
    foreign_key_kwargs = foreign_key_kwargs or {}
    column_kwargs = column_kwargs or {}
 
    return Column(
        db.ForeignKey(f"{tablename}.{pk_name}", **foreign_key_kwargs),
        nullable=nullable,
        **column_kwargs,
    )

personas_episodes = db.Table(
    "personas_episodes",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("episode_id", db.ForeignKey("episodes.id"), primary_key=True),
)
 
personas_clips = db.Table(
    "personas_clips",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("clip_id", db.ForeignKey("clips.id"), primary_key=True),
)
 
personas_images = db.Table(
    "personas_images",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("image_id", db.ForeignKey("images.id"), primary_key=True),
)
 
 
class Persona(PkModel):
    """One of Roger's personas."""
 
    __tablename__ = "personas"
    name = Column(db.String(80), unique=True, nullable=False)
    bio = Column(db.Text)
    # relationships
    episodes = relationship("Episode", secondary=personas_episodes, back_populates="personas")
    clips = relationship("Clip", secondary=personas_clips, back_populates="personas")
    images = relationship("Image", secondary=personas_images, back_populates="personas")
 
    def __repr__(self):
        """Represent instance as a unique string."""
        return f"<Persona({self.name!r})>"
 
 
class Image(PkModel):
    """An image of one of Roger's personas from an episode of American Dad."""
    
    __tablename__ = "images"
    link = Column(db.String(120), unique=True)
    path = Column(db.String(120), unique=True)
    episode_id = reference_col("episodes")
    # relationships
    personas = relationship("Persona", secondary=personas_images, back_populates="images")
    
 
 
class Episode(PkModel):
    """An episode of American Dad."""
    
    # FIXME: We can add Clips and Images linked to Personas that are not assigned to this episode
 
    __tablename__ = "episodes"
    title = Column(db.String(120), unique=True, nullable=False)
    plot = Column(db.Text)
    tmdb_id = Column(db.String(10))
    tvdb_id = Column(db.String(10))
    imdb_id = Column(db.String(10))
    # relationships
    personas = relationship("Persona", secondary=personas_episodes, back_populates="episodes")
    images = relationship("Image", backref="episode")
    clips = relationship("Clip", backref="episode")
 
    def __repr__(self):
        """Represent instance as a unique string."""
        return f"<Episode({self.title!r})>"
 
 
class Clip(PkModel):
    """A clip from an episode of American Dad that contains one or more of Roger's personas."""
 
    __tablename__ = "clips"
    title = Column(db.String(80), unique=True, nullable=False)
    timestamp = Column(db.String(7), nullable=True)  # 00M:00S
    link = Column(db.String(7), nullable=True)
    episode_id = reference_col("episodes")
    # relationships
    personas = relationship("Persona", secondary=personas_clips, back_populates="clips")

Однако обратите внимание на FIXME комментарий. Мне трудно понять, как ограничить отношения «многие ко многим» в персонажах изображениях, персонажах клипах и персонажах эпизодах таким образом, чтобы все они смотрели друг на друга, прежде чем добавлять новую запись, чтобы ограничить возможные дополнения к подмножеству элементов, которые соответствуют критериям этих других отношений «многие ко многим».

Может ли кто-нибудь, пожалуйста, предоставить решение для обеспечения того, чтобы отношения «многие ко многим episode_id » уважали отношения в родительских таблицах?

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

 # omitting some detail fields for brevity
e1 = Episode(title="Some Episode")
e2 = Episode(title="Another Episode")
p1 = Persona(name="Raider Dave", episodes=[e1])
p2 = Persona(name="Ricky Spanish", episodes=[e2])
c1 = Clip(title="A clip", episode=e1, personas=[p2])  # should fail
i1 = Image(title="An image", episode=e2, personas=[p1]) # should fail
c2 = Clip(title="Another clip", episode=e1, personas=[p1])  # should succeed
i2 = Image(title="Another image", episode=e2, personas=[p2]) # should succeed

Ответ №1:

Добавить:

  • столбец, не подлежащий аннулированию episode_id ,
  • составная ссылка на внешний ключ personas_episode и
  • триггер для автозаполнения episode_id .

Столбец, не допускающий значения null, и составной внешний ключ достаточны для создания правильных ограничений на уровне базы данных, а также для обеспечения того, чтобы за пределами моделей SQLAlchemy можно было добавлять только правильные данные.

Триггер предлагается в результате отсутствия поддержки в моделях SQLAlchemy для перехвата before_insert события, на Table которое ссылаются relationship.secondary .

Реализация

SQLite не поддерживает изменение NEW.episode_id в BEFORE INSERT триггере, что означает, что мы должны автоматически заполнить AFTER INSERT триггер. Итак, мы разрешаем столбцу быть nullable и добавляем еще 2 триггера, чтобы проверить episode_id ограничение позже.

 episode_id_nullable = db.engine.dialect.name == "sqlite"                # Add this

personas_clips = db.Table(
    "personas_clips",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("episode_id", db.Integer, nullable=episode_id_nullable),  # Add this
    db.Column("clip_id", db.ForeignKey("clips.id"), primary_key=True),
    db.ForeignKeyConstraint(["persona_id", "episode_id"], ["personas_episodes.persona_id", "personas_episodes.episode_id"]),  # Add this
)

personas_images = db.Table(
    "personas_images",
    db.Column("persona_id", db.ForeignKey("personas.id"), primary_key=True),
    db.Column("episode_id", db.Integer, nullable=episode_id_nullable),  # Add this
    db.Column("image_id", db.ForeignKey("images.id"), primary_key=True),
    db.ForeignKeyConstraint(["persona_id", "episode_id"], ["personas_episodes.persona_id", "personas_episodes.episode_id"]),  # Add this
)

Триггеры SQLite:

  1. Перед вставкой убедитесь, что clip_id / image_id ссылается на clip / image в episode где persona находится (на основе persona_episodes ).
  2. Перед обновлением убедитесь, что значение episode_id не установлено NULL .
  3. После вставки автоматически заполните episode_id .
 SQLITE_CHECK_EPISODE_ID_BEFORE_INSERT = """
CREATE TRIGGER {table_name}_check_episode_id_before_insert BEFORE INSERT ON {table_name}
  FOR EACH ROW
  WHEN NEW.episode_id IS NULL
  BEGIN
    SELECT RAISE(ABORT, 'NOT NULL constraint failed: {table_name}.episode_id') WHERE NOT EXISTS (
        SELECT 1
        FROM {fk_target_table_name}
        JOIN personas_episodes ON {fk_target_table_name}.episode_id = personas_episodes.episode_id
        WHERE {fk_target_table_name}.{fk_target_name} = NEW.{fk_name}
          AND personas_episodes.persona_id = NEW.persona_id
    );
  END;
"""

SQLITE_CHECK_EPISODE_ID_BEFORE_UPDATE = """
CREATE TRIGGER {table_name}_check_episode_id_before_update BEFORE UPDATE ON {table_name}
  FOR EACH ROW
  WHEN NEW.episode_id IS NULL
  BEGIN
    SELECT RAISE(ABORT, 'NOT NULL constraint failed: {table_name}.episode_id');
  END;
"""

SQLITE_AUTOFILL_EPISODE_ID = """
CREATE TRIGGER {table_name}_autofill_episode_id AFTER INSERT ON {table_name}
  FOR EACH ROW
  WHEN NEW.episode_id IS NULL
  BEGIN
    UPDATE {table_name}
    SET episode_id = (SELECT {fk_target_table_name}.episode_id
        FROM {fk_target_table_name}
        JOIN personas_episodes ON {fk_target_table_name}.episode_id = personas_episodes.episode_id
        WHERE {fk_target_table_name}.{fk_target_name} = NEW.{fk_name}
          AND personas_episodes.persona_id = NEW.persona_id)
    WHERE {fk_name} = NEW.{fk_name}
      AND persona_id = NEW.persona_id;
  END;
"""

Триггер PostgreSQL:

  1. Перед вставкой выполните автозаполнение episode_id .
 POSTGRESQL_AUTOFILL_EPISODE_ID = """
CREATE OR REPLACE FUNCTION {table_name}_autofill_episode_id() RETURNS TRIGGER AS ${table_name}_autofill_episode_id$
  DECLARE
    _episode_id INT;
    in_episode BOOL;
  BEGIN
    IF NEW.episode_id IS NULL THEN
        SELECT episode_id INTO _episode_id FROM {fk_target_table_name} WHERE {fk_target_name} = NEW.{fk_name};
        SELECT TRUE INTO in_episode FROM personas_episodes WHERE persona_id = NEW.persona_id AND episode_id = _episode_id;
        IF in_episode IS NOT NULL THEN
            NEW.episode_id = _episode_id;
        END IF;
    END IF;
    RETURN NEW;
  END;
${table_name}_autofill_episode_id$ LANGUAGE plpgsql;

CREATE TRIGGER {table_name}_autofill_episode_id BEFORE INSERT OR UPDATE ON {table_name}
  FOR EACH ROW EXECUTE PROCEDURE {table_name}_autofill_episode_id();
"""

Добавление триггеров after_create в таблицы personas_clips и personas_images :

 from sqlalchemy import event, text


def after_create_trigger_autofill_episode_id(target, connection, **kw):
    fk = next(fk for fk in target.foreign_keys if "personas" not in fk.column.table.name)
    if connection.dialect.name == "sqlite":
        connection.execute(text(SQLITE_CHECK_EPISODE_ID_BEFORE_INSERT.format(table_name=target.name, fk_target_table_name=fk.column.table.name, fk_target_name=fk.column.name,fk_name=fk.parent.name)))
        connection.execute(text(SQLITE_CHECK_EPISODE_ID_BEFORE_UPDATE.format(table_name=target.name, fk_target_table_name=fk.column.table.name, fk_target_name=fk.column.name, fk_name=fk.parent.name)))
        connection.execute(text(SQLITE_AUTOFILL_EPISODE_ID.format(table_name=target.name, fk_target_table_name=fk.column.table.name, fk_target_name=fk.column.name, fk_name=fk.parent.name)))
    elif connection.dialect.name == "postgresql":
        connection.execute(text(POSTGRESQL_AUTOFILL_EPISODE_ID.format(table_name=target.name, fk_target_table_name=fk.column.table.name, fk_target_name=fk.column.name, fk_name=fk.parent.name)))


event.listen(personas_clips, "after_create", after_create_trigger_autofill_episode_id)
event.listen(personas_images, "after_create", after_create_trigger_autofill_episode_id)

Тестовые примеры

Вот что у меня есть на данный момент, основываясь на ожидаемом поведении в вопросе.

 from sqlalchemy.exc import IntegrityError
from sqlalchemy.sql import select

from models import *

if db.engine.dialect.name == "sqlite":
    db.session.execute("pragma foreign_keys=on")
else:
    db.session.execute("""
DROP TABLE IF EXISTS episodes CASCADE;
DROP TABLE IF EXISTS personas CASCADE;
DROP TABLE IF EXISTS personas_episodes CASCADE;
DROP TABLE IF EXISTS clips CASCADE;
DROP TABLE IF EXISTS personas_clips;
DROP TABLE IF EXISTS images CASCADE;
DROP TABLE IF EXISTS personas_images;
""")
    db.session.commit()

db.create_all()

e1 = Episode(title="Some Episode")
e2 = Episode(title="Another Episode")
db.session.add(e1)
db.session.add(e2)
db.session.commit()

p1 = Persona(name="Raider Dave", episodes=[e1])
p2 = Persona(name="Ricky Spanish", episodes=[e2])
db.session.add(p1)
db.session.add(p2)
db.session.commit()

c1 = Clip(title="A clip", episode=e1, personas=[p2])         # should fail
db.session.add(c1)
try:
    db.session.commit()
    assert False
except IntegrityError:
    db.session.rollback()
assert Clip.query.first() is None, list(db.session.execute(select(personas_clips)))

i1 = Image(link="An image", episode=e2, personas=[p1])       # should fail
db.session.add(i1)
try:
    db.session.commit()
    assert False
except IntegrityError:
    db.session.rollback()
assert Image.query.first() is None, list(db.session.execute(select(personas_images)))

c2 = Clip(title="Another clip", episode=e1, personas=[p1])   # should succeed
db.session.add(c2)
db.session.commit()
assert Clip.query.first() is not None

i2 = Image(link="Another image", episode=e2, personas=[p2])  # should succeed
db.session.add(i2)
db.session.commit()
assert Image.query.first() is not None

Альтернативы, которые не сработали

SQLAlchemy , похоже, не поддерживает before_insert только события для Table Model .
https://docs.sqlalchemy.org/en/14/orm/events.html#sqlalchemy.orm.MapperEvents.before_insert

Я попытался использовать прокси-сервер ассоциации, но не смог c2.personas.remove(p1) его чисто поддержать. https://docs.sqlalchemy.org/en/14/orm/extensions/associationproxy.html

Комментарии:

1. Это именно то, что мне было нужно, спасибо!

2. Извините за задержку с продолжением, но я просто копаюсь в этом сейчас. Если бы я преобразовал Table s в Model s и разрешил, чтобы clips.episode_id и images.episode_id были равны НУЛЮ, это позволило бы мне отказаться от 3 триггеров?

3. relationship.secondary разбирается с Table таким, судя по тому, что я пробовал, нет.

Ответ №2:

Я не могу придумать никакого способа добавить эту логику в базу данных. Было бы приемлемо управлять этими ограничениями в вашем коде? Подобный этому:

Событие: в БД будет создано новое изображение

 # create new image with id = 1 and linked episode = 1
my_image = Image(...) 

# my personas
Persona.query.all()
[<Persona('Homer')>, <Persona('Marge')>, <Persona('Pikachu')>]

# my episodes
>>> Episode.query.all()
[<Episode('the simpson')>]

#my images
>>> Image.query.all()
[<Image 1>, <Image 2>]

# personas in first image
>>> Image.query.all()[0].personas
[<Persona('Marge')>]

# which episode?
>>> Image.query.all()[0].episode
<Episode('the simpson')>

# same as above but with next personas (Note that Pikachu is linked to the wrong episode)
>>> Image.query.all()[1].personas
[<Persona('Pikachu')>]
>>> Image.query.all()[1].episode
<Episode('the simpson')>

# before saving the Image object check its episode id, then for that episode get a list of personas that appear.
# Look for your persona id of Image inside this list to see if this persona appear in the episode

my_image.episode.id # return 1

# get a list of persona(s).id that appear in the episode linked to the image!
personas_in_episode = [ persona.id for persona in Episode.query.filter_by(id=1).first().personas ] # return list of id of Persona objects [1,2] (Homer and Marge as episode with id 1 is The Simpson)

my_image_personas = [ persona.id for persona in my_image.personas ] # return a list of id of Persona objects linked in image [3] Persona is Pikachu

>>> my_image_personas
[3]

>>> for persona_id in my_image_personas:
...     if persona_id not in personas_in_episode:
...         print(f"persona with id {persona_id} does not appear in the episode for this image") 
... 
persona with id 3 does not appear in the episode for this image

То же самое касается клипа.

Комментарии:

1. Я ценю ваши усилия, но я считаю, что это, вероятно, возможно с использованием составных ключей и внешних ограничений. Я просто не могу понять, где (и как правильно) их реализовать.

Ответ №3:

Я думаю, что вам нужно изменить personas_images, добавить столбец episode_id, затем добавить составной внешний ключ в personas_episode на episode_id/personas_id и изменить внешний ключ изображения, чтобы он был составным на image_id/episode_id. Это гарантирует, что персонаж находится в эпизоде и что изображение находится в том же эпизоде.

Затем сделайте то же самое для клипов.

Комментарии:

1. Не могли бы вы предоставить обновленные модели и тест, который подтвердит, что это будет работать?

2. Просто следую вашему ответу. У вас была возможность это проверить? Похоже, что ваше решение может быть на правильном пути. Мне просто нужно увидеть модели sqlalchemy и простой тест с их использованием, чтобы показать, что все работает так, как ожидалось.

Ответ №4:

Я рассмотрел вас и обнаружил проблему во взаимоотношениях нижеперечисленных субъектов. После приведенных ниже изменений ваша схема будет завершена.

 CREATE TABLE episodes (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(120) NOT NULL UNIQUE,
plot TEXT,
tmdb_id VARCHAR(10) NOT NULL,
tvdb_id VARCHAR(10) NOT NULL,
imdb_id VARCHAR(10) NOT NULL
-- Need relationship as below
);

CREATE TABLE clips (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(100) NOT NULL,
timestamp VARCHAR(7) NOT NULL,
link VARCHAR(100) NOT NULL,
episode_id INT NOT NULL,
FOREIGN KEY(episode_id) REFERENCES episodes(id) -- No need as below
);

CREATE TABLE images (
id INT NOT NULL PRIMARY KEY,
link VARCHAR(120) NOT NULL UNIQUE,
path VARCHAR(120) NOT NULL UNIQUE,
episode_id INT NOT NULL,
FOREIGN KEY(episode_id) REFERENCES episodes(id)-- No need as below
);

Правильная связь должна быть такой, как показано ниже:

 CREATE TABLE episodes (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(120) NOT NULL UNIQUE,
plot TEXT,
tmdb_id VARCHAR(10) NOT NULL,
tvdb_id VARCHAR(10) NOT NULL,
imdb_id VARCHAR(10) NOT NULL,
clips_id VARCHAR(10) NOT NULL,
clips_id INT NOT NULL ,
images_id INT NOT NULL ,
FOREIGN KEY(clips_id) REFERENCES clips(id)
FOREIGN KEY(images_id) REFERENCES images(id)
);

CREATE TABLE clips (
id INT NOT NULL PRIMARY KEY,
title VARCHAR(100) NOT NULL,
timestamp VARCHAR(7) NOT NULL,
link VARCHAR(100) NOT NULL,
episode_id INT NOT NULL
);

CREATE TABLE images (
id INT NOT NULL PRIMARY KEY,
link VARCHAR(120) NOT NULL UNIQUE,
path VARCHAR(120) NOT NULL UNIQUE,
episode_id INT NOT NULL
);

Комментарии:

1. Это кажется неправильным… Не приведет ли это к тому, что мне придется дублировать эпизод для каждого связанного клипа/изображения? Пожалуйста, предоставьте полную схему с моделями sqlalchemy и пример, показывающий, что это работает должным образом.

2. Почему вам нужно сделать дубликат эпизода. один эпизод будет отображать несколько человек, а несколько эпизодов будут отображаться с одним человеком в «personas_episodes». Это будет обрабатывать все ваши сценарии.

3. Кроме того, вы можете добавить несколько эпизодов с различными комбинациями клипа/изображений.

4. Вы разместили clips_id images_id поля и на episodes столе. Это позволит добавить в эпизод только один клип и изображение. Это определенно неправильно. Кроме того, вам еще предстоит поделиться обновленными моделями с тестом, демонстрирующим, что он работает так, как вы говорите.

Метки: ,