Деплой Flask приложения

my_flask_app/
├── app/
│   ├── __init__.py  # Инициализация Flask
│   ├── models/      # Модели данных
│   ├── routes/      # Маршруты API
│   ├── services/    # Бизнес-логика
│   └── templates/   # HTML шаблоны
├── config.py        # Конфигурация
├── requirements.txt # Зависимости
└── wsgi.py          # Точка входа для WSGI-сервера

flask==2.2.3
SQLAlchemy==2.0.4

flask
sqlalchemy
psycopg2-binary

import os
 
class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hardcoded-secret-for-dev-only'
    
class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'
    
class ProductionConfig(Config):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
    
# Выбор конфигурации по переменной окружения
config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}
 
def get_config():
    return config[os.environ.get('FLASK_ENV', 'default')]

from config import get_config
 
def create_app():
    app = Flask(__name__)
    app.config.from_object(get_config())
    # Остальная инициализация
    return app

from dotenv import load_dotenv
load_dotenv()  # Загружает переменные из файла .env
 
# Теперь можно использовать os.environ.get()

DATABASE_URL=postgresql://user:password@localhost/dbname
SECRET_KEY=your-secret-key-here
FLASK_ENV=development

DATABASE_URL=
SECRET_KEY=
FLASK_ENV=development

app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'

bind = "0.0.0.0:8000"
workers = 4  # Обычно 2-4 воркера на каждое ядро процессора
worker_class = "gevent"  # Асинхронные воркеры для лучшей производительности
max_requests = 1000  # Перезапуск воркера после 1000 запросов
max_requests_jitter = 50  # Случайное число до 50, чтобы избежать одновременного перезапуска
timeout = 30  # Таймаут в секундах
keepalive = 2  # Время в секундах для удержания соединения

from flask import Flask
from flask_caching import Cache
 
app = Flask(__name__)
cache = Cache(app, config={'CACHE_TYPE': 'SimpleCache'})
 
@app.route('/expensive-operation')
@cache.cached(timeout=60)  # Кэшируем результат на 60 секунд
def expensive_operation():
    # Здесь какая-то тяжелая операция
    return result

@cache.memoize(timeout=50)
def get_user(user_id):
    return User.query.get(user_id)

@app.route('/update-user/<int:user_id>', methods=['POST'])
def update_user(user_id):
    # Обновляем пользователя
    # ...
    # Инвалидируем кэш для этого пользователя
    cache.delete_memoized(get_user, user_id)
    return "Updated"

user = User.query.get(1)  # Запрос к БД
same_user = User.query.get(1)  # Берется из кэша сессии, без запроса к БД

from dogpile.cache import make_region
 
region = make_region().configure(
    'dogpile.cache.redis',
    arguments = {
        'host': 'localhost',
        'port': 6379,
        'db': 0,
        'redis_expiration_time': 60*60*2,  # 2 часа
    }
)
 
@region.cache_on_arguments()
def get_popular_products():
    return Product.query.filter(Product.popular == True).all()

cache = Cache(app, config={
    'CACHE_TYPE': 'RedisCache',
    'CACHE_REDIS_HOST': 'localhost',
    'CACHE_REDIS_PORT': 6379,
    'CACHE_REDIS_DB': 0
})

cache = Cache(app, config={
    'CACHE_TYPE': 'MemcachedCache',
    'CACHE_MEMCACHED_SERVERS': ['127.0.0.1:11211']
})

from flask_session import Session
 
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = Redis(host='localhost', port=6379)
Session(app)

@app.route('/')
def index():
    homepage_news = cache.get('homepage_news')
    if homepage_news is None:
        homepage_news = get_latest_news()  # Тяжелая операция
        cache.set('homepage_news', homepage_news, timeout=300)  # 5 минут
    
    return render_template('index.html', news=homepage_news)

app.config['STATIC_URL'] = 'https://cdn.example.com/static/'

@app.route('/api/products')
def products_api():
    response = make_response(jsonify(products))
    # Кэшировать на стороне клиента на 1 час
    response.headers['Cache-Control'] = 'public, max-age=3600'
    return response

@app.route('/api/user/<int:user_id>')
def user_api(user_id):
    user = get_user(user_id)
    last_modified = user.updated_at.timestamp()
    # Проверка If-Modified-Since
    if request.if_modified_since and request.if_modified_since >= last_modified:
        return '', 304  # Not Modified
    
    response = make_response(jsonify(user.to_dict()))
    response.headers['Last-Modified'] = format_date_time(last_modified)
    return response

# Плохо: N+1 запросов
users = User.query.all()
for user in users:
    print(user.profile.name)  # Для каждого пользователя отдельный запрос к профилю
 
# Хорошо: 1 запрос с JOIN
users = User.query.options(joinedload(User.profile)).all()
for user in users:
    print(user.profile.name)  # Данные уже загружены

from flask import Flask
from celery import Celery
 
app = Flask(__name__)
celery = Celery(app.name, broker='redis://localhost:6379/0')
 
@celery.task
def send_email(recipient, subject, body):
    # Логика отправки email
    pass
 
@app.route('/register', methods=['POST'])
def register():
    # Регистрация пользователя
    send_email.delay(user.email, 'Welcome!', 'Thanks for registering')
    return 'Registration successful'

# В Nginx можно настроить переключение upstream
upstream flask_app {
  server blue.example.com;  # Текущая версия
  # server green.example.com;  # Новая версия (закомментирована)
}

import logging
from logging.handlers import RotatingFileHandler
import os
 
def setup_logging(app):
    if not os.path.exists('logs'):
        os.mkdir('logs')
    
    file_handler = RotatingFileHandler('logs/app.log', maxBytes=10240, backupCount=10)
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
    ))
    file_handler.setLevel(logging.INFO)
    
    app.logger.addHandler(file_handler)
    app.logger.setLevel(logging.INFO)
    app.logger.info('Flask application startup')

app = Flask(__name__)
setup_logging(app)

@app.route('/api/resource', methods=['POST'])
def create_resource():
    app.logger.info('Creating new resource')
    try:
        # Логика создания ресурса
        app.logger.info(f'Resource {resource.id} created successfully')
        return jsonify(resource.to_dict())
    except Exception as e:
        app.logger.error(f'Error creating resource: {str(e)}')
        return jsonify({'error': 'Could not create resource'}), 500

import json
from datetime import datetime
 
class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        if hasattr(record, 'request_id'):
            log_record['request_id'] = record.request_id
        if record.exc_info:
            log_record['exception'] = self.formatException(record.exc_info)
        
        return json.dumps(log_record)

import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
 
sentry_sdk.init(
    dsn="ваш-sentry-dsn",
    integrations=[FlaskIntegration()],
    traces_sample_rate=1.0  # Для APM, можно уменьшить в продакшене
)
 
app = Flask(__name__)

from sentry_sdk import configure_scope
 
@app.before_request
def before_request():
    with configure_scope() as scope:
        if current_user.is_authenticated:
            scope.user = {"id": current_user.id, "email": current_user.email}
        scope.set_tag("route", request.endpoint)

import time
from flask import request, g
 
@app.before_request
def start_timer():
    g.start = time.time()
 
@app.after_request
def log_request(response):
    if hasattr(g, 'start'):
        elapsed = time.time() - g.start
        app.logger.info(f"{request.method} {request.path} completed in {elapsed:.2f}s")
    return response

from flask_profiler import Profiler
 
app = Flask(__name__)
app.config["flask_profiler"] = {
    "enabled": True,
    "storage": {
        "engine": "sqlite",
        "FILE": "profiler.sqlite"
    }
}
 
profiler = Profiler(app)

from prometheus_flask_exporter import PrometheusMetrics
 
metrics = PrometheusMetrics(app)
 
# Стандартные метрики включают:
# - Количество запросов
# - Время ответа по квантилям
[H2]- Размер ответа[/H2]
 
# Можно добавлять и свои метрики
custom_counter = metrics.counter(
    'flask_my_custom_counter', 'Количество определенных действий',
    labels={'action': lambda: request.endpoint}
)
 
@app.route('/api/action')
@custom_counter
def api_action():
    # Ваша логика
    return jsonify({'status': 'ok'})

from flask import Flask
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentation
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
app = Flask(__name__)
 
# Настраиваем трассировку
resource = Resource(attributes={SERVICE_NAME: "flask-service"})
provider = TracerProvider(resource=resource)
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)
processor = BatchSpanProcessor(jaeger_exporter)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
 
# Добавляем инструментацию Flask
FlaskInstrumentation().instrument_app(app)
 
# Создаем трейсер для ручного создания спанов
tracer = trace.get_tracer(__name__)

@app.route('/health')
def health_check():
    status = {
        'status': 'ok',
        'db_connection': check_db_connection(),
        'redis_connection': check_redis_connection(),
        'version': app.config.get('VERSION', 'unknown'),
        'uptime_seconds': int(time.time() - app_start_time)
    }
    
    # Если что-то не в порядке, меняем статус и код ответа
    if not status['db_connection'] or not status['redis_connection']:
        status['status'] = 'degraded'
        return jsonify(status), 503
    
    return jsonify(status)

from healthcheck import HealthCheck
 
health = HealthCheck()
 
# Добавляем проверки
health.add_check(check_db_connection)
health.add_check(check_redis_connection)
 
# Регистрируем эндпоинт
app.add_url_rule('/healthcheck', 'healthcheck', view_func=lambda: health.run())

def send_alert(message, severity='warning'):
    if severity == 'critical':
        # Отправка в Slack и SMS
        requests.post(SLACK_WEBHOOK, json={'text': f'КРИТИЧЕСКАЯ ОШИБКА: {message}'})
        send_sms_alert(message)
    else:
        # Только Slack для некритичных проблем
        requests.post(SLACK_WEBHOOK, json={'text': f'Предупреждение: {message}'})
    
    # Логируем все алерты
    app.logger.error(f"ALERT: {severity} - {message}")

@app.route('/problematic-endpoint')
def problematic_endpoint():
    request_id = str(uuid.uuid4())
    app.logger.info(f"[DEBUG:{request_id}] Starting problematic endpoint")
    
    # Ваш код с доп. логированием
    result = some_function()
    app.logger.info(f"[DEBUG:{request_id}] Result: {result}")
    
    return jsonify({'result': result})

@app.route('/api/feature')
def feature_endpoint():
    if app.config.get('ENABLE_NEW_FEATURE'):
        return new_implementation()
    else:
        return old_implementation()

try:
    # Сложная операция
    pass
except Exception as e:
    # Сохраняем контекст
    snapshot = {
        'user_id': current_user.id,
        'params': request.args,
        'data': request.json,
        'time': datetime.utcnow().isoformat()
    }
    save_error_snapshot(snapshot)
    raise

my_awesome_app/
├── app/
│   ├── __init__.py
│   ├── models.py
│   ├── routes.py
│   ├── services.py
│   ├── static/
│   └── templates/
├── migrations/
├── tests/
├── .env.example
├── .gitignore
├── config.py
├── docker-compose.yml
├── Dockerfile
├── gunicorn.conf.py
├── nginx.conf
├── requirements.txt
└── wsgi.py

from app import create_app
 
app = create_app()
 
if __name__ == "__main__":
    app.run()

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
import os
from config import config
 
db = SQLAlchemy()
migrate = Migrate()
 
def create_app():
    app = Flask(__name__)
    config_name = os.environ.get('FLASK_ENV', 'development')
    app.config.from_object(config[config_name])
    
    # Инициализация расширений
    db.init_app(app)
    migrate.init_app(app, db)
    
    # Регистрация blueprints
    from app.routes import main_bp
    app.register_blueprint(main_bp)
    
    # Настройка логирования в продакшене
    if not app.debug and not app.testing:
        import logging
        from logging.handlers import RotatingFileHandler
        
        if not os.path.exists('logs'):
            os.mkdir('logs')
        file_handler = RotatingFileHandler('logs/app.log', maxBytes=10240, backupCount=10)
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
        ))
        file_handler.setLevel(logging.INFO)
        
        app.logger.addHandler(file_handler)
        app.logger.setLevel(logging.INFO)
        app.logger.info('Flask application startup')
    
    return app

import os
from dotenv import load_dotenv
 
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))
 
class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-key-please-change-in-production'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
 
class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'dev.db')
 
class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
        'sqlite:///:memory:'
 
class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'prod.db')
    
    # Настройки безопасности для продакшена
    @classmethod
    def init_app(cls, app):
        # Настройка ProxyFix для работы за Nginx
        from werkzeug.middleware.proxy_fix import ProxyFix
        app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
 
config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

import multiprocessing
 
bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "gevent"
keepalive = 5
timeout = 120
accesslog = "-"  # Логи в stdout для Docker
errorlog = "-"
loglevel = "info"