Три примера работы с SQL базой данных в Python (sqlalchemy.orm) @ 900913

В данной заметке будут предложены 3 примера работы с БД из Python:

Работа через адаптер sqlite3.
Взаимодействие через sqlalchemy.Table.
Использование sqlalchemy.orm для тех же задач.
Бонус - PonyORM!

Хочется сконцентрироваться на различиях и преимуществах, которые дают те или иные подходы.

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

Взаимодействие через sqlalchemy.orm

Мы уже имеем реализацию на sqlite3 и sqlalchemy.Table. Оба подхода имели ориентацию на SQL синтаксис. В первом варианте мы использовали сырые SQL-запросы, во втором - за нас генерировали их по схожему по сути Python коду.

Теперь же мы несколько отдалимся от абстракций базы данных и перейдём больше к объектно-ориентированному программированию. Для начала - описание моделей:

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

    engine = create_engine('sqlite:///example.db', echo=True)
    Base = declarative_base()
    Session = sessionmaker(bind=engine)


    class PhotoModel(Base):
        __tablename__ = 'photo'
        id = Column(Integer, primary_key=True)
        url = Column(String)

        def __init__(self, url):
            self.url = url

        def __repr__(self):
            return f'<{self.__class__.__name__} #{self.id}>'


    class TagModel(Base):
        __tablename__ = 'tag'
        id = Column(Integer, primary_key=True)
        name = Column(String)

        def __init__(self, name):
            self.name = name

        def __repr__(self):
            return f'<{self.__class__.__name__} #{self.id}>'

        def get_photos(self) -> Query:
            with Session() as session:
                return (
                    session.query(PhotoModel, TagModel, PhotoTagModel)
                    .with_entities(PhotoModel)
                    .filter(PhotoModel.id == PhotoTagModel.photo_id)
                    .filter(TagModel.id == PhotoTagModel.tag_id)
                    .filter(TagModel.id == self.id)
                )


    class PhotoTagModel(Base):
        __tablename__ = 'photo_tag'
        id = Column(Integer, primary_key=True)
        photo_id = Column(Integer, ForeignKey('photo.id'))
        tag_id = Column(Integer, ForeignKey('tag.id'))

        def __init__(self, photo, tag):
            self.photo_id = photo.id
            self.tag_id = tag.id

        def __repr__(self):
            return f'<PhotoTagModel photo#{self.photo_id} tag#{self.tag_id}>'


    Base.metadata.create_all(engine)

Как можно увидеть, мы указываем название таблицы, поля таблицы и - это то, что относится непосредственно к базе данных. Также мы можем добавить свои поля, а что важнее - методы.

К примеру, видим, что мы хотим получать фотографии по тегу - значит очень неплохо бы расположить данное действие (метод) внутри класса, отвечающего за TagModel.

Также стандартно - __repr__ для вывода в общем виде (для отладки) информации об объекте, __init__ для простенькой инициализации при создании записи в БД. Также могут понадобиться различные @classmethodы для более удобного построения частых запросов к конкретной модели.

В общем, можем держать методы для работы с данной моделью в ней же. Даёшь "жирные модели" и "тонкие тупые контроллеры"!

Теперь о том, как же пользоваться всем этим добром. Для начала заведём данные, которыми хотим манипулировать:

    def fill_data():
        with Session() as session:
            photo = PhotoModel('https://900913.ru/static/img/logo-192x192.png')
            session.add(photo)
            tag1 = TagModel(name='python')
            session.add(tag1)
            tag2 = TagModel(name='website')
            session.add(tag2)
            session.commit()
            session.add(PhotoTagModel(photo, tag1))
            session.commit()

Всё делается через сессию - читай "транзакцию". В данном случае используем сессию в виде контекстного менеджера, чтобы в случае какого-либо исключения совершить session.rollback() автоматически - чтобы неверные данные не попали в базу данных.

Если же в какой-то момент хотим зафиксировать изменения как в данном примере - для того, чтобы теги и фото получили свои id, которые они будут использовать для создания связи между ними - делаем session.commit().

Ну и запуск всего этого добра с поиском по тегу фотографий через метод Tag.get_photos():

    def main():
        fill_data()

        with Session() as session:
            tag = session.query(TagModel).filter(TagModel.name == 'python').first()

            for photo in tag.get_photos().all():
                print(photo.id, photo.url)


    if __name__ == '__main__':
        main()

Ну и "на сдачу" - обновление и удаление данных:

    with Session() as session:
        tag = session.query(TagModel).get(1)
        tag.name = 'new-tag-name'
        session.commit()

    with Session() as session:
        session.query(TagModel).filter(TagModel.name == 'new-tag-name').delete()
        session.commit()

В первом случае мы обновляем конкретный объект, получив его предварительно по первичному ключу (primary_key).

Во втором случае - выполняем удаление по запросу session.query. Не забываем зафиксировать изменения.

Использование одной и той же таблицы в одном sqlalchemy.orm запросе несколько раз (aliased)

Допустим, у нас есть задача: получить фотографии на которых есть и первый тег и второй одновременно. Тогда нам потребуется написать запрос с двумя многие-ко-многим JOIN-ами. То есть нечто следующее:

        query = (
            session.query(PhotoModel, TagModel, PhotoTagModel)
            .with_entities(PhotoModel.url)
            .join(PhotoTagModel, PhotoModel.id == PhotoTagModel.photo_id)
            .join(TagModel, TagModel.id == PhotoTagModel.tag_id)
            .join(PhotoTagModel, PhotoModel.id == PhotoTagModel.photo_id)
            .join(TagModel, TagModel.id == PhotoTagModel.tag_id)
            .filter(TagModel.name == 'python')
            .filter(TagModel.name == 'sql')
        )

Тогда sqlalchemy не сможет понять: где какое поле из какого представления таблицы мы используем. В итоге мы получим нечто следующее:

...
sqlite3.OperationalError: ambiguous column name: photo_tag.photo_id

The above exception was the direct cause of the following exception:
...
sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) ambiguous column name: photo_tag.photo_id

Для решения данной проблемы можно использовать алиасы наших моделей. Мы назовём разные представления (тут речь не о VIEW, сразу скажу) одних и тех же моделей разными переменными, и sqlalchemy тогда сможет правильно понять нашу задумку:

    pt1 = aliased(PhotoTagModel)
    pt2 = aliased(PhotoTagModel)
    t1 = aliased(TagModel)
    t2 = aliased(TagModel)

    query = (
        session.query(PhotoModel, TagModel, PhotoTagModel)
        .with_entities(PhotoModel.url)
        .join(pt1, PhotoModel.id == pt1.photo_id)
        .join(t1, t1.id == pt1.tag_id)
        .join(pt2, PhotoModel.id == pt2.photo_id)
        .join(t2, t2.id == pt2.tag_id)
        .filter(t1.name == 'python')
        .filter(t2.name == 'sql')
    )

    for url in query.all():
        print(url)

В целом, всё то же взаимодействие с базой данных, но более в стиле ООП.

Работа через адаптер sqlite3.
Взаимодействие через sqlalchemy.Table.
Использование sqlalchemy.orm для тех же задач.
Бонус - PonyORM!