Работа с БД из Python и веб-приложений

1.

Работа с БД из Python и
Web-приложений
Лекция 13

2.

1.1 Архитектура приложений с БД
Типичная многослойная архитектура
Клиент (браузер, моб. приложение)
HTTP
Веб-сервер (FastAPI)
Обработка HTTP
Бизнес-логика
Правила, валидация
Слой доступа к БД
psycopg / SQLAlchemy
SQL
База данных (PostgreSQL)

3.

1.1 Архитектура приложений с БД
Зачем разделять слои?
● Тестируемость: Можно тестировать слои независимо
● Замена БД: Сменить PostgreSQL → MySQL, не трогая бизнес-логику
● Переиспользование: Один слой доступа к данным для разных
интерфейсов
● Поддерживаемость: Изменения локализованы

4.

1.1 Архитектура приложений с БД
1. Клиент (браузер / мобильное приложение)
Это UI-уровень. Он:
● отправляет HTTP-запросы
● отображает данные
● не должен содержать бизнес-логику (в идеале)
Частая ошибка: часть логики начинают дублировать на клиенте →
рассинхрон.

5.

1.1 Архитектура приложений с БД
2. Веб-сервер (например, FastAPI)
Это входная точка:
принимает HTTP
роутинг (/users, /orders)
сериализация/десериализация (JSON ↔ объекты)
аутентификация (иногда)
Важно: этот слой не должен разрастаться логикой!

6.

1.1 Архитектура приложений с БД
3. Бизнес-логика (service layer)
Самый важный слой:
● правила системы (например: “нельзя купить товар без оплаты”)
● валидация
● оркестрация действий
○
○
○
принимаются решения
вызываются репозитории
формируются сценарии (use cases)
Если этот слой слабый → система превращается в CRUD без структуры.

7.

1.1 Архитектура приложений с БД
4. Слой доступа к БД (repository / DAO)
Инструменты:
● SQLAlchemy
● psycopg
Функции:
● работа с SQL
● изоляция БД от остального кода

8.

1.1 Архитектура приложений с БД
5. База данных (например, PostgreSQL)
Хранит:
● данные
● индексы
● ограничения (constraints)
Часть бизнес-логики иногда осознанно дублируется здесь:
● UNIQUE
● CHECK
● FK

9.

Драйверы баз данных в Python
БД
Драйвер
Описание
PostgreSQL
psycopg2 / psycopg3
Стандарт для PostgreSQL
MySQL
mysql-connector / PyMySQL
Драйверы для MySQL
SQLite
sqlite3
Встроен в Python
MongoDB
pymongo
Официальный драйвер
Redis
redis-py
Официальный клиент

10.

1.2 DB-API 2.0 (PEP 249)
DB-API 2.0 — стандартный интерфейс Python для работы с реляционными
БД, описанный в PEP 249.
PEP 249 — это стандарт, который описывает, как Python-код должен
работать с реляционными базами данных.
Цель: Один и тот же код может работать с разными БД (с минимальными
изменениями).

11.

1.2 DB-API 2.0 (PEP 249)
Основные объекты: Connection → Cursor → SQL
● Connection — канал к БД
● Cursor — инструмент для выполнения SQL
● fetch — получение результата

12.

1.2 DB-API 2.0 (PEP 249)
Connection (подключение)
Объект соединения с БД (например через psycopg):
conn = psycopg.connect(...)
Он отвечает за:
● открытие/закрытие соединения
● транзакции (commit, rollback)
● создание курсоров

13.

1.2 DB-API 2.0 (PEP 249)
Cursor (курсор)
Создаётся из connection:
cur = conn.cursor()
cur.execute("SELECT * FROM users")
Он:
● выполняет SQL
● получает результаты (fetchone, fetchall)

14.

Методы Connection
Метод
Описание
cursor()
Создать курсор
commit()
Зафиксировать транзакцию
rollback()
Откатить транзакцию
close()
Закрыть соединение

15.

Методы Cursor
Метод
Описание
execute(sql, params)
Выполнить запрос
executemany(sql, list)
Выполнить пакетно
fetchone()
Получить одну строку
fetchall()
Получить все строки
fetchmany(size)
Получить N строк

16.

Базовый шаблон работы
import psycopg2
conn = psycopg2.connect("dbname=test user=postgres")
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
result = cursor.fetchone()
conn.commit()
cursor.close()
conn.close()

17.

Транзакции в PEP 249
BEGIN обычно не вызывается явно — он происходит автоматически.
В большинстве драйверов (например, psycopg):
conn = psycopg.connect(...)
cur = conn.cursor()
cur.execute("SELECT 1")
В этот момент:
● соединение не было в транзакции
● после первого execute() драйвер автоматически делает BEGIN

18.

Транзакции в PEP 249
Жизненный цикл транзакции фактически происходит так:
1. (нет транзакции)
2. execute() → BEGIN (неявно)
3. ещё execute()
4. conn.commit() → COMMIT
5. новый execute() → снова BEGIN

19.

1.3 SQL Injection и параметризация
SQL Injection — атака, при которой злоумышленник внедряет вредоносный
SQL через входные данные.
Опасный код:
# НЕ ДЕЛАЙТЕ ТАК
username = input("Username: ")
query = f"SELECT * FROM users WHERE name = '{username}'"
cursor.execute(query)

20.

1.3 SQL Injection и параметризация
Если ввести: ' OR '1'='1
SELECT * FROM users WHERE name = '' OR '1'='1'
-- Вернёт ВСЕХ пользователей!
Если ввести: '; DROP TABLE users; -SELECT * FROM users WHERE name = ''; DROP TABLE users; --'
-- Удалит таблицу!

21.

1.3 SQL Injection и параметризация
Безопасный код: параметризованные запросы
# ПРАВИЛЬНО
username = input("Username: ")
cursor.execute(
"SELECT * FROM users WHERE name = %s",
(username,) # параметры передаются отдельно
)
Как это работает:
● Драйвер БД экранирует значения
● SQL и данные передаются отдельно
● БД понимает, где запрос, а где данные

22.

1.4 ORM vs Raw SQL
ORM (Object-Relational Mapping) — отображение объектов программы на
таблицы БД.
# Raw SQL
cursor.execute("SELECT * FROM users WHERE id = %s", (1,))
row = cursor.fetchone()
name = row[1]
# ORM
user = session.query(User).filter(User.id == 1).first()
name = user.name

23.

1.4 ORM vs Raw SQL
Критерий
Raw SQL
ORM
Скорость разработки
Медленнее в начале
Быстрее для CRUD
Производительность
Максимальный контроль
Почти такая же
Контроль
Полный
Высокий, но абстрактный
Безопасность
Зависит от программиста
Помогает, но не гарантирует
Сложные запросы
Удобнее
Код менее читаемый
Миграции
Вручную
Автоматическая (Alembic)
Переключение БД
Сложно
Легко
Кривая обучения
Только SQL
SQL + ORM API

24.

1.4 ORM vs Raw SQL
Raw SQL подходит для:
Сложных аналитических запросов
Высоконагруженных приложений
Когда нужен полный контроль
Скриптов миграции данных
ORM подходит для:
Большинства веб-приложений
CRUD-операций
Быстрого прототипирования
Команд с разным уровнем знания SQL

25.

Гибридный подход
В реальных проектах часто комбинируют:
● ORM для CRUD (90% операций)
● Raw SQL для сложных отчётов и оптимизаций

26.

1.5 Connection Pool и N+1 проблема
Проблема: Создание подключения к БД — медленная операция (50-100 мс).
# ПЛОХО: новое подключение на каждый запрос
def get_user(user_id):
conn = psycopg2.connect(...) # 50 мс
# запрос — 1 мс
conn.close()

27.

1.5 Connection Pool и N+1 проблема
Решение: Connection Pool — пул переиспользуемых подключений.
1. Приложение стартует
2. Пул создаёт, например, 10 соединений
3. При запросе: берётся готовое соединение
4. После работы: возвращается обратно в пул
Преимущества:
● Подключения создаются один раз
● Готовы к использованию мгновенно
● Управление количеством одновременных подключений

28.

Реализация в psycopg3
from psycopg_pool import ConnectionPool
pool = ConnectionPool(
conninfo="dbname=test user=postgres password=secret",
min_size=1,
max_size=10
)
with pool.connection() as conn:
with conn.cursor() as cur:
cur.execute("SELECT * FROM users WHERE id = %s", (1,))
result = cur.fetchone()
print(result)

29.

1.5 Connection Pool и N+1 проблема
Сценарий: Получить 10 заказов и для каждого — данные о пользователе.
# N+1 ПРОБЛЕМА
orders = session.query(Order).limit(10).all()
# 1 запрос
for order in orders:
user = order.user # +1 запрос на каждый заказ
print(user.name)
# Итого: 1 + 10 = 11 запросов!
ORM под капотом при order.user вызывает отдельный SELECT запрос.

30.

1.5 Connection Pool и N+1 проблема
Решение: Eager loading (жадная загрузка)
# ПРАВИЛЬНО
orders =
session.query(Order).options(joinedload(Order.user)).limit(10).all()
# Один JOIN-запрос вместо 11
for order in orders:
print(order.user.name)
В Raw SQL:
-- Один запрос с JOIN
SELECT o.*, u.* FROM orders o JOIN users u ON o.user_id = u.id LIMIT 10;

31.

1.6 Веб-архитектура: Controller-Service-Repository
Слоистая архитектура
Controller
(роуты, FastAPI handlers)
HTTP, валидация запроса
Service
(правила, оркестрация)
Бизнес-логика
Repository
(запросы к БД)
Доступ к данным
База данных

32.

1.6 Веб-архитектура: Controller-Service-Repository
Repository скрывает детали работы с БД за интерфейсом.
class UserRepository:
def get_by_id(self, user_id):
return session.query(User).filter_by(id=user_id).first()
def create(self, user_data):
user = User(**user_data)
session.add(user)
session.commit()
return user
def find_by_email(self, email):
return session.query(User).filter_by(email=email).first()
Преимущество: Бизнес-логика не знает, как именно хранятся данные.

33.

1.6 Веб-архитектура: Controller-Service-Repository
Service содержит бизнеслогику.
Преимущество: Логика
регистрации не привязана
к HTTP или БД.
class UserService:
def __init__(self, user_repo, email_sender):
self.repo = user_repo
self.email_sender = email_sender
def register(self, email, password):
# Проверка — пользователь не существует
if self.repo.find_by_email(email):
raise ValueError("User exists")
# Создание
user = self.repo.create({
"email": email,
"password_hash": hash(password)
})
# Побочные действия
self.email_sender.send_welcome(email)
return user

34.

1.7 REST API: минимальная база
REST (Representational State Transfer) — архитектурный стиль для веб-API.
Идея: Ресурсы (данные) идентифицируются URL, операции — HTTPметодами.

35.

1.7 REST API: минимальная база
Основные HTTP-методы
GET — Получить данные
GET /users/1
POST — Создать ресурс
POST /users
PUT — Обновить полностью
PUT /users/1
PATCH — Обновить частично
PATCH /users/1
DELETE — Удалить
DELETE /users/1

36.

1.7 REST API: минимальная база
Структура URL
GET /users
GET /users/1
POST /users
PUT /users/1
DELETE /users/1
GET /users/1/orders
Список всех пользователей
Пользователь с id=1
Создать нового пользователя
Обновить пользователя 1
Удалить пользователя 1
Заказы пользователя 1

37.

Коды ответов API
Код
Значение
200 OK
Успех
201 Created
Ресурс создан
204 No Content
Успех без тела ответа
400 Bad Request
Неверный запрос
401 Unauthorized
Нет авторизации
403 Forbidden
Доступ запрещен
404 Not Found
Ресурс не найден
500 Server Error
Ошибка сервера

38.

Формат данных: JSON
HTTP-запрос:
HTTP-ответ:
POST /users HTTP/1.1
Content-Type: application/json
HTTP/1.1 201 Created
Content-Type: application/json
{
{
"name": "Иван",
"email": "ivan@example.com"
}
"id": 42,
"name": "Иван",
"email": "ivan@example.com",
"created_at": "2024-0115T10:30:00Z"
}

39.

2.1 psycopg: работа с PostgreSQL
Установка
Базовое подключение
pip install psycopg2-binary
# или новая версия psycopg3
pip install psycopg
import psycopg2
# Подключение
conn = psycopg2.connect(
host="localhost",
port=5432,
database="techshop",
user="postgres",
password="password"
)
Альтернативное подключение
conn =
psycopg2.connect("postgresql://p
ostgres:password@localhost:5432/
techshop")

40.

2.1 psycopg: работа с PostgreSQL
Выполнение запросов
cursor = conn.cursor()
# SELECT
cursor.execute("SELECT id, name, price FROM products LIMIT 5")
rows = cursor.fetchall()
for row in rows:
print(row) # ('1', 'iPhone', 99990)
# Одна строка
cursor.execute("SELECT * FROM products WHERE id = %s", (1,))
row = cursor.fetchone()

41.

2.1 psycopg: работа с PostgreSQL
Параметризованные запросы (защита от SQL Injection)
# Правильно
cursor.execute(
"SELECT * FROM products WHERE brand = %s AND price < %s",
("Apple", 100000)
)
# Именованные параметры
cursor.execute(
"SELECT * FROM products WHERE brand = %(brand)s AND price <
%(max_price)s",
{"brand": "Apple", "max_price": 100000}
)

42.

2.1 psycopg: работа с PostgreSQL
INSERT
cursor.execute(
"INSERT INTO products (name,
price, brand) VALUES (%s, %s, %s)
RETURNING id",
("New Product", 50000, "Brand
X")
)
new_id = cursor.fetchone()[0]
conn.commit() # Не забываем!
# Множественная вставка
products = [
("Product 1", 1000, "A"),
("Product 2", 2000, "B"),
]
cursor.executemany(
"INSERT INTO products (name,
price, brand) VALUES (%s, %s, %s)",
products
)
conn.commit()

43.

2.1 psycopg: работа с PostgreSQL
UPDATE
cursor.execute(
"UPDATE products SET price = %s WHERE id = %s",
(45000, 1)
)
conn.commit()
DELETE
cursor.execute("DELETE FROM products WHERE id = %s", (5,))
conn.commit()

44.

2.1 psycopg: работа с PostgreSQL
Транзакции
try:
cursor.execute("UPDATE accounts SET balance = balance - 100
WHERE id = 1")
cursor.execute("UPDATE accounts SET balance = balance + 100
WHERE id = 2")
conn.commit() # Фиксируем
except Exception as e:
conn.rollback() # Откатываем при ошибке
print(f"Error: {e}")

45.

2.1 psycopg: работа с PostgreSQL
Context managers (рекомендуемый подход)
# Pythonic — автоматическое управление
with psycopg2.connect("...") as conn:
with conn.cursor() as cursor:
cursor.execute("SELECT * FROM products")
rows = cursor.fetchall()
# cursor автоматически закроется
# conn.commit() при выходе без ошибок, иначе rollback()

46.

2.1 psycopg: работа с PostgreSQL
Получение результатов как словарь
import psycopg2.extras
with psycopg2.connect("...") as conn:
with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as
cursor:
cursor.execute("SELECT id, name, price FROM products")
for row in cursor.fetchall():
print(row['name'], row['price']) # доступ по имени!

47.

2.1 psycopg: работа с PostgreSQL
Connection Pool в psycopg2
Использование
from psycopg2 import pool
conn = connection_pool.getconn()
try:
with conn.cursor() as cursor:
cursor.execute("SELECT 1")
finally:
connection_pool.putconn(conn) #
Вернуть в пул
Создание пула
connection_pool =
pool.SimpleConnectionPool(
minconn=1,
maxconn=10,
host="localhost",
database="techshop",
user="postgres",
password="password"
)

48.

2.2 SQLAlchemy: ORM для Python
Установка
pip install sqlalchemy
pip install psycopg2-binary
# драйвер для PostgreSQL
Core vs ORM
SQLAlchemy Core — низкоуровневый интерфейс (как улучшенный DB-API).
SQLAlchemy ORM — высокоуровневый, объектно-ориентированный подход.

49.

2.2 SQLAlchemy: ORM для Python
Подключение
from sqlalchemy import create_engine
# Engine — точка входа
engine = create_engine(
"postgresql://postgres:password@localhost:5432/techshop",
echo=True # вывод SQL в консоль (для отладки)
)

50.

2.2 SQLAlchemy: ORM для Python
Определение моделей
class Category(Base):
__tablename__ = "categories"
from sqlalchemy.orm import
declarative_base, relationship
from sqlalchemy import Column,
Integer, String, Numeric,
ForeignKey, DateTime
from datetime import datetime
id = Column(Integer,
primary_key=True)
name = Column(String(100),
nullable=False)
Base = declarative_base()
# Связь "один ко многим"
products =
relationship("Product",
back_populates="category")

51.

2.2 SQLAlchemy: ORM для Python
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True)
name = Column(String(200), nullable=False)
price = Column(Numeric(10, 2), nullable=False)
brand = Column(String(100))
category_id = Column(Integer, ForeignKey("categories.id"))
created_at = Column(DateTime, default=datetime.utcnow)
# Обратная связь
category = relationship("Category", back_populates="products")
# Создание всех таблиц
Base.metadata.create_all(engine)

52.

2.2 SQLAlchemy: ORM для Python
Session — объект для работы с БД. Управляет транзакциями.
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

53.

2.2 SQLAlchemy: ORM для Python
Create:
new_product = Product(name="iPhone 15", price=129990,
brand="Apple")
session.add(new_product)
session.commit()
print(new_product.id) # уже есть после commit

54.

2.2 SQLAlchemy: ORM для Python
Read:
# Все товары
products = session.query(Product).all()
# По ID
product = session.query(Product).get(1)
# С фильтром
expensive =
session.query(Product).filter(Product.pri
ce > 100000).all()
# Несколько условий
apple_phones =
session.query(Product).filter(
Product.brand == "Apple",
Product.price < 150000
).all()
# Сортировка и лимит
top_5 =
session.query(Product).order_by(Product.p
rice.desc()).limit(5).all()
# Один результат
first =
session.query(Product).filter_by(brand="A
pple").first()

55.

2.2 SQLAlchemy: ORM для Python
Update:
product = session.query(Product).get(1)
product.price = 119990
session.commit()
Delete:
product = session.query(Product).get(5)
session.delete(product)
session.commit()

56.

2.2 SQLAlchemy: ORM для Python
Связи между моделями
# Получить категорию товара
product = session.query(Product).get(1)
category_name = product.category.name # автоматический JOIN
# Явный LEFT JOIN
session.query(Product, Category).outerjoin(Category).all()
# Получить все товары категории
category = session.query(Category).get(1)
products = category.products # список товаров

57.

2.2 SQLAlchemy: ORM для Python
Решение N+1 проблемы
from sqlalchemy.orm import joinedload
# N+1
products = session.query(Product).all()
for p in products:
print(p.category.name) # +1 запрос на каждый товар
# Eager loading
products =
session.query(Product).options(joinedload(Product.category)).all()
for p in products:
print(p.category.name) # уже загружено

58.

2.2 SQLAlchemy: ORM для Python
Context manager для сессии
from contextlib import contextmanager
@contextmanager
def get_session():
session = Session()
try:
yield session
session.commit()
except Exception:
session.rollback()
raise
finally:
session.close()
# Использование
with get_session() as session:
products =
session.query(Product).all()

59.

2.2 SQLAlchemy: ORM для Python
Repository Pattern с SQLAlchemy
class ProductRepository:
def __init__(self, session):
self.session = session
def get_by_id(self, product_id):
return
self.session.query(Product).get(produc
t_id)
def get_all(self):
return
self.session.query(Product).all()
def find_by_brand(self, brand):
return
self.session.query(Product).filter_by(
brand=brand).all()
def create(self, **kwargs):
product = Product(**kwargs)
self.session.add(product)
self.session.commit()
return product

60.

2.3 redis-py: кэширование на Python
Установка Redis на Windows
Redis официально не поддерживает Windows.
Варианты:
WSL (Windows Subsystem for Linux) — рекомендуется
Memurai — Redis-совместимая БД для Windows
Redis для Windows — сторонняя сборка с github
Docker Desktop — если есть

61.

2.3 redis-py: кэширование на Python
Установка клиента Python (работает
везде)
Подключение
import redis
pip install redis
Проверка
r.ping()
# True
r = redis.Redis(
host='localhost',
port=6379,
db=0,
decode_responses=True #
автоматически декодировать в
строки
)

62.

2.3 redis-py: кэширование на Python
Базовые операции
Строки
r.set("user:1:name", "Иван")
r.get("user:1:name") # "Иван"
С TTL (time-to-live)
r.setex("session:abc", 3600, "user_data")
Счётчики
r.incr("page:views")
r.incrby("page:views", 10)
# 1 час

63.

2.3 redis-py: кэширование на Python
Хеши
r.hset("user:1", mapping={"name": "Иван", "age": "25"})
r.hget("user:1", "name") # "Иван"
r.hgetall("user:1") # {"name": "Иван", "age": "25"}
Списки
r.rpush("queue", "task1", "task2")
r.lpop("queue") # "task1"
Удаление
r.delete("user:1:name")

64.

Паттерн кэширования
import json
import redis
r = redis.Redis(decode_responses=True)
def get_product(product_id):
cache_key =
f"product:{product_id}"
# 1. Проверяем кэш
cached = r.get(cache_key)
if cached:
print("Cache HIT")
return json.loads(cached)
# 2. Если нет — берём из БД
print("Cache MISS")
product =
fetch_from_db(product_id) # медленный
запрос
# 3. Сохраняем в кэш на 5 минут
r.setex(cache_key, 300,
json.dumps(product))
return product

65.

Инвалидация кэша
def update_product(product_id, data):
# Обновляем БД
update_in_db(product_id, data)
# Удаляем из кэша
r.delete(f"product:{product_id}")

66.

2.4 pymongo: работа с MongoDB
Установка
pip install pymongo
Подключение
from pymongo import MongoClient
client = MongoClient("mongodb://localhost:27017")
База и коллекция
db = client.techshop
collection = db.products

67.

2.4 pymongo: работа с MongoDB
Create:
collection.insert_one({
"name": "iPhone 15",
"price": 129990,
"brand": "Apple",
"tags": ["smartphone", "ios"]
})
collection.insert_many([
{"name": "Galaxy S24", "price": 89990, "brand": "Samsung"},
{"name": "Pixel 8", "price": 79990, "brand": "Google"}
])

68.

2.4 pymongo: работа с MongoDB
Read:
all_products = collection.find()
for product in all_products:
print(product)
# С фильтром
apple_products = collection.find({"brand": "Apple"})
# Один документ
product = collection.find_one({"name": "iPhone 15"})
# Операторы
expensive = collection.find({"price": {"$gt": 100000}})

69.

2.4 pymongo: работа с MongoDB
Update:
collection.update_one(
{"name": "iPhone 15"},
{"$set": {"price": 119990}}
)
Delete:
collection.delete_one({"name": "Pixel 8"})
Подсчёт:
count = collection.count_documents({"brand": "Apple"})

70.

2.4 pymongo: работа с MongoDB
Агрегация
pipeline = [
{"$match": {"price": {"$gt": 50000}}},
{"$group": {
"_id": "$brand",
"count": {"$sum": 1},
"avg_price": {"$avg": "$price"}
}},
{"$sort": {"avg_price": -1}}
]
result = collection.aggregate(pipeline)
for doc in result:
print(doc)

71.

2.5 FastAPI: что это и зачем
FastAPI — современный Python-фреймворк для создания REST API.
Ключевые особенности:
Скорость: Один из самых быстрых Python-фреймворков
Автодокументация: Swagger UI генерируется автоматически
Валидация: Через Pydantic-модели
Async: Нативная поддержка async/await
Type hints: Использует подсказки типов Python

72.

Простейший пример
Файл main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def hello():
return {"message": "Hello, World!"}
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id}

73.

Простейший пример
Запуск:
pip install fastapi uvicorn
uvicorn main:app --reload
Откройте http://localhost:8000/docs — там автоматически сгенерированная
документация!

74.

Pydantic — валидация данных
from pydantic import BaseModel
class UserCreate(BaseModel):
name: str
email: str
age: int
@app.post("/users")
def create_user(user: UserCreate):
# FastAPI автоматически:
# - проверит JSON в теле запроса
# - проверит типы
# - вернёт 422 при ошибке
return {"created": user.name}

75.

2.5 FastAPI: что это и зачем
Альтернативы FastAPI:
● Flask — проще, синхронный
● Django — фул-стек фреймворк
● Litestar — конкурент FastAPI

76.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
Структура проекта
myapp/
├── main.py
# FastAPI приложение
├── database.py
# Настройка PostgreSQL
├── models.py
# SQLAlchemy модели
├── schemas.py
# Pydantic схемы
├── repository.py
# Repository слой
├── cache.py
# Redis кэш
└── requirements.txt

77.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
requirements.txt — это файл со списком Python-зависимостей проекта.
Наполнение:
fastapi
uvicorn
sqlalchemy
psycopg2-binary
redis
pydantic

78.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
database.py — подключение к
PostgreSQL
from sqlalchemy import create_engine
from sqlalchemy.orm import
sessionmaker, declarative_base
DATABASE_URL =
"postgresql://postgres:password@localh
ost:5432/techshop"
engine = create_engine(DATABASE_URL)
SessionLocal =
sessionmaker(bind=engine)
Base = declarative_base()
# Зависимость для FastAPI
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

79.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
models.py — модели БД
from sqlalchemy import Column, Integer, String, Numeric
from database import Base
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(200), nullable=False)
price = Column(Numeric(10, 2), nullable=False)
brand = Column(String(100))

80.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
schemas.py — Pydantic-схемы
class ProductCreate(ProductBase):
pass
from pydantic import BaseModel
class ProductBase(BaseModel):
name: str
price: float
brand: str
class ProductResponse(ProductBase):
id: int
class Config:
from_attributes = True
SQLAlchemy
# для

81.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
cache.py — Redis-кэш
import redis
import json
redis_client = redis.Redis(
host='localhost',
port=6379,
decode_responses=True
)
def get_cached(key: str):
value = redis_client.get(key)
return json.loads(value) if value
else None
def set_cached(key: str, value: dict,
ttl: int = 300):
redis_client.setex(key, ttl,
json.dumps(value))
def invalidate(key: str):
redis_client.delete(key)

82.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
repository.py — Repository
from sqlalchemy.orm import Session
from models import Product
class ProductRepository:
def __init__(self, db: Session):
self.db = db
def get_all(self):
return self.db.query(Product).all()
def get_by_id(self, product_id: int):
return self.db.query(Product).filter_by(id=product_id).first()

83.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
repository.py — Repository
def create(self, name: str, price: float, brand: str):
product = Product(name=name, price=price, brand=brand)
self.db.add(product)
self.db.commit()
self.db.refresh(product)
return product
def update(self, product_id: int, data: dict):
product = self.get_by_id(product_id)
if product:
for key, value in data.items():
setattr(product, key, value)
self.db.commit()
self.db.refresh(product)
return product

84.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
repository.py — Repository
def delete(self, product_id: int):
product = self.get_by_id(product_id)
if product:
self.db.delete(product)
self.db.commit()
return product

85.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
main.py — FastAPI приложение
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from database import get_db, engine, Base
from models import Product
from schemas import ProductCreate, ProductResponse
from repository import ProductRepository
from cache import get_cached, set_cached, invalidate
# Создаём таблицы
Base.metadata.create_all(engine)
app = FastAPI(title="TechShop API")

86.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
main.py — FastAPI приложение
@app.get("/products", response_model=list[ProductResponse])
def list_products(db: Session = Depends(get_db)):
repo = ProductRepository(db)
return repo.get_all()
@app.get("/products/{product_id}", response_model=ProductResponse)
def get_product(product_id: int, db: Session = Depends(get_db)):
cache_key = f"product:{product_id}"
# 1. Пробуем кэш
cached = get_cached(cache_key)
if cached:
return cached

87.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
main.py — FastAPI приложение
# 2. БД
repo = ProductRepository(db)
product = repo.get_by_id(product_id)
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# 3. Кэшируем
response = {
"id": product.id,
"name": product.name,
"price": float(product.price),
"brand": product.brand
}
set_cached(cache_key, response, ttl=300)
return response

88.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
main.py — FastAPI приложение
@app.post("/products", response_model=ProductResponse, status_code=201)
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
repo = ProductRepository(db)
return repo.create(product.name, product.price, product.brand)
@app.put("/products/{product_id}", response_model=ProductResponse)
def update_product(
product_id: int,
product: ProductCreate,
db: Session = Depends(get_db)
):
repo = ProductRepository(db)
updated = repo.update(product_id, product.dict())
if not updated:
raise HTTPException(status_code=404, detail="Product not found")

89.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
main.py — FastAPI приложение
# Инвалидация кэша
invalidate(f"product:{product_id}")
return updated
@app.delete("/products/{product_id}", status_code=204)
def delete_product(product_id: int, db: Session = Depends(get_db)):
repo = ProductRepository(db)
deleted = repo.delete(product_id)
if not deleted:
raise HTTPException(status_code=404, detail="Product not found")
# Инвалидация кэша
invalidate(f"product:{product_id}")
return None

90.

2.6 Пример приложения: CRUD с PostgreSQL и Redis
Запуск
uvicorn main:app --reload
Тестирование
# Создать товар
curl -X POST http://localhost:8000/products \
-H "Content-Type: application/json" \
-d '{"name":"iPhone 15","price":129990,"brand":"Apple"}'
# Получить товар (первый раз — из БД)
curl http://localhost:8000/products/1
# Получить ещё раз (из кэша!)
curl http://localhost:8000/products/1

91.

2.7 Сравнение подходов и рекомендации
Сценарий
Инструмент
Простой скрипт с PostgreSQL
psycopg
Веб-приложение с PostgreSQL
SQLAlchemy ORM
Сложные аналитические запросы
psycopg или SQLAlchemy Core
Высокая нагрузка на чтение
Redis для кэша
Гибкая схема данных
pymongo (MongoDB)
Современное API
FastAPI + SQLAlchemy + Redis

92.

2.7 Сравнение подходов и рекомендации
Рекомендации по архитектуре
1. Разделяйте слои: Controller → Service → Repository
2. Параметризуйте запросы: всегда защита от SQL Injection
3. Используйте Connection Pool: не создавайте подключения вручную
4. Решайте N+1: через eager loading или JOIN
5. Кэшируйте аккуратно: инвалидация при изменениях
6. Тестируйте каждый слой отдельно

93.

Что мы изучили
1. DB-API 2.0 — стандартный интерфейс Python для БД
2. SQL Injection — критическая уязвимость, защита через параметризацию
3. ORM vs RAW SQL — сравнение подходов
4. Connection Pool — переиспользование подключений
5. N+1 проблема — частая ошибка, решается eager loading
6. Архитектура — Controller → Service → Repository
7. REST API — Минимальная база про архитектурный стиль
8. psycopg — драйвер PostgreSQL: подключение, транзакции, параметры
9. SQLAlchemy — ORM: модели, связи, сессии, eager loading
10. redis-py — кэширование с TTL, паттерн Cache-Aside
11. pymongo — работа с MongoDB из Python
12. FastAPI — создание REST API

94.

the end