Flask 3 — полное руководство

1. Введение

Flask — лёгкий WSGI-фреймворк для Python, созданный Армином Ронахером. Начиная с версии 3.0 (2023) Flask получил значительные улучшения: нативную async-поддержку, новый CLI, улучшенную типизацию и многое другое.

Это руководство охватывает Flask 3.x — от самых основ до production-ready решений. Материал построен так, чтобы быть полезным как новичкам, так и опытным разработчикам.

Версия	Дата	Ключевые изменения
3.0	Сентябрь 2023	Async views, новый CLI, улучшенная типизация
3.1	Март 2024	Improvements, bugfixes
3.2	Ноябрь 2024	Улучшения производительности, security-фиксы

2. Установка и быстрый старт

Flask устанавливается через pip. Рекомендуется использовать виртуальное окружение.

python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows
pip install flask

2.1 Быстрый старт

Минимальное приложение на Flask 3 выглядит так:

from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello, Flask 3!'

if __name__ == '__main__':
    app.run()

Запуск: python app.py — сервер стартует на http://127.0.0.1:5000.

Flask CLI (рекомендуемый способ)

flask --app app run --debug

Флаг --debug включает hot-reload и детальные ошибки.

3. Основы Flask

3.1 Маршруты (Routes)

Маршруты определяются декоратором @app.route(). Flask 3 поддерживает типизированные параметры пути.

@app.route('/user/<int:user_id>')
def profile(user_id: int):
    return f'User ID: {user_id}'

@app.route('/post/<slug>')
def post(slug: str):
    return f'Post: {slug}'

# HTTP methods
@app.route('/data', methods=['GET', 'POST'])
def data():
    if request.method == 'POST':
        return 'Created', 201
    return 'Data'

Конвертер	Тип	Пример
`string`	Строка (по умолчанию)	`/post/<slug>`
`int`	Целое число	`/user/<int:id>`
`float`	Число с точкой	`/price/<float:val>`
`path`	Путь (со слешами)	`/files/<path:filepath>`
`uuid`	UUID	`/item/<uuid:uid>`

3.2 Шаблоны (Jinja2)

Flask использует Jinja2 — мощный шаблонизатор. Шаблоны хранятся в папке templates/.

from flask import render_template

@app.route('/hello/<name>')
def hello(name: str):
    return render_template('hello.html', name=name)

templates/hello.html:

<!DOCTYPE html>
<html>
<head><title>Hello</title></head>
<body>
  <h1>Hello, {{ name }}!</h1>
</body>
</html>

Передача данных в шаблоны

Данные передаются именованными аргументами в render_template. Можно передавать любые объекты: строки, числа, списки, словари, модели SQLAlchemy.

@app.route('/user/<int:id>')
def profile(id):
    user = User.query.get_or_404(id)
    posts = Post.query.filter_by(author=user).all()
    stats = {'posts': len(posts), 'likes': sum(p.likes for p in posts)}
    return render_template('profile.html', user=user, posts=posts, stats=stats)

{# profile.html #}
<h1>{{ user.username }}</h1>
<p>Публикаций: {{ stats.posts }}, лайков: {{ stats.likes }}</p>
<ul>
{% for post in posts %}
  <li>{{ post.title }} ({{ post.created_at.strftime('%d.%m.%Y') }})</li>
{% else %}
  <li>Нет публикаций</li>
{% endfor %}
</ul>

Контекст-провайдеры (глобальные переменные)

Контекст-провайдеры внедряют переменные во все шаблоны без передачи их в каждый render_template. Используются для текущего пользователя, настроек сайта, меню.

@app.context_processor
def inject_globals():
    return {
        'site_name': 'MyApp',
        'current_year': datetime.utcnow().year,
        'menu': [{'title': 'Home', 'url': '/'}, {'title': 'About', 'url': '/about'}]
    }

@app.context_processor
def inject_user():
    from flask import g
    return {'current_user': g.get('user')}

# {{ site_name }} — доступен во всех шаблонах
# {{ current_year }} — автоматически обновляется каждый год

Blueprint-specific контекст: Для добавления переменных только в шаблоны конкретного blueprint используйте @blueprint.context_processor — работает как глобальный, но применяется только для маршрутов этого blueprint.

Объект g (request-bound storage)

g — контейнер для данных, живущих в рамках одного запроса. Идеален для подгрузки пользователя, БД-сессий, кешированных данных. Передаётся в шаблон через контекст-провайдер.

from flask import g

@app.before_request
def load_user():
    if 'user_id' in session:
        g.user = User.query.get(session['user_id'])
        g.cart_count = Cart.query.filter_by(user_id=g.user.id).count()
    else:
        g.user = None

@app.context_processor
def inject_g():
    return {'g': g}

template: {{ g.user.username if g.user else 'Гость' }}
template: Корзина ({{ g.cart_count }})

Важно: g сбрасывается после каждого запроса. Не используйте его для хранения данных между запросами — для этого есть session или БД.

Кастомные фильтры и тесты

Jinja2 позволяет добавлять собственные фильтры и тесты для форматирования данных в шаблонах.

@app.template_filter('format_date')
def format_date(date):
    return date.strftime('%d %b %Y, %H:%M')

@app.template_filter('truncate')
def truncate(text, length=100):
    return text[:length] + '...' if len(text) > length else text

@app.template_filter('pluralize')
def pluralize(count, singular='item', plural='items'):
    return f'{count} {singular if count == 1 else plural}'

# {{ post.created_at|format_date }}
# {{ post.body|truncate(50) }}
# {{ items|length|pluralize('post', 'posts') }}

@app.template_test('admin')
def is_admin(user):
    return user.role == 'admin'

@jinja2 {% if user is admin %}Admin Panel{% endif %}

Для Blueprint: @bp.app_template_filter('filter_name') — регистрирует фильтр для всего приложения из blueprint. @bp.app_template_test('test_name') — аналогично для тестов.

include и макросы (macros)

{% include %} вставляет содержимое другого шаблона. Макросы — переиспользуемые фрагменты с параметрами, аналоги функций.

{# templates/macros/forms.html #}
{% macro input(name, label='', type='text', value='') %}
  <div class="field">
    <label for="{{ name }}">{{ label }}</label>
    <input type="{{ type }}" name="{{ name }}" id="{{ name }}" value="{{ value }}">
  </div>
{% endmacro %}

{% macro render_pagination(pagination, endpoint) %}
  <nav class="pagination">
    {% if pagination.has_prev %}
      <a href="{{ url_for(endpoint, page=pagination.prev_num) }}">&laquo;</a>
    {% endif %}
    <span>Стр. {{ pagination.page }} из {{ pagination.pages }}</span>
    {% if pagination.has_next %}
      <a href="{{ url_for(endpoint, page=pagination.next_num) }}">&raquo;</a>
    {% endif %}
  </nav>
{% endmacro %}

{# templates/post.html #}
{% from "macros/forms.html" import input, render_pagination %}

{% extends "base.html" %}
{% block content %}
  <form method="POST">
    {{ input('title', 'Заголовок') }}
    {{ input('email', 'Email', type='email') }}
    <button>Submit</button>
  </form>

  {% include "ads.html" ignore missing %}

  {{ render_pagination(posts, 'blog.index') }}
{% endblock %}

Совет: Макросы храните в templates/macros/. Для переиспользования между приложениями используйте пакет flask-jinja-extras или пакет Python с кастомными расширениями Jinja.

3.3 Статические файлы

Статика хранится в static/ и доступна через /static/....

<link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
<script src="{{ url_for('static', filename='app.js') }}"></script>
<img src="{{ url_for('static', filename='logo.png') }}" alt="Logo">

3.4 Запросы и ответы

from flask import request, jsonify

@app.route('/api', methods=['POST'])
def api():
    # GET параметры
    page = request.args.get('page', 1, type=int)

    # POST JSON
    data = request.get_json()

    # POST Form
    name = request.form.get('name')

    # Заголовки
    auth = request.headers.get('Authorization')

    # Куки
    token = request.cookies.get('session')

    # Ответ
    return jsonify({'status': 'ok', 'data': data})

4. Новое в Flask 3

Flask 3.0 — первый major-релиз с 2018 года. Вот ключевые изменения:

4.1 Async views (нативная async-поддержка)

Flask 3 позволяет использовать async def для view-функций без дополнительных расширений.

@app.route('/async')
async def async_view():
    result = await some_async_function()
    return {'result': result}

Важно: Async-views работают через ASGI-сервер (Uvicorn, Hypercorn) или WSGI с asgiref.sync.async_to_sync. Для production рекомендуется Uvicorn: pip install uvicorn и uvicorn app:app.

4.2 Новый CLI

Flask 3 представил новый CLI на основе Click с улучшенным автокомплитом и группировкой команд.

flask routes        # Показать все маршруты
flask --app app shell  # Интерактивная оболочка
flask db upgrade     # Миграции (с Flask-Migrate)

4.3 Улучшенная конфигурация

Flask 3 поддерживает строгую типизацию через dataclass-подобные объекты конфигурации.

class Config:
    SECRET_KEY: str
    DATABASE_URL: str = 'sqlite:///app.db'
    DEBUG: bool = False

app.config.from_object(Config)
app.config['DATABASE_URL'] = 'postgresql://localhost/app'

5. Blueprints — модульность

Blueprints позволяют организовать приложение в модули. Каждый blueprint — независимый компонент со своими маршрутами, шаблонами, статикой, контекстом и обработчиками ошибок.

5.1 Создание и регистрация

Blueprint создаётся через Blueprint() и регистрируется в приложении через app.register_blueprint().

# admin/__init__.py
from flask import Blueprint

admin = Blueprint(
    'admin',                    # имя blueprint
    __name__,
    url_prefix='/admin',        # префикс всех маршрутов
    template_folder='templates',# папка с шаблонами (относительно blueprint)
    static_folder='static',     # папка со статикой
    static_url_path='/admin/static',  # URL для статики
    subdomain='admin'           # опционально: поддомен
)

from . import views

# admin/views.py
from flask import render_template
from . import admin

@admin.route('/')
def dashboard():
    return render_template('admin/dashboard.html')

@admin.route('/users')
@login_required
def users():
    return render_template('admin/users.html')

# app.py
from admin import admin
app.register_blueprint(admin)

# Множественная регистрация — один blueprint на разные префиксы:
app.register_blueprint(admin, url_prefix='/dashboard')
app.register_blueprint(admin, url_prefix='/admin', subdomain='api')

5.2 Контекст и передача данных в шаблоны

Каждый blueprint может иметь собственные контекст-провайдеры, обработчики запросов и переменные, доступные только в его шаблонах.

# admin/__init__.py
from flask import Blueprint, g

admin = Blueprint('admin', __name__, url_prefix='/admin')

# before_request — выполняется перед каждым маршрутом blueprint
@admin.before_request
def before_admin():
    if not g.get('user') or g.user.role != 'admin':
        return redirect(url_for('auth.login'))
    g.admin_section = 'dashboard'

# context_processor — переменные только для шаблонов этого blueprint
@admin.context_processor
def inject_admin_data():
    return {
        'admin_title': 'Панель управления',
        'admin_menu': [
            {'name': 'Дашборд', 'url': url_for('admin.dashboard')},
            {'name': 'Пользователи', 'url': url_for('admin.users')},
            {'name': 'Настройки', 'url': url_for('admin.settings')}
        ],
        'stats': {
            'users': User.query.count(),
            'orders': Order.query.count()
        }
    }

# after_request — выполняется после каждого ответа blueprint
@admin.after_request
def add_admin_header(response):
    response.headers['X-Admin-Panel'] = 'true'
    return response

В шаблонах blueprint-а доступны все переменные из контекст-провайдера:

{# templates/admin/base.html #}
<h1>{{ admin_title }}</h1>
<nav>
  {% for item in admin_menu %}
    <a href="{{ item.url }}">{{ item.name }}</a>
  {% endfor %}
</nav>
<p>Пользователей: {{ stats.users }}, заказов: {{ stats.orders }}</p>

5.3 Шаблоны и URL в Blueprints

Для ссылок на маршруты внутри blueprint используйте url_for('blueprint_name.endpoint'). Наследование шаблонов работает между blueprints через абсолютные пути.

{# admin/templates/admin/base.html — базовый шаблон для админки #}
<!DOCTYPE html>
<html>
<head>
  <title>{% block title %}Admin{% endblock %}</title>
  <link rel="stylesheet" href="{{ url_for('admin.static', filename='style.css') }}">
</head>
<body>
  <header>
    <a href="{{ url_for('admin.dashboard') }}">Dashboard</a>
    <a href="{{ url_for('admin.users') }}">Users</a>
    <a href="{{ url_for('blog.index') }}">Blog (другой blueprint)</a>
  </header>
  <main>{% block content %}{% endblock %}</main>
</body>
</html>

{# admin/templates/admin/users.html — наследование внутри blueprint #}
{% extends "admin/base.html" %}
{% block title %}Users — Admin{% endblock %}
{% block content %}
  <h1>{{ admin_title }}: Пользователи</h1>
  <table>
    {% for user in users %}
      <tr>
        <td>{{ user.username }}</td>
        <td>{{ user.email }}</td>
        <td><a href="{{ url_for('admin.edit_user', id=user.id) }}">Edit</a></td>
      </tr>
    {% endfor %}
  </table>
{% endblock %}

Наследование между blueprints

{# blog/templates/blog/base.html — наследует от admin/base.html #}
{% extends "admin/base.html" %}
{% block title %}Blog{% endblock %}
{% block content %}
  <h1>{{ admin_title }}: Блог</h1>
  {% for post in posts %}
    <article>
      <h2>{{ post.title }}</h2>
      <p>{{ post.body|truncate(200) }}</p>
      <a href="{{ url_for('blog.detail', slug=post.slug) }}">Читать</a>
    </article>
  {% endfor %}
{% endblock %}

5.4 Поддомены и вложенные Blueprints

Blueprints поддерживают привязку к поддоменам и вложенность (через регистрацию на разные префиксы).

# Поддомены — каждый blueprint на своём поддомене
api = Blueprint('api', __name__, subdomain='api')
admin = Blueprint('admin', __name__, subdomain='admin', url_prefix='/admin')

app.register_blueprint(api)
app.register_blueprint(admin)

# app.config['SERVER_NAME'] = 'example.com'
# api.example.com/ — API
# admin.example.com/admin/ — Админка

# Вложенные blueprints (через префиксы)
blog = Blueprint('blog', __name__, url_prefix='/blog')
posts = Blueprint('posts', __name__)

# posts регистрируется ВНУТРИ blog
@posts.route('/')
def index():
    return 'Posts index'

@posts.route('/<int:id>')
def detail(id):
    return f'Post {id}'

app.register_blueprint(blog)
blog.register_blueprint(posts, url_prefix='/posts')
# /blog/posts/ — Posts index
# /blog/posts/42 — Post 42

Совет: Вложенные blueprints удобны для иерархических разделов (блог → посты → комментарии). URL-построение: url_for('blog.posts.detail', id=42).

5.5 Обработчики ошибок и middleware

Blueprints могут иметь собственные обработчики ошибок, которые перекрывают глобальные только в рамках своих маршрутов.

@admin.error_handler(403)
def admin_forbidden(e):
    return render_template('admin/403.html'), 403

@admin.error_handler(404)
def admin_not_found(e):
    return render_template('admin/404.html'), 404

# Глобальный обработчик 500 (если blueprint не переопределяет)
@app.errorhandler(500)
def internal_error(e):
    db.session.rollback()
    return render_template('errors/500.html'), 500

# app_errorhandler — работает во всём приложении, даже для blueprint
@admin.app_errorhandler(500)
def admin_internal_error(e):
    return render_template('admin/500.html'), 500

Разница между errorhandler и app_errorhandler для blueprint:

Метод	Область
`@bp.errorhandler`	Только маршруты этого blueprint
`@bp.app_errorhandler`	Всё приложение, НО определён в blueprint

6. Базы данных

6.1 Flask-SQLAlchemy

Flask-SQLAlchemy — стандартное расширение для работы с БД. Поддерживает PostgreSQL, MySQL, SQLite и другие.

pip install flask-sqlalchemy

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)
    email: Mapped[str]

    def __repr__(self):
        return f'<User {self.username}>'

with app.app_context():
    db.create_all()

CRUD операции

# Create
user = User(username='alice', email='alice@example.com')
db.session.add(user)
db.session.commit()

# Read
User.query.get(1)
User.query.filter_by(username='alice').first()
User.query.all()

# Update
user.email = 'new@example.com'
db.session.commit()

# Delete
db.session.delete(user)
db.session.commit()

6.2 Миграции (Flask-Migrate)

pip install flask-migrate

from flask_migrate import Migrate
migrate = Migrate(app, db)

# CLI
flask db init
flask db migrate -m "Initial migration"
flask db upgrade

7. Формы и валидация

Flask-WTF — стандартное расширение для работы с формами, CSRF-защитой и валидацией.

pip install flask-wtf

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length

class LoginForm(FlaskForm):
    email = StringField('Email', validators=[
        DataRequired(), Email()])
    password = PasswordField('Пароль', validators=[
        DataRequired(), Length(min=6)])
    submit = SubmitField('Войти')

@app.route('/login', methods=['GET', 'POST'])
def login():
    form = LoginForm()
    if form.validate_on_submit():
        return f'Logged in as {form.email.data}'
    return render_template('login.html', form=form)

8. REST API

Flask отлично подходит для создания REST API. Встроенные инструменты + Flask-RESTful или ручная реализация.

@app.route('/api/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify([{
        'id': u.id,
        'username': u.username,
        'email': u.email
    } for u in users])

@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    user = User(username=data['username'],
                email=data['email'])
    db.session.add(user)
    db.session.commit()
    return jsonify({'id': user.id}), 201

8.1 Class-based Views (MethodView)

from flask.views import MethodView

class UserAPI(MethodView):
    def get(self, user_id):
        user = User.query.get_or_404(user_id)
        return jsonify({'id': user.id, 'username': user.username})

    def put(self, user_id):
        user = User.query.get_or_404(user_id)
        data = request.get_json()
        user.username = data.get('username', user.username)
        db.session.commit()
        return jsonify({'status': 'updated'})

    def delete(self, user_id):
        user = User.query.get_or_404(user_id)
        db.session.delete(user)
        db.session.commit()
        return '', 204

app.add_url_rule('/api/users/<int:user_id>',
                 view_func=UserAPI.as_view('user_api'))

8.2 Валидация запросов

Для валидации API-запросов используйте Marshmallow или Pydantic.

pip install marshmallow

from marshmallow import Schema, fields, validate

class UserSchema(Schema):
    username = fields.String(required=True,
        validate=validate.Length(min=3, max=50))
    email = fields.Email(required=True)

schema = UserSchema()
errors = schema.validate(request.get_json())
if errors:
    return jsonify(errors), 400

9. Аутентификация

Flask-Login — стандартное решение для управления сессиями пользователей.

pip install flask-login

from flask_login import LoginManager, UserMixin, login_user, login_required, current_user, logout_user

login_manager = LoginManager(app)
login_manager.login_view = 'login'

class User(UserMixin, db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

@app.route('/login', methods=['POST'])
def login():
    user = User.query.filter_by(
        username=request.form['username']).first()
    if user and check_password(user.password, request.form['password']):
        login_user(user)
        return redirect('/profile')
    return 'Invalid credentials', 401

@app.route('/profile')
@login_required
def profile():
    return f'Hello, {current_user.username}!'

JWT-аутентификация: Для API используйте flask-jwt-extended — поддержка access/refresh токенов, JWT в куках или заголовках.

10. Тестирование

Flask предоставляет встроенный test client для unit-тестов.

pip install pytest

# tests/test_app.py
import pytest
from app import app

@pytest.fixture
def client():
    app.config['TESTING'] = True
    with app.test_client() as client:
        yield client

def test_home(client):
    resp = client.get('/')
    assert resp.status_code == 200
    assert b'Hello' in resp.data

def test_api_create_user(client):
    resp = client.post('/api/users', json={
        'username': 'test',
        'email': 'test@example.com'
    })
    assert resp.status_code == 201
    assert resp.json['id'] == 1

def test_protected_route(client):
    resp = client.get('/profile')
    assert resp.status_code == 302
    assert '/login' in resp.location

11. Обработка ошибок

@app.errorhandler(404)
def not_found(e):
    return jsonify({'error': 'Not found'}), 404

@app.errorhandler(500)
def server_error(e):
    app.logger.error(f'Server error: {e}')
    return jsonify({'error': 'Internal server error'}), 500

# Кастомные исключения
class APIError(Exception):
    def __init__(self, message, code=400):
        self.message = message
        self.code = code

@app.errorhandler(APIError)
def handle_api_error(e):
    return jsonify({'error': e.message}), e.code

12. Middleware и Before/After Request

@app.before_request
def before_request():
    # Логирование, проверка заголовков, rate-limiting
    if request.path.startswith('/api/'):
        if not request.headers.get('X-API-Key'):
            return jsonify({'error': 'API key required'}), 401

@app.after_request
def after_request(response):
    # CORS, заголовки безопасности
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-Content-Type-Options'] = 'nosniff'
    return response

@app.teardown_request
def teardown(e):
    # Очистка ресурсов
    db.session.remove()

13. Безопасность

Угроза	Защита	Расширение
CSRF	Flask-WTF (встроенная защита форм)	`flask-wtf`
XSS	Jinja2 autoescape (включён по умолчанию)	Встроено
SQL Injection	SQLAlchemy ORM (параметризация)	`flask-sqlalchemy`
Clickjacking	`X-Frame-Options: DENY`	Middleware
Session hijacking	Secure cookies, HTTPS, `SESSION_COOKIE_HTTPONLY=True`	Конфигурация
Rate limiting	Flask-Limiter	`flask-limiter`
CORS	Flask-CORS	`flask-cors`

Критично: Никогда не храните SECRET_KEY в коде. Используйте переменные окружения: app.config['SECRET_KEY'] = os.environ['SECRET_KEY']. Для production — убедитесь, что DEBUG=False.

14. Кеширование

pip install flask-caching

from flask_caching import Cache

app.config['CACHE_TYPE'] = 'SimpleCache'
app.config['CACHE_DEFAULT_TIMEOUT'] = 300
cache = Cache(app)

@app.route('/expensive')
@cache.cached(timeout=60)
def expensive_view():
    return compute_heavy_result()

@app.route('/user/<int:user_id>')
@cache.cached(key_prefix=lambda: f'user_{request.view_args["user_id"]}')
def user_detail(user_id):
    user = User.query.get(user_id)
    return render_template('user.html', user=user)

15. Асинхронность и фоновые задачи

15.1 WebSockets (Flask-SocketIO)

pip install flask-socketio

from flask_socketio import SocketIO, emit

socketio = SocketIO(app, cors_allowed_origins='*')

@socketio.on('connect')
def handle_connect():
    emit('message', {'data': 'Connected'})

@socketio.on('chat')
def handle_chat(data):
    emit('message', data, broadcast=True)

if __name__ == '__main__':
    socketio.run(app)

15.2 Фоновые задачи (Celery)

pip install celery redis

from celery import Celery

celery = Celery(app.name,
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0')

@celery.task
def send_email_task(recipient, subject, body):
    # Отправка email
    return f'Email sent to {recipient}'

@app.route('/send')
def send():
    send_email_task.delay('user@example.com',
                          'Hello', 'Body text')
    return 'Email queued'

16. Загрузка файлов

import os
from werkzeug.utils import secure_filename

UPLOAD_FOLDER = 'uploads'
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'pdf'}
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16 MB

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return 'No file', 400
    file = request.files['file']
    if file.filename == '' or not allowed_file(file.filename):
        return 'Invalid file', 400
    filename = secure_filename(file.filename)
    file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
    return jsonify({'filename': filename})

17. Интернационализация (i18n)

pip install flask-babel

from flask_babel import Babel, _

babel = Babel(app)

@babel.localeselector
def get_locale():
    return request.accept_languages.best_match(['ru', 'en'])

@app.route('/')
def home():
    return _('Hello, World!')  # Автоперевод

18. CLI-команды

Flask использует Click для создания кастомных команд.

import click
from flask.cli import with_appcontext

@app.cli.command('create-user')
@click.argument('username')
@click.option('--email', default=None)
@with_appcontext
def create_user(username, email):
    user = User(username=username, email=email or f'{username}@example.com')
    db.session.add(user)
    db.session.commit()
    click.echo(f'Created user: {user}')

Запуск: flask create-user alice --email alice@example.com

19. Docker

# Dockerfile
FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 5000

CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:5000", "app:app"]

# docker-compose.yml
version: '3.8'
services:
  web:
    build: .
    ports:
      - "5000:5000"
    environment:
      - FLASK_ENV=production
      - SECRET_KEY=${SECRET_KEY}
      - DATABASE_URL=postgresql://user:pass@db/app
    depends_on:
      - db
  db:
    image: postgres:16
    volumes:
      - pgdata:/var/lib/postgresql/data

20. Production — деплой

Для production используйте WSGI-сервер (Gunicorn, Waitress) или ASGI-сервер (Uvicorn) вместо встроенного сервера Flask.

pip install gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 app:app

# С Nginx
# server {
#     listen 80;
#     location / {
#         proxy_pass http://127.0.0.1:8000;
#         proxy_set_header Host $host;
#         proxy_set_header X-Real-IP $remote_addr;
#     }
# }

Production checklist: 1) DEBUG=False, 2) Случайный SECRET_KEY, 3) HTTPS (Let's Encrypt), 4) Rate limiting, 5) Мониторинг, 6) Логирование, 7) Регулярные бэкапы БД, 8) Health check endpoint.

20.1 Мониторинг и логирование

import logging
from logging.handlers import RotatingFileHandler

if not app.debug:
    handler = RotatingFileHandler('app.log', maxBytes=10240, backupCount=10)
    handler.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)s: %(message)s'))
    handler.setLevel(logging.INFO)
    app.logger.addHandler(handler)

@app.route('/health')
def health():
    return jsonify({
        'status': 'ok',
        'version': '1.0.0'
    })

21. Популярные расширения

Расширение	Назначение	Установка
Flask-SQLAlchemy	ORM для БД	`flask-sqlalchemy`
Flask-Migrate	Миграции БД	`flask-migrate`
Flask-WTF	Формы и CSRF	`flask-wtf`
Flask-Login	Аутентификация	`flask-login`
Flask-JWT-Extended	JWT-токены	`flask-jwt-extended`
Flask-RESTful	REST API	`flask-restful`
Flask-Caching	Кеширование	`flask-caching`
Flask-SocketIO	WebSockets	`flask-socketio`
Flask-CORS	CORS	`flask-cors`
Flask-Limiter	Rate limiting	`flask-limiter`
Flask-Babel	i18n	`flask-babel`
Flask-Mail	Email	`flask-mail`
Flask-Admin	Админ-панель	`flask-admin`
Flask-DebugToolbar	Отладка	`flask-debugtoolbar`

22. Чеклист: от разработки до production

Разработка

Виртуальное окружение (venv)
Структура проекта (blueprints, models, templates, static)
Конфигурация через переменные окружения
Миграции БД (Flask-Migrate)
Тесты (pytest + test client)
Обработка ошибок (404, 500, кастомные)

Безопасность

SECRET_KEY — случайная строка, не в коде
CSRF-защита (Flask-WTF)
CORS (Flask-CORS) — только нужные origins
Rate limiting (Flask-Limiter)
HTTPS (Let's Encrypt / reverse proxy)
Заголовки безопасности (X-Frame-Options, CSP)

Production

WSGI/ASGI сервер (Gunicorn / Uvicorn)
DEBUG=False
Логирование в файл (RotatingFileHandler)
Health check endpoint (/health)
Мониторинг (Prometheus + Grafana / Sentry)
Docker-образ (многоступенчатая сборка)
Регулярные бэкапы БД
CI/CD (GitHub Actions / GitLab CI)

Заключение: Flask 3 — современный, гибкий и зрелый фреймворк. Он подходит как для микросервисов и API, так и для полноценных веб-приложений. Главное преимущество Flask — минимальный порог входа при максимальной гибкости. Выбирайте расширения под задачи, не перегружайте проект лишними зависимостями, и Flask будет служить вам долго.

23. Практические примеры

Готовые к запуску проекты. Копируйте, сохраняйте и запускайте.

23.1 Мини-блог на Flask + SQLAlchemy

Полный CRUD для постов блога с авторизацией.

from flask import Flask, render_template, redirect, url_for, request, flash
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, UserMixin, login_user, login_required, logout_user, current_user
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.db'
db = SQLAlchemy(app)
login_manager = LoginManager(app)
login_manager.login_view = 'login'

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(200))
    def set_password(self, password):
        self.password_hash = generate_password_hash(password)
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    body = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    author = db.relationship('User', backref='posts')

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

@app.route('/')
def index():
    posts = Post.query.order_by(Post.created_at.desc()).all()
    return render_template('index.html', posts=posts)

@app.route('/post/<int:id>')
def post_detail(id):
    post = Post.query.get_or_404(id)
    return render_template('post.html', post=post)

@app.route('/post/new', methods=['GET', 'POST'])
@login_required
def create_post():
    if request.method == 'POST':
        post = Post(title=request.form['title'], body=request.form['body'], author=current_user)
        db.session.add(post); db.session.commit()
        return redirect(url_for('post_detail', id=post.id))
    return render_template('create_post.html')

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

Как запустить: pip install flask flask-sqlalchemy flask-login, python blog.py. Добавьте шаблоны в templates/.

23.2 REST API + JWT аутентификация

API для задач (todo) с JWT-токенами, пагинацией и поиском.

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
from datetime import timedelta
from werkzeug.security import generate_password_hash, check_password_hash

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///todo.db'
app.config['JWT_SECRET_KEY'] = 'jwt-secret'
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)
db = SQLAlchemy(app)
jwt = JWTManager(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(200))
    def set_password(self, password):
        self.password_hash = generate_password_hash(password)
    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

class Todo(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    completed = db.Column(db.Boolean, default=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

@app.route('/api/auth/register', methods=['POST'])
def register():
    data = request.get_json()
    if User.query.filter_by(username=data['username']).first():
        return jsonify({'error': 'User exists'}), 409
    user = User(username=data['username'])
    user.set_password(data['password'])
    db.session.add(user); db.session.commit()
    return jsonify({'token': create_access_token(identity=str(user.id))}), 201

@app.route('/api/auth/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(username=data['username']).first()
    if not user or not user.check_password(data['password']):
        return jsonify({'error': 'Invalid credentials'}), 401
    return jsonify({'token': create_access_token(identity=str(user.id))})

@app.route('/api/todos', methods=['GET'])
@jwt_required()
def get_todos():
    user_id = int(get_jwt_identity())
    q = request.args.get('q', '')
    page = request.args.get('page', 1, type=int)
    query = Todo.query.filter(Todo.user_id == user_id, Todo.title.ilike(f'%{q}%'))
    pagination = query.paginate(page=page, per_page=10, error_out=False)
    return jsonify({
        'todos': [{'id': t.id, 'title': t.title, 'completed': t.completed} for t in pagination.items],
        'total': pagination.total, 'page': page, 'pages': pagination.pages
    })

@app.route('/api/todos', methods=['POST'])
@jwt_required()
def create_todo():
    data = request.get_json()
    todo = Todo(title=data['title'], user_id=int(get_jwt_identity()))
    db.session.add(todo); db.session.commit()
    return jsonify({'id': todo.id}), 201

@app.route('/api/todos/<int:todo_id>', methods=['PUT'])
@jwt_required()
def update_todo(todo_id):
    todo = Todo.query.filter_by(id=todo_id, user_id=int(get_jwt_identity())).first_or_404()
    data = request.get_json()
    todo.title = data.get('title', todo.title)
    todo.completed = data.get('completed', todo.completed)
    db.session.commit()
    return jsonify({'status': 'updated'})

@app.route('/api/todos/<int:todo_id>', methods=['DELETE'])
@jwt_required()
def delete_todo(todo_id):
    todo = Todo.query.filter_by(id=todo_id, user_id=int(get_jwt_identity())).first_or_404()
    db.session.delete(todo); db.session.commit()
    return '', 204

if __name__ == '__main__':
    with app.app_context():
        db.create_all()
    app.run(debug=True)

Тест:

curl -X POST localhost:5000/api/auth/register -H 'Content-Type: application/json' -d '{"username":"test","password":"123"}'

. Используйте токен: curl localhost:5000/api/todos -H 'Authorization: Bearer TOKEN'.

23.3 Чат в реальном времени (WebSocket)

from flask import Flask, render_template
from flask_socketio import SocketIO, emit, join_room, leave_room
from datetime import datetime

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret'
socketio = SocketIO(app, cors_allowed_origins='*')
rooms = {}

@app.route('/')
def index():
    return render_template('chat.html')

@socketio.on('join')
def on_join(data):
    join_room(data['room'])
    if data['room'] not in rooms:
        rooms[data['room']] = []
    emit('message', {'user': 'System', 'text': f"{data['username']} joined", 'time': datetime.now().strftime('%H:%M')}, room=data['room'])

@socketio.on('message')
def on_message(data):
    msg = {'user': data['username'], 'text': data['text'], 'time': datetime.now().strftime('%H:%M')}
    rooms[data['room']].append(msg)
    emit('message', msg, room=data['room'])

@socketio.on('history')
def on_history(data):
    emit('history', rooms.get(data['room'], []))

@socketio.on('typing')
def on_typing(data):
    emit('typing', {'user': data['username']}, room=data['room'], include_self=False)

if __name__ == '__main__':
    socketio.run(app, debug=True)

Клиент: <script src="https://cdn.socket.io/4.7.5/socket.io.min.js"></script>. JS: const socket = io(); socket.emit('join', {room: 'general', username: 'alice'});.

23.4 Фоновая обработка (Celery + Redis)

# tasks.py
from celery import Celery
import time

celery = Celery('tasks', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

@celery.task(bind=True, max_retries=3)
def process_file(self, filepath):
    for i in range(10):
        time.sleep(1)
        self.update_state(state='PROGRESS', meta={'current': i, 'total': 10})
    return {'status': 'done', 'file': filepath}

@celery.task
def send_email(recipient, subject):
    time.sleep(2)
    return f'Email sent to {recipient}: {subject}'

@celery.task
def cleanup_temp_files(days=7):
    return f'Cleaned files older than {days} days'

# app.py
from flask import Flask, request, jsonify
from tasks import process_file

app = Flask(__name__)

@app.route('/upload', methods=['POST'])
def upload():
    task = process_file.delay(request.files['file'].filename)
    return jsonify({'task_id': task.id}), 202

@app.route('/status/<task_id>')
def status(task_id):
    task = process_file.AsyncResult(task_id)
    if task.state == 'PENDING':
        return jsonify({'state': 'PENDING'})
    elif task.state == 'PROGRESS':
        return jsonify({'state': 'PROGRESS', **task.info})
    elif task.state == 'SUCCESS':
        return jsonify({'state': 'SUCCESS', 'result': task.result})
    return jsonify({'state': task.state, 'error': str(task.info)})

Запуск: docker run -d -p 6379:6379 redis. celery -A tasks worker -l info. python app.py.

23.5 Webhook-приёмник с верификацией

import hashlib, hmac, json, subprocess
from flask import Flask, request, jsonify, abort

app = Flask(__name__)
app.config['WEBHOOK_SECRET'] = 'your-secret'

def verify(payload, signature):
    expected = hmac.new(app.config['WEBHOOK_SECRET'].encode(), payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(f'sha256={expected}', signature or '')

@app.route('/webhook/github', methods=['POST'])
def github_webhook():
    if not verify(request.get_data(), request.headers.get('X-Hub-Signature-256')):
        abort(401)
    event = request.headers.get('X-GitHub-Event')
    data = request.get_json()
    print(f'GitHub {event}: {data.get("ref", "unknown")}')
    return jsonify({'status': 'ok'})

@app.route('/webhook/deploy', methods=['POST'])
def deploy():
    data = request.get_json()
    if data.get('ref', '').endswith('/main'):
        result = subprocess.run(['./deploy.sh'], capture_output=True, text=True)
        return jsonify({'output': result.stdout, 'error': result.stderr})
    return jsonify({'deploy': 'skipped'})

Тест:

curl -X POST localhost:8000/webhook/github -H 'X-GitHub-Event: push' -H 'Content-Type: application/json' -d '{"ref":"refs/heads/main"}'

23.6 Docker Compose: Flask + PostgreSQL + Redis + Nginx

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "app:app"]

# docker-compose.yml
version: '3.8'
services:
  app:
    build: .
    env_file: .env
    depends_on: { db: { condition: service_healthy }, redis: { condition: service_started } }
  db:
    image: postgres:16-alpine
    environment: { POSTGRES_DB: app, POSTGRES_USER: app, POSTGRES_PASSWORD: password }
    healthcheck: { test: ["CMD-SHELL", "pg_isready -U app"], interval: 10s, retries: 5 }
  redis:
    image: redis:7-alpine
  nginx:
    image: nginx:alpine
    ports: ["80:80"]
    volumes: ["./nginx.conf:/etc/nginx/conf.d/default.conf"]
    depends_on: [app]
  worker:
    build: .
    command: celery -A tasks worker -l info
    env_file: .env
    depends_on: [redis, db]
volumes: { pgdata: }

Запуск: docker compose up -d --build

23.7 Async-загрузка данных

import asyncio, aiohttp
from flask import Flask, jsonify
from asgiref.sync import async_to_sync

app = Flask(__name__)

async def fetch_rate(session, currency):
    async with session.get(f'https://api.exchangerate-api.com/v4/latest/{currency}') as resp:
        data = await resp.json()
        return {'currency': currency, 'rate': data['rates']['RUB']}

@app.route('/rates')
def get_rates():
    async def run():
        async with aiohttp.ClientSession() as session:
            tasks = [fetch_rate(session, c) for c in ['USD', 'EUR', 'GBP']]
            return await asyncio.gather(*tasks)
    return jsonify(async_to_sync(run)())

23.8 Генерация PDF

from flask import Flask, send_file
from weasyprint import HTML
import tempfile

app = Flask(__name__)

@app.route('/report')
def report():
    html = '<h1>Report</h1><p>Generated at {{ now }}</p>'
    pdf = HTML(string=html).write_pdf()
    with tempfile.NamedTemporaryFile(suffix='.pdf', delete=False) as f:
        f.write(pdf)
        return send_file(f.name, mimetype='application/pdf')

Установка: pip install weasyprint. Требуется система: apt install libpango-1.0-0 libpangocairo-1.0-0.

Flask 3 — полное руководство • Версия 1.0 • 2026