Добавление данных в базу через Flask-SQLAlchemy

Добавление данных

Чтобы добавить новую запись с помощью SQLAlchemy, следуйте следующим шагам:

1. Создайте экземпляр объекта
2. Добавьте его в текущую сессию
3. Завершите процесс коммитом сессии

Работа с базой данных в SQLAlchemy осуществляется с использованием сессий. В Flask-SQLAlchemy это автоматизировано, поэтому сессия доступна через db.session, управляющий подключением и транзакциями. Транзакция открывается и действует до выполнения коммита или отката.

Давайте откроем Python shell и создадим несколько объектов модели:

(env) gvido@vm:~/flask_app$ python main2.py shell
>>>
>>> from main2 import db, Post, Tag, Category
>>>
>>> c1 = Category(name='Python', slug='python')
>>> c2 = Category(name='Java', slug='java')
>>> 

Мы создали два объекта класса Category. Доступ к их полям возможен через точечную нотацию (.):

>>> c1.name, c1.slug
('Python', 'python')
>>> c2.name, c2.slug
('Java', 'java')
>>> 

Перейдите к добавлению объектов в сессию:

>>> db.session.add(c1)
>>> db.session.add(c2)
>>> 

На данном этапе добавленные объекты еще не сохранены в базе данных. Проверьте их первичные ключи:

>>> print(c1.id)
None
>>> print(c2.id)
None
>>> 

Идентификатор id обоих объектов равен None, что свидетельствует о том, что они не добавлены в базу данных.

Можно добавить множество объектов одновременно, используя метод add_all():

>>> db.session.add_all([c1, c2])
>>> 

Добавление объекта несколько раз не вызовет ошибки. Все объекты в сессии видны через db.session.new.

>>> db.session.new
IdentitySet([, ])
>>> 

Чтобы сохранить объекты, выполните коммит с помощью commit():

>>> db.session.commit()
>>> 

Проверив id объекта Category, вы убедитесь, что ему присвоен первичный ключ:

>>> print(c1.id)
1
>>> print(c2.id)
2
>>> 

Теперь таблица categories в HeidiSQL будет выглядеть следующим образом:

У созданных категорий нет связывающихся постов, поэтому c1.posts и c2.posts вернут пустые списки:

>>> c1.posts
[]
>>> c2.posts
[]
>>> 

Давайте создадим несколько постов:

>>> p1 = Post(title='Post 1', slug='post-1', content='Post 1', category=c1)
>>> p2 = Post(title='Post 2', slug='post-2', content='Post 2', category=c1)
>>> p3 = Post(title='Post 3', slug='post-3', content='Post 3', category=c2)
>>> 

Категорию можно указать позже, как показано ниже:

>>> p1.category = c1
>>> 

Добавьте посты в сессию и выполните коммит:

>>> db.session.add_all([p1, p2, p3])
>>> db.session.commit()
>>> 

После этого атрибут posts у объекта Category будет возвращать список привязанных объектов:

>>> c1.posts
[<1:Post 1>, <2:Post 2>]
>>> c2.posts
[<3:Post 3>]
>>> 

Также можно получить категорию из объекта Post через атрибут category:

>>> p1.category
<1:Python>
>>> p2.category
<1:Python>
>>> p3.category
<2:Java>
>>> 

Это возможно благодаря вызову relationship() в модели Category. Теперь в базе данных есть три поста без тегов.

>>> p1.tags, p2.tags, p3.tags
([], [], [])
>>> 

Создадим несколько тегов:

>>> t1 = Tag(name="refactoring", slug="refactoring")
>>> t2 = Tag(name="snippet", slug="snippet")
>>> t3 = Tag(name="analytics", slug="analytics")
>>> db.session.add_all([t1, t2, t3])
>>> db.session.commit()
>>> 

Эти команды создают три объекта Tag, которые мы затем сохраните в базе. Свяжем теги с постами:

>>> p1.tags.append(t1)
>>> p1.tags.extend([t2, t3])
>>> p2.tags.append(t2)
>>> p3.tags.append(t3)
>>> db.session.add_all([p1, p2, p3])
>>> db.session.commit()
>>> 

Эти действия добавят пять записей в таблицу post_tags.

Теперь посты содержат привязанные теги:

>>> p1.tags
[<1:refactoring>, <2:snippet>, <3:analytics>]
>>> p2.tags
[<2:snippet>]
>>> p3.tags
[<3:analytics>]
>>> 

Извлечение постов по тегам теперь становится простым через атрибут posts тега:

>>> t1.posts
[<1:Post 1>]
>>> t2.posts
[<1:Post 1>, <2:Post 2>]
>>> t3.posts
[<1:Post 1>, <3:Post 3>]
>>> 

Интересно, что вместо сохранения объектов Tag сначала, а затем привязки их к Post, вы можете сделать это одновременно:

 t1 = Tag(name="refactoring", slug="refactoring")
 t2 = Tag(name="snippet", slug="snippet")
 t3 = Tag(name="analytics", slug="analytics")

 p1.tags.append(t1)
 p1.tags.extend([t2, t3])
 p2.tags.append(t2)
 p3.tags.append(t3)

 db.session.add(p1)
 db.session.add(p2)
 db.session.add(p3)

 db.session.commit()

Здесь важно, что строки 11-13 добавляют в сессию только объекты Post. Поскольку между Tag и Post существует отношение "многие-к-многим", добавление Post автоматически добавит связанные Tag. Если вы хотите, можно добавить Tag вручную, это не вызовет ошибок.

Обновление данных

Чтобы изменить объект, достаточно обновить его атрибуты, добавить объект в сессию и выполнить коммит.

 p1.content # текущее значение
'Post 1'

 p1.content = "This is content for post 1"  # задаем новое значение
 db.session.add(p1)

 db.session.commit()

 p1.content  # обновленное значение
'This is content for post 1'

Удаление данных

Чтобы удалить объект, используйте метод delete() в сессии. Он пометит объект как удаляемый при следующем коммите.

Давайте создадим временный тег seo, который привяжем к p1 и p2:

 tmp = Tag(name='seo', slug='seo')  # создаем временный объект Tag

 p1.tags.append(tmp)
 p2.tags.append(tmp)

 db.session.add_all([p1, p2])
 db.session.commit()

Этот коммит добавит три записи: одну в table и две в post_tags. В базе они представлены так:

Теперь удалите тег seo:

 db.session.delete(tmp)
 db.session.commit()

Последующий коммит удалит все три добавленные ранее записи. Однако это не затрагивает посты, с которыми связан тег.

Удаление объекта в родительской таблице, как правило, приводит к установке связанного внешнего ключа в дочерней таблице в NULL. В следующем примере создадим категорию и пост, чтобы затем удалить категорию:

 c4 = Category(name='css', slug='css')
 p4 = Post(title='Post 4', slug='post-4', content='Post 4', category=c4)

 db.session.add(c4)

 db.session.new
IdentitySet([<None:css>, <None:Post 4>])

 db.session.commit()

Этот коммит добавляет по одной записи в categories и в posts.

Давайте теперь посмотрим, что произойдет при удалении категории:

 db.session.delete(c4)
 db.session.commit()

Устранение категории css из categories приводит к занулению category_id в связанных записях posts.

В некоторых случаях требуется автоматическое удаление связанных записей при удалении родительских. Это обеспечивается, используя параметр cascade=’all,delete-orphan’ в db.relationship(). Изменив main2.py, добавьте эту инструкцию в модель Category:

#...
class Category(db.Model):
    #...
    posts = db.relationship('Post', backref='category', cascade='all,delete-orphan')
#...

Удаление категории теперь автоматически удаляет все её посты. После внесенных изменений вам нужно перезапустить оболочку, импортировать компоненты и создать новую категорию с постом:

(env) gvido@vm:~/flask_app$ python main2.py shell

 from main2 import db, Post, Tag, Category

  c5 = Category(name='css', slug='css')
  p5 = Post(title='Post 5', slug='post-5', content='Post 5', category=c5)

 db.session.add(c5)
 db.session.commit()

Теперь проверьте, как изменились данные в базе после коммита.

Теперь удалите категорию.

 db.session.delete(c5)
 db.session.commit()

После окончательного коммита база данных будет выглядеть следующим образом:

Запрос данных

Для работы с базой данных применяется метод query() из объекта session. Этот метод возвращает экземпляр flask_sqlalchemy.BaseQuery, который улучшает возможности sqlalchemy.orm.query.Query. По сути, он представляет собой SQL команду SELECT, используемую для запросов. Ниже указаны ключевые методы класса flask_sqlalchemy.BaseQuery.

Метод all()

Метод query() может принимать модельные классы или колонки как параметры. В примере ниже возвращаются все записи из таблицы posts.

>>>
>>> db.session.query(Post).all()
[<1:Post 1>, <2:Post 2>, <3:Post 3>, <4:Post 4>]
>>>

Такой подход также возвращает все записи из таблиц categories и tags.

>>>
>>> db.session.query(Category).all()
[<1:Python>, <2:Java>]
>>>
>>>
>>> db.session.query(Tag).all()
[<1:refactoring>, <2:snippet>, <3:analytics>]
>>>

Чтобы увидеть, какой SQL создается для выполнения запроса, можно вывести объект flask_sqlalchemy.BaseQuery:

>>>
>>> print(db.session.query(Post))
SELECT
    posts.id AS posts_id,
    posts.title AS posts_title,
    posts.slug AS posts_slug,
    posts.content AS posts_content,
    posts.created_on AS posts_created_on,
    posts.updated_on AS posts_updated_on,
    posts.category_id AS posts_category_id
FROM
    posts
>>>

В вышеописанных примерах были отобраны все данные из таблицы. Но если нужны лишь отдельные колонки, их следует указать в query():

>>>
>>> db.session.query(Post.id, Post.title).all()
[(1, 'Post 1'), (2, 'Post 2'), (3, 'Post 3'), (4, 'Post 4')]
>>>

Метод count()

Метод count() позволяет посчитать количество результатов, полученных в результате запроса.

>>>
>>> db.session.query(Post).count()  # общее количество записей в таблице Post
4
>>> db.session.query(Category).count()  # общее количество записей в таблице Category
2
>>> db.session.query(Tag).count()  # общее количество записей в таблице Tag
3
>>>

Метод first()

Метод first() позволяет получить первый результат или возвращает None, если данных нет.

>>>
>>> db.session.query(Post).first()
<1:Post 1>
>>>
>>> db.session.query(Category).first()
<1:Python>
>>>
>>> db.session.query(Tag).first()
<1:refactoring>
>>>

Метод get()

С помощью метода get() можно найти объект по его первичному ключу или получить None, если объект не найден.

>>>
>>> db.session.query(Post).get(2)
<2:Post 2>
>>>
>>> db.session.query(Category).get(1)
<1:Python>
>>>
>>> print(db.session.query(Category).get(10))  # объект с ключом 10 не найден
None
>>>

Метод get_or_404()

Аналогично get(), но вместо возвращения None, генерирует ошибку 404, если объект отсутствует.

>>>
>>> db.session.query(Post).get_or_404(1)
<1:Post 1>
>>>
>>>
>>> db.session.query(Post).get_or_404(100)
Traceback (most recent call last):
...
werkzeug.exceptions.NotFound: 404 Not Found: The requested URL was not found on the server. If you entered the URL manually, please check your spelling and try again.
>>>

Метод filter()

Метод filter() добавляет условия WHERE в запрос, позволяя фильтровать данные по заданным критериям. Пример использования:

>>>
>>> db.session.query(Post).filter(Post.title == 'Post 1').all()
[<1:Post 1>]
>>>

Этот запрос извлекает все посты с заголовком "Post 1". Соответствующий SQL выглядит так:

>>>
>>> print(db.session.query(Post).filter(Post.title == 'Post 1'))
SELECT
    posts.id AS posts_id,
    posts.title AS posts_title,
    posts.slug AS posts_slug,
    posts.content AS posts_content,
    posts.created_on AS posts_created_on,
    posts.updated_on AS posts_updated_on,
    posts.category_id AS posts_category_id
FROM
    posts
WHERE
    posts.title = % (title_1) s
>>>
>>>

Заполнитель % (title_1) s в условии WHERE заменяется фактическим значением при выполнении запроса.

Можно также комбинировать несколько условий, соединяя их с помощью оператора AND. Пример:

>>>
>>> db.session.query(Post).filter(Post.id >= 1, Post.id <= 2).all()
[<1:Post 1>, <2:Post 2>]
>>>
>>>

Этот пример возвращает посты с идентификатором от 1 до 2 включительно. Соответствующий SQL-запрос:

>>>
>>> print(db.session.query(Post).filter(Post.id >= 1, Post.id <= 2))
SELECT
    posts.id AS posts_id,
    posts.title AS posts_title,
    posts.slug AS posts_slug,
    posts.content AS posts_content,
    posts.created_on AS posts_created_on,
    posts.updated_on AS posts_updated_on,
    posts.category_id AS posts_category_id
FROM
    posts
WHERE
    posts.id >= % (id_1) s
AND posts.id <= % (id_2) s
>>>

Метод first_or_404()

Метод схожий с first(), однако вместо возвращения None генерирует ошибку 404 в случае отсутствия результатов.

>>>
>>> db.session.query(Post).filter(Post.id > 1).first_or_404()
<2:Post 2>
>>>
>>> db.session.query(Post).filter(Post.id > 10).first_or_404().all()
Traceback (most recent call last):
...
werkzeug.exceptions.NotFound: 404 Not Found: The requested URL was not found on the server. If you entered the URL manually, please check your spelling and try again.
>>>

Метод limit()

Метод limit() позволяет ограничить количество возвращаемых записей в запросе с помощью SQL оператора LIMIT.

>>>
>>> db.session.query(Post).limit(2).all()
[<1:Post 1>, <2:Post 2>]
>>>
>>> db.session.query(Post).filter(Post.id >= 2).limit(1).all()
[<2:Post 2>]
>>>

SQL-эквивалент:

>>>
>>> print(db.session.query(Post).limit(2))
SELECT
posts.id AS posts_id,
posts.title AS posts_title,
posts.slug AS posts_slug,
posts.content AS posts_content,
posts.created_on AS posts_created_on,
posts.updated_on AS posts_updated_on,
posts.category_id AS posts_category_id
FROM
    posts
LIMIT % (param_1) s
>>>
>>>
>>> print(db.session.query(Post).filter(Post.id >= 2).limit(1))
SELECT
    posts.id AS posts_id,
    posts.title AS posts_title,
    posts.slug AS posts_slug,
    posts.content AS posts_content,
    posts.created_on AS posts_created_on,
    posts.updated_on AS posts_updated_on,
    posts.category_id AS posts_category_id
FROM
    posts
WHERE
    posts.id >= % (id_1) s
LIMIT % (param_1) s
>>>
>>>

Метод offset()

Функция offset() позволяет добавить в SQL-запрос условие OFFSET. Эта функция принимает число, которое определяет сдвиг от начала набора данных. Как правило, используется совместно с limit() для разбиения данных на страницы.

>>>
>>> db.session.query(Post).filter(Post.id > 1).limit(3).offset(1).all()
[<3:Post 3>, <4:Post 4>]
>>>

SQL аналог:

>>>
>>> print(db.session.query(Post).filter(Post.id > 1).limit(3).offset(1))
SELECT
    posts.id AS posts_id,
    posts.title AS posts_title,
    posts.slug AS posts_slug,
    posts.content AS posts_content,
    posts.created_on AS posts_created_on,
    posts.updated_on AS posts_updated_on,
    posts.category_id AS posts_category_id
FROM
    posts
WHERE
    posts.id > % (id_1) s
LIMIT % (param_1) s, % (param_2) s
>>>

Заполнители % (param_1) s и % (param_2) указывают на параметры для ограничения (лимита) и сдвига (оффсета) выборки данных соответственно.

Метод order_by()

Метод order_by() управляет последовательностью вывода данных, применяя SQL-команду ORDER BY. Он принимает в качестве аргумента список колонок, используемых для упорядочивания. Если порядок не указан, по умолчанию сортировка производится по возрастанию.

>>>
>>> db.session.query(Tag).all()
[<1:refactoring>, <2:snippet>, <3:analytics>]
>>>
>>> db.session.query(Tag).order_by(Tag.name).all()
[<3:analytics>, <1:refactoring>, <2:snippet>]
>>>

Для сортировки по убыванию применяется db.desc():

>>>
>>> db.session.query(Tag).order_by(db.desc(Tag.name)).all()
[<2:snippet>, <1:refactoring>, <3:analytics>]
>>>

Метод join()

С помощью метода join() можно выполнять SQL JOIN оператор, чтобы объединить несколько таблиц на основе заданных условий. Имя таблицы, с которой нужно создать JOIN, передается как аргумент в метод.

>>>
>>> db.session.query(Post).join(Category).all()
[<1:Post 1>, <2:Post 2>, <3:Post 3>]
>>>

SQL аналог:

>>>
>>> print(db.session.query(Post).join(Category))
SELECT
    posts.id  AS posts_id,
    posts.title AS posts_title,
    posts.slug AS posts_slug,
    posts.content AS posts_content,
    posts.created_on AS posts_created_on,
    posts.updated_on AS posts_updated_on,
    posts.category_id AS posts_category_id
FROM
    posts

Функция join() часто применяется для получения данных из нескольких таблиц за один SQL-запрос. Например:

>>>
>>> db.session.query(Post.title,  Category.name).join(Category).all()
[('Post 1', 'Python'),  ('Post 2', 'Python'), ('Post 3', 'Java')]
>>>

Череда методов join() позволяет организовать связи между более чем двумя таблицами:

db.session.query(Table1).join(Table2).join(Table3).join(Table4).all()

Как завершающий этап, вы можете завершить создание формы для обратной связи.

Напоминаем, что ранее в уроке «Работа с формами во Flask» была реализована контактная форма для сбора отзывов от пользователей. На данный момент функция contact() выводит принятые данные только в консоль. Чтобы сохранить их, потребуется создать новую таблицу. В main2.py добавьте модель Feedback после определения модели Tag:

#...
class Feedback(db.Model):
    __tablename__ = 'feedbacks'
    id = db.Column(db.Integer), primary_key=True)
    name = db.Column(db.String(1000), nullable=False)
    email = db.Column(db.String(100), nullable=False)
    message = db.Column(db.Text(), nullable=False)
    created_on = db.Column(db.DateTime(), default=datetime.utcnow)

    def __repr__(self):
	return "<{}:{}>".format(self.id, self.name)
#...

Теперь следует запустить интерактивную оболочку Python и вызвать метод create_all() объекта db для создания в базе структуры новой таблицы feedbacks:

(env) gvido@vm:~/flask_app$ python main2.py shell
>>>
>>> from main2 import db
>>>
>>> db.create_all()
>>>

Также следует обновить функцию contact(), чтобы данные сохранялись в базу:

#...
@app.route('/contact/', methods=['get', 'post'])
def contact():
    form = ContactForm()
    if form.validate_on_submit():
	name = form.name.data
	email = form.email.data
	message = form.message.data
	print(name)
	print(Post)
	print(email)
	print(message)

	# логика работы с базой данных
	feedback = Feedback(name=name, email=email, message=message)
	db.session.add(feedback)
	db.session.commit()

	print("\nData received. Now redirecting ...")
	flash("Message Received", "success")
	return redirect(url_for('contact'))

    return render_template('contact.html', form=form)
#...

Запустите сервер и перейдите по адресу http://127.0.0.1:5000/contact/, чтобы протестировать отправку формы.

Запись, созданная в базе данных через HeidiSQL, будет выглядеть следующим образом: