✅ Ответы на вопросы
Вопрос 1. Сопоставьте методы
Верный ответ: 1-c, 2-b, 3-d, 4-a.
pymysql.connect(...)— подключает Python к базе данных.connection.cursor()— создаёт объект курсора для отправки SQL-запросов.cursor.execute(...)— выполняет SQL-запрос.connection.close()— закрывает соединение с базой данных.
Вопрос 2. Порядок действий
Верный ответ: b → a → c → e → d.
- Подключиться к базе (
pymysql.connect). - Создать курсор (
connection.cursor()). - Выполнить запрос (
cursor.execute(...)). - Закрыть курсор (
cursor.close()). - Закрыть соединение (
connection.close()).
При использовании with шаги закрытия выполняются автоматически.
Вопрос 3. Методы получения данных
Верный ответ: 1-b, 2-a, 3-c.
fetchone()— читает одну следующую строку.fetchall()— считывает все оставшиеся строки разом.fetchmany()— считывает заданное количество строк.
Вопрос 4. Повторный вызов fetchall()
Верный ответ: c — вернётся пустой список.
Курсор работает как итератор: после того как все строки считаны, указатель находится в конце выборки. Повторный вызов fetchall() вернёт пустой список, fetchone() — None.
Вопрос 5. Параметризованные запросы
Верный ответ: c — и позиционные %s, и именованные %(name)s.
Позиционные параметры передаются кортежем, именованные — словарём. Прямая подстановка строк в SQL небезопасна.
Вопрос 6. Команда USE
Ответ: команда USE выбирает активную базу данных для текущего соединения.
После выполнения USE market все следующие запросы к таблицам будут работать с базой market.
Вопрос 7. DictCursor
Ответ: DictCursor возвращает каждую строку результата в виде словаря, где ключами являются названия колонок.
Это позволяет обращаться к значениям по имени: row["first_name"] вместо row[0].
Вопрос 8. Метод commit()
Верный ответ: b — сохраняет изменения в базе.
commit() не закрывает соединение. Для отката изменений используется rollback().
Вопрос 9. Метод rollback()
Верный ответ: c — при возникновении ошибки во время транзакции.
rollback() отменяет все изменения, сделанные с начала текущей транзакции. Используется в блоке except после неудачных изменяющих запросов.
Вопрос 10. Цели транзакций
Верный ответ: b — транзакции не ускоряют SELECT-запросы.
Транзакции нужны для обеспечения целостности, атомарности и возможности отката изменений при ошибке.
Практика 1. Список сотрудников по убыванию зарплаты
import pymysql
config = {
"host": "ich-db.edu.itcareerhub.de",
"user": "ich1",
"password": "password",
"database": "hr",
}
department = input("Enter department: ")
with pymysql.connect(**config) as connection:
with connection.cursor() as cursor:
cursor.execute(
"""
SELECT first_name, last_name, job_id, salary
FROM employees e
JOIN departments d ON e.department_id = d.department_id
WHERE d.department_name = %s
ORDER BY salary DESC
""",
(department,)
)
rows = cursor.fetchall()
for index, row in enumerate(rows, start=1):
print(f"{index}. {row[0]} {row[1]} — {row[2]} — {row[3]}")Запрос использует параметризованный %s для безопасной подстановки названия департамента. Сортировка по убыванию зарплаты выполняется на стороне базы данных.
Практика 2. Выбор департамента по номеру
import pymysql
config = {
"host": "ich-db.edu.itcareerhub.de",
"user": "ich1",
"password": "password",
"database": "hr",
}
with pymysql.connect(**config) as connection:
with connection.cursor() as cursor:
cursor.execute("SELECT department_id, department_name FROM departments")
departments = cursor.fetchall()
for dep_id, dep_name in departments:
print(f"{dep_id}. {dep_name}")
dep_number = int(input("Enter department number: "))
cursor.execute(
"SELECT department_name FROM departments WHERE department_id = %s",
(dep_number,)
)
result = cursor.fetchone()
if result:
print(f"You choose: {result[0]}")
else:
print("Department not found.")Сначала загружается список департаментов, затем пользователь выбирает один по номеру. Для получения названия выполняется дополнительный запрос.
Практика 3. Пустой департамент
cursor.execute(
"""
SELECT first_name, last_name, job_id, salary
FROM employees e
JOIN departments d ON e.department_id = d.department_id
WHERE d.department_name = %s
ORDER BY salary DESC
""",
(department,)
)
rows = cursor.fetchall()
if not rows:
print(f"No employees found in {department} department.")
else:
for index, row in enumerate(rows, start=1):
print(f"{index}. {row[0]} {row[1]} — {row[2]} — {row[3]}")Проверка if not rows позволяет вывести сообщение, если в выборке нет строк.
Практика 4. Фильтрация по зарплате
need_filter = input("Would you like to filter employees by salary? (y/n): ")
if need_filter.lower() == "y":
condition = input("Enter condition (>, <, =, >=, <=): ")
salary = float(input("Enter salary: "))
allowed_conditions = {">", "<", "=", ">=", "<="}
if condition in allowed_conditions:
# Используем параметризованный запрос для значения, оператор проверяем заранее
sql = f"""
SELECT first_name, last_name, job_id, salary
FROM employees e
JOIN departments d ON e.department_id = d.department_id
WHERE d.department_name = %s AND salary {condition} %s
ORDER BY salary DESC
"""
cursor.execute(sql, (department, salary))
rows = cursor.fetchall()
for index, row in enumerate(rows, start=1):
print(f"{index}. {row[0]} {row[1]} — {row[2]} — {row[3]}")
else:
print("Invalid condition.")Знак сравнения нельзя передавать как параметр %s, потому что библиотека экранирует его как строку. Поэтому оператор проверяется по белому списку и вставляется в запрос напрямую, а значение зарплаты подставляется безопасно через %s.
Практика 5. Повторный ввод при ошибке
while True:
try:
dep_number = int(input("Enter department number: "))
except ValueError:
print("Invalid input. Please enter a number.")
continue
cursor.execute(
"SELECT department_name FROM departments WHERE department_id = %s",
(dep_number,)
)
result = cursor.fetchone()
if result:
print(f"You selected: {result[0]}")
break
else:
print("Invalid department number. Please try again.")Цикл while True продолжается до тех пор, пока пользователь не введёт корректный номер департамента. Некорректный ввод числа обрабатывается через try/except ValueError.