Django & Flask Topic
Django & Flask Interview Questions Advanced
1. How does Django's database transaction management work?
Answer:
Django provides several ways to manage database transactions:
python
from django.db import transaction
# Method 1: Using the transaction.atomic decorator
@transaction.atomic
def update_inventory(order):
# All database operations in this function are atomic
order.item.quantity -= order.quantity
order.item.save()
order.save()
# Method 2: Using context manager
def process_order(self):
try:
with transaction.atomic():
# Atomic block
self.update_inventory()
self.process_payment()
except Exception as e:
handle_error(e)
# Method 3: Savepoints for nested transactions
def complex_operation():
with transaction.atomic():
# Outer transaction
obj1.save()
sid = transaction.savepoint() # Create savepoint
try:
# Inner transaction that might fail
with transaction.atomic():
obj2.risky_operation()
obj2.save()
except Exception:
transaction.savepoint_rollback(sid) # Rollback to savepoint
2. Explain Django's caching framework and its backends.
Answer:
Django provides a robust caching system with multiple backends:
python
# settings.py
CACHES = {
'default': {
'BACKEND': 'django.core.cache.backends.redis.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
},
'local': {
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
'LOCATION': 'unique-snowflake',
},
'file': {
'BACKEND': 'django.core.cache.backends.filebased.FileBasedCache',
'LOCATION': '/var/tmp/django_cache',
}
}
# View-level caching
from django.views.decorators.cache import cache_page
@cache_page(60 * 15) # Cache for 15 minutes
def my_view(request):
# Expensive operation
return render(request, 'template.html')
# Template fragment caching
{% load cache %}
{% cache 500 sidebar %}
<!-- Expensive sidebar content -->
{% endcache %}
# Low-level cache API
from django.core.cache import cache
cache.set('my_key', 'hello', timeout=3600)
value = cache.get('my_key')
3. What are Django's QuerySet annotations and aggregations?
Answer:
Annotations add computed fields to each object in a QuerySet, while aggregations compute values across the entire QuerySet.
python
from django.db.models import Count, Avg, Sum, F, Value
from django.db.models.functions import Concat
# Annotations - add computed fields to each object
books = Book.objects.annotate(
author_full_name=Concat('author__first_name', Value(' '), 'author__last_name'),
profit=F('price') - F('production_cost')
).filter(profit__gt=0)
# Aggregations - compute values across entire QuerySet
from django.db.models import Avg, Max, Min
stats = Book.objects.aggregate(
average_price=Avg('price'),
max_price=Max('price'),
total_books=Count('id')
)
# Complex annotations with conditional expressions
from django.db.models import Case, When, IntegerField
books = Book.objects.annotate(
price_category=Case(
When(price__lt=10, then=Value(1)),
When(price__lt=20, then=Value(2)),
default=Value(3),
output_field=IntegerField(),
)
)
4. How do you implement custom model managers in Django?
Answer:
Custom managers allow you to add custom QuerySet methods and modify the initial QuerySet.
python
class PublishedManager(models.Manager):
def get_queryset(self):
return super().get_queryset().filter(status='published')
def with_comments(self):
return self.get_queryset().annotate(
comment_count=Count('comments')
)
def published_recently(self, days=7):
from django.utils import timezone
return self.filter(
published_date__gte=timezone.now() - timezone.timedelta(days=days)
)
class CustomQuerySet(models.QuerySet):
def active(self):
return self.filter(is_active=True)
def by_user(self, user):
return self.filter(user=user)
class Article(models.Model):
title = models.CharField(max_length=200)
status = models.CharField(max_length=20, choices=STATUS_CHOICES)
published_date = models.DateTimeField(null=True)
# Default manager
objects = models.Manager()
# Custom manager
published = PublishedManager()
# Manager from custom QuerySet
custom_objects = CustomQuerySet.as_manager()
# Usage
recent_articles = Article.published.published_recently()
active_items = Article.custom_objects.active()
5. Explain Django's contenttypes framework.
Answer:
The contenttypes framework provides generic relationships between models, allowing a model to relate to any other model.
python
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
class TaggedItem(models.Model):
tag = models.SlugField()
# ContentType and object_id fields for the generic relation
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
object_id = models.PositiveIntegerField()
content_object = GenericForeignKey('content_type', 'object_id')
class Article(models.Model):
title = models.CharField(max_length=200)
tags = GenericRelation(TaggedItem) # Reverse generic relation
class Video(models.Model):
title = models.CharField(max_length=200)
tags = GenericRelation(TaggedItem)
# Usage
article = Article.objects.get(id=1)
tag = TaggedItem(content_object=article, tag='python')
tag.save()
# Get all tags for an article
article_tags = article.tags.all()
# Get all objects tagged with 'python'
content_type = ContentType.objects.get_for_model(Article)
tagged_articles = TaggedItem.objects.filter(
content_type=content_type,
tag='python'
)
6. How do you implement database routing in Django for multiple databases?
Answer:
Database routers allow you to control which database different models use.
python
# settings.py
DATABASES = {
'default': {
'NAME': 'app_data',
'ENGINE': 'django.db.backends.postgresql',
'USER': 'postgres_user',
'PASSWORD': 's3krit'
},
'users': {
'NAME': 'user_data',
'ENGINE': 'django.db.backends.mysql',
'USER': 'mysql_user',
'PASSWORD': 'priv4te'
}
}
DATABASE_ROUTERS = ['path.to.AuthRouter', 'path.to.PrimaryReplicaRouter']
# routers.py
class AuthRouter:
"""A router to control all database operations on models in the auth application"""
def db_for_read(self, model, **hints):
if model._meta.app_label == 'auth':
return 'users'
return None
def db_for_write(self, model, **hints):
if model._meta.app_label == 'auth':
return 'users'
return None
def allow_relation(self, obj1, obj2, **hints):
if (obj1._meta.app_label == 'auth' or
obj2._meta.app_label == 'auth'):
return True
return None
def allow_migrate(self, db, app_label, model_name=None, **hints):
if app_label == 'auth':
return db == 'users'
return None
# Manual database selection
User.objects.using('users').all()
user.save(using='users')
7. What are Django's system checks and how do you create custom ones?
Answer:
System checks validate your project for common problems and misconfigurations.
python
from django.core.checks import register, Error, Warning
@register()
def check_model_settings(app_configs, **kwargs):
errors = []
# Check if all models have verbose_name
from django.apps import apps
for model in apps.get_models():
if not model._meta.verbose_name:
errors.append(
Error(
'Model missing verbose_name',
obj=model,
hint='Add verbose_name to Meta class',
id='myapp.E001',
)
)
# Check settings
from django.conf import settings
if not hasattr(settings, 'CUSTOM_REQUIRED_SETTING'):
errors.append(
Warning(
'Custom setting not configured',
hint='Add CUSTOM_REQUIRED_SETTING to settings',
id='myapp.W001',
)
)
return errors
# Run checks manually
from django.core.management import call_command
call_command('check')
8. How do you implement real-time features in Django?
Answer:
While Django is synchronous by default, you can integrate real-time features using:
python
# Using Django Channels for WebSockets
# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.room_name = self.scope['url_route']['kwargs']['room_name']
self.room_group_name = f'chat_{self.room_name}'
# Join room group
await self.channel_layer.group_add(
self.room_group_name,
self.channel_name
)
await self.accept()
async def disconnect(self, close_code):
# Leave room group
await self.channel_layer.group_discard(
self.room_group_name,
self.channel_name
)
async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
# Send message to room group
await self.channel_layer.group_send(
self.room_group_name,
{
'type': 'chat_message',
'message': message
}
)
async def chat_message(self, event):
message = event['message']
await self.send(text_data=json.dumps({
'message': message
}))
# Using Celery for background tasks with real-time updates
from celery import shared_task
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
@shared_task
def process_long_running_task(task_id):
# Simulate long-running task
import time
time.sleep(10)
# Send update via WebSocket
channel_layer = get_channel_layer()
async_to_sync(channel_layer.group_send)(
f'task_{task_id}',
{
'type': 'task_update',
'message': 'Task completed successfully'
}
)
Advanced Flask Questions
9. How do you implement application factories with dependency injection in Flask?
Answer:
Application factories combined with dependency injection create highly testable and configurable applications.
python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from config import config
db = SQLAlchemy()
login_manager = LoginManager()
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
# Initialize extensions
db.init_app(app)
login_manager.init_app(app)
# Dependency injection container
class ServiceContainer:
def __init__(self, app):
self.app = app
def get_email_service(self):
from app.services import EmailService
return EmailService(self.app.config['EMAIL_SETTINGS'])
def get_payment_service(self):
from app.services import PaymentService
return PaymentService(self.app.config['PAYMENT_SETTINGS'])
app.container = ServiceContainer(app)
# Register blueprints with dependency injection
from app.main import main as main_blueprint
app.register_blueprint(main_blueprint)
from app.api import api as api_blueprint
app.register_blueprint(api_blueprint, url_prefix='/api')
return app
# Usage in views
def create_user_route():
from flask import current_app
email_service = current_app.container.get_email_service()
email_service.send_welcome_email('user@example.com')
10. Explain Flask's application and request context stacking.
Answer:
Flask allows multiple application and request contexts to be active simultaneously, which is useful for testing and CLI operations.
python
from flask import Flask, current_app, g, request
app = Flask(__name__)
app.config['DATABASE'] = 'primary'
# Context stacking example
def test_context_stacking():
# First application context
with app.app_context():
current_app.config['DATABASE'] = 'primary'
print(f"Outer context: {current_app.config['DATABASE']}")
# Second application context
with app.app_context():
current_app.config['DATABASE'] = 'secondary'
print(f"Inner context: {current_app.config['DATABASE']}")
# Back to outer context
print(f"Back to outer: {current_app.config['DATABASE']}")
# Manual context management for testing
def test_manual_contexts():
# Create application context
ctx = app.app_context()
ctx.push()
try:
# current_app is available here
print(current_app.name)
# Create request context
req_ctx = app.test_request_context('/')
req_ctx.push()
try:
# request is available here
print(request.path)
finally:
req_ctx.pop()
finally:
ctx.pop()
# Using in testing
import unittest
class TestApp(unittest.TestCase):
def setUp(self):
self.app = create_app('testing')
self.app_context = self.app.app_context()
self.app_context.push()
self.client = self.app.test_client()
def tearDown(self):
self.app_context.pop()
def test_something(self):
with self.app.test_request_context('/'):
response = self.client.get('/')
self.assertEqual(response.status_code, 200)
11. How do you implement complex database relationships in Flask-SQLAlchemy?
Answer:
Flask-SQLAlchemy supports all SQLAlchemy relationship patterns.
python
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import relationship, backref
from sqlalchemy import ForeignKey, Table
db = SQLAlchemy()
# Many-to-Many with association table
post_tags = Table('post_tags', db.Model.metadata,
db.Column('post_id', db.Integer, ForeignKey('posts.id')),
db.Column('tag_id', db.Integer, ForeignKey('tags.id'))
)
# Polymorphic relationships
class ContentItem(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100))
type = db.Column(db.String(50)) # discriminator
__mapper_args__ = {
'polymorphic_identity': 'content',
'polymorphic_on': type
}
class Article(ContentItem):
id = db.Column(db.Integer, ForeignKey('content_item.id'), primary_key=True)
body = db.Column(db.Text)
__mapper_args__ = {
'polymorphic_identity': 'article'
}
class Video(ContentItem):
id = db.Column(db.Integer, ForeignKey('content_item.id'), primary_key=True)
url = db.Column(db.String(200))
__mapper_args__ = {
'polymorphic_identity': 'video'
}
# Self-referential many-to-many (friendship)
friendship = Table('friendship', db.Model.metadata,
db.Column('user_id', db.Integer, ForeignKey('users.id'), primary_key=True),
db.Column('friend_id', db.Integer, ForeignKey('users.id'), primary_key=True),
db.Column('created_at', db.DateTime, default=db.func.now())
)
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50))
# Self-referential many-to-many
friends = relationship(
'User',
secondary=friendship,
primaryjoin=id==friendship.c.user_id,
secondaryjoin=id==friendship.c.friend_id,
backref=backref('friend_of', lazy='dynamic'),
lazy='dynamic'
)
# One-to-many with backref
posts = relationship('Post', backref='author', lazy='dynamic')
def add_friend(self, user):
if not self.is_friend(user):
self.friends.append(user)
user.friends.append(self)
def is_friend(self, user):
return self.friends.filter(
friendship.c.friend_id == user.id
).count() > 0
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100))
user_id = db.Column(db.Integer, ForeignKey('users.id'))
# Many-to-many with tags
tags = relationship('Tag', secondary=post_tags, backref=db.backref('posts', lazy='dynamic'))
class Tag(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50), unique=True)
12. How do you implement advanced authentication with Flask?
Answer:
Advanced authentication can include JWT, OAuth, and role-based access control.
python
from flask import Flask, request, jsonify
from flask_jwt_extended import JWTManager, jwt_required, create_access_token, get_jwt_identity
from functools import wraps
import datetime
app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'super-secret'
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = datetime.timedelta(hours=1)
jwt = JWTManager(app)
# Role-based access control decorator
def roles_required(*roles):
def wrapper(fn):
@wraps(fn)
@jwt_required()
def decorator(*args, **kwargs):
current_user = get_jwt_identity()
user_roles = current_user.get('roles', [])
if not any(role in user_roles for role in roles):
return jsonify({"msg": "Access denied"}), 403
return fn(*args, **kwargs)
return decorator
return wrapper
# Multi-factor authentication
class MFAManager:
def __init__(self, app):
self.app = app
def generate_mfa_secret(self, user_id):
import pyotp
return pyotp.random_base32()
def verify_mfa_code(self, secret, code):
import pyotp
totp = pyotp.TOTP(secret)
return totp.verify(code)
mfa_manager = MFAManager(app)
# OAuth2 implementation
from authlib.integrations.flask_client import OAuth
oauth = OAuth(app)
google = oauth.register(
name='google',
client_id='google-client-id',
client_secret='google-client-secret',
server_metadata_url='https://accounts.google.com/.well-known/openid_configuration',
client_kwargs={'scope': 'openid email profile'},
)
@app.route('/login/mfa', methods=['POST'])
@jwt_required()
def mfa_verify():
current_user = get_jwt_identity()
code = request.json.get('code')
user_secret = get_user_mfa_secret(current_user['id'])
if mfa_manager.verify_mfa_code(user_secret, code):
# Grant full access token
access_token = create_access_token(
identity=current_user,
additional_claims={'mfa_verified': True}
)
return jsonify(access_token=access_token)
return jsonify({"msg": "Invalid MFA code"}), 401
@app.route('/admin')
@roles_required('admin')
def admin_dashboard():
return "Admin dashboard"
# JWT claims loader for custom data
@jwt.user_identity_loader
def user_identity_lookup(user):
return {
'id': user.id,
'email': user.email,
'roles': user.roles
}
13. How do you implement database migrations with Flask-Alembic?
Answer:
Alembic provides database migration capabilities for Flask-SQLAlchemy.
python
# migrations/env.py (Alembic configuration)
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
from flask import current_app
# this is the Alembic Config object
config = context.config
# Interpret the config file for Python logging
fileConfig(config.config_file_name)
# add your model's MetaData object here
from app.models import db
target_metadata = db.metadata
def run_migrations_online():
"""Run migrations in 'online' mode."""
connectable = engine_from_config(
config.get_section(config.config_ini_section),
prefix='sqlalchemy.',
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
compare_type=True,
compare_server_default=True
)
with context.begin_transaction():
context.run_migrations()
# Creating custom migration operations
"""create user table
Revision ID: 001
Revises:
Create Date: 2023-01-01 00:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
def upgrade():
# Create users table
op.create_table('users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('email', sa.String(length=255), nullable=False),
sa.Column('password_hash', sa.String(length=255), nullable=False),
sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('email')
)
# Create indexes
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
# Add column with batch operations (for SQLite)
with op.batch_alter_table('users') as batch_op:
batch_op.add_column(sa.Column('full_name', sa.String(length=255)))
def downgrade():
with op.batch_alter_table('users') as batch_op:
batch_op.drop_column('full_name')
op.drop_index(op.f('ix_users_email'), table_name='users')
op.drop_table('users')
# Data migrations
def upgrade():
# Schema changes
op.add_column('users', sa.Column('status', sa.String(50)))
# Data migration
connection = op.get_bind()
connection.execute(
sa.text("UPDATE users SET status = 'active' WHERE status IS NULL")
)
# Using migrations in deployment
"""
# Initialize migrations
flask db init
# Create new migration
flask db migrate -m "add user status"
# Apply migrations
flask db upgrade
# Rollback migration
flask db downgrade
"""
14. How do you implement caching strategies in Flask applications?
Answer:
Flask supports multiple caching strategies and backends.
python
from flask import Flask, request
from flask_caching import Cache
import functools
app = Flask(__name__)
app.config['CACHE_TYPE'] = 'RedisCache'
app.config['CACHE_REDIS_URL'] = 'redis://localhost:6379/0'
app.config['CACHE_DEFAULT_TIMEOUT'] = 300
cache = Cache(app)
# View caching
@app.route('/expensive-view')
@cache.cached(timeout=50, query_string=True)
def expensive_view():
# This will be cached for 50 seconds
expensive_operation()
return render_template('index.html')
# Memoization for function results
@cache.memoize(timeout=60)
def expensive_calculation(user_id, filter_criteria):
# Results are cached based on function arguments
return perform_expensive_query(user_id, filter_criteria)
# Cache invalidation strategies
class CacheManager:
def __init__(self, cache):
self.cache = cache
def invalidate_user_cache(self, user_id):
# Delete all cache keys for a user
keys_pattern = f'user_{user_id}_*'
self._delete_pattern(keys_pattern)
def invalidate_product_cache(self, product_id):
# Invalidate specific product and related caches
self.cache.delete(f'product_{product_id}')
self.cache.delete('recent_products')
self.cache.delete('featured_products')
def _delete_pattern(self, pattern):
# Redis-specific pattern deletion
if hasattr(self.cache.cache, 'delete_pattern'):
self.cache.cache.delete_pattern(pattern)
cache_manager = CacheManager(cache)
# Conditional caching based on request headers
def conditional_caching(timeout=300):
def decorator(f):
@functools.wraps(f)
def decorated_function(*args, **kwargs):
# Don't cache for authenticated users
if request.headers.get('Authorization'):
return f(*args, **kwargs)
# Use cached version for anonymous users
cache_key = f"view_{request.path}_{request.args}"
cached_response = cache.get(cache_key)
if cached_response is not None:
return cached_response
response = f(*args, **kwargs)
cache.set(cache_key, response, timeout=timeout)
return response
return decorated_function
return decorator
@app.route('/api/data')
@conditional_caching(timeout=60)
def api_data():
return jsonify({'data': get_expensive_data()})
15. How do you implement background task processing in Flask?
Answer:
Celery is commonly used for background task processing in Flask.
python
from flask import Flask, jsonify, request
from celery import Celery
import time
app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
def make_celery(app):
celery = Celery(
app.import_name,
backend=app.config['CELERY_RESULT_BACKEND'],
broker=app.config['CELERY_BROKER_URL']
)
class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
celery.Task = ContextTask
return celery
celery = make_celery(app)
# Simple background task
@celery.task
def send_email_async(to, subject, body):
# Simulate email sending
time.sleep(5)
print(f"Email sent to {to}")
return f"Email to {to} sent successfully"
# Periodic tasks
from celery.schedules import crontab
celery.conf.beat_schedule = {
'cleanup-every-hour': {
'task': 'app.tasks.cleanup_old_sessions',
'schedule': crontab(minute=0, hour='*/1'),
},
'generate-reports-daily': {
'task': 'app.tasks.generate_daily_reports',
'schedule': crontab(minute=0, hour=0), # Midnight
},
}
@celery.task
def cleanup_old_sessions():
from app.models import Session
Session.delete_old_sessions()
return "Old sessions cleaned up"
# Task with progress tracking
@celery.task(bind=True)
def long_running_task(self, items):
total = len(items)
for i, item in enumerate(items):
# Process item
process_item(item)
# Update progress
self.update_state(
state='PROGRESS',
meta={
'current': i + 1,
'total': total,
'percent': int((i + 1) / total * 100),
'status': f'Processing item {i + 1} of {total}'
}
)
return {'current': total, 'total': total, 'status': 'Task completed'}
@app.route('/start-task', methods=['POST'])
def start_task():
task = long_running_task.delay(['item1', 'item2', 'item3'])
return jsonify({'task_id': task.id}), 202
@app.route('/task-status/<task_id>')
def task_status(task_id):
task = long_running_task.AsyncResult(task_id)
if task.state == 'PENDING':
response = {
'state': task.state,
'status': 'Pending...'
}
elif task.state == 'PROGRESS':
response = {
'state': task.state,
'status': task.info.get('status', ''),
'progress': task.info.get('percent', 0)
}
elif task.state == 'SUCCESS':
response = {
'state': task.state,
'result': task.info
}
else: # FAILURE
response = {
'state': task.state,
'status': str(task.info)
}
return jsonify(response)
16. How do you implement API versioning in Flask?
Answer:
API versioning can be implemented through URL paths, headers, or query parameters.
python
from flask import Flask, request, jsonify
from functools import wraps
app = Flask(__name__)
# URL-based versioning
@app.route('/api/v1/users')
def users_v1():
return jsonify({
'version': 'v1',
'users': get_users_v1() # Old format
})
@app.route('/api/v2/users')
def users_v2():
return jsonify({
'version': 'v2',
'users': get_users_v2() # New format with additional fields
})
# Header-based versioning
def versioned_api(version):
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
requested_version = request.headers.get('API-Version', 'v1')
if requested_version != version:
return jsonify({
'error': f'Version {version} required',
'supported_versions': ['v1', 'v2']
}), 400
return f(*args, **kwargs)
return decorated_function
return decorator
@app.route('/api/users')
@versioned_api('v2')
def users_header_versioned():
version = request.headers.get('API-Version', 'v1')
if version == 'v1':
return jsonify(get_users_v1())
else:
return jsonify(get_users_v2())
# Content negotiation versioning
class APIVersionManager:
def __init__(self):
self.versions = {}
def register_version(self, version, handler):
self.versions[version] = handler
def get_handler(self, version):
return self.versions.get(version, self.versions['v1'])
api_manager = APIVersionManager()
# Register version handlers
api_manager.register_version('v1', lambda: get_users_v1())
api_manager.register_version('v2', lambda: get_users_v2())
@app.route('/api/users')
def users_content_negotiation():
version = request.args.get('version', 'v1')
accept_header = request.headers.get('Accept', '')
# Extract version from Accept header: application/vnd.myapp.v2+json
if 'vnd.myapp.' in accept_header:
version = accept_header.split('vnd.myapp.')[1].split('+')[0]
handler = api_manager.get_handler(version)
return jsonify(handler())
# Blueprint-based versioning
from flask import Blueprint
v1_bp = Blueprint('v1', __name__, url_prefix='/api/v1')
v2_bp = Blueprint('v2', __name__, url_prefix='/api/v2')
@v1_bp.route('/users')
def v1_users():
return jsonify(get_users_v1())
@v2_bp.route('/users')
def v2_users():
return jsonify(get_users_v2())
app.register_blueprint(v1_bp)
app.register_blueprint(v2_bp)
17. How do you implement rate limiting in Flask applications?
Answer:
Rate limiting protects your API from abuse and ensures fair usage.
python
from flask import Flask, request, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
import redis
app = Flask(__name__)
# Basic rate limiting
limiter = Limiter(
app=app,
key_func=get_remote_address,
default_limits=["200 per day", "50 per hour"],
storage_uri="redis://localhost:6379/0"
)
@app.route('/api/public')
@limiter.limit("10 per minute")
def public_api():
return jsonify({"data": "public data"})
@app.route('/api/expensive')
@limiter.limit("5 per minute")
def expensive_operation():
return jsonify({"result": "expensive computation"})
# User-based rate limiting
def get_user_identifier():
if hasattr(request, 'user') and request.user.is_authenticated:
return request.user.id
return get_remote_address()
user_limiter = Limiter(key_func=get_user_identifier)
@app.route('/api/user-specific')
@user_limiter.limit("100 per hour")
def user_specific():
return jsonify({"data": "user data"})
# Dynamic rate limiting based on user tiers
def tiered_rate_limit():
user_tier = getattr(request, 'user_tier', 'free')
limits = {
'free': "10 per hour",
'premium': "1000 per hour",
'enterprise': "10000 per hour"
}
return limits.get(user_tier, "10 per hour")
@app.route('/api/tiered')
@limiter.limit(tiered_rate_limit)
def tiered_api():
return jsonify({"data": "tiered data"})
# Custom rate limit exceeded handler
@app.errorhandler(429)
def ratelimit_handler(e):
return jsonify({
"error": "Rate limit exceeded",
"message": str(e.description),
"retry_after": e.retry_after
}), 429
# Complex rate limiting with sliding windows
class AdvancedRateLimiter:
def __init__(self, redis_client):
self.redis = redis_client
def check_rate_limit(self, key, limit, window):
"""
key: unique identifier for the client
limit: maximum number of requests
window: time window in seconds
"""
pipeline = self.redis.pipeline()
current_time = int(time.time())
# Remove old requests outside the window
pipeline.zremrangebyscore(key, 0, current_time - window)
# Count current requests in the window
pipeline.zcard(key)
# Add current request
pipeline.zadd(key, {str(current_time): current_time})
# Set expiration
pipeline.expire(key, window)
results = pipeline.execute()
request_count = results[1]
return request_count <= limit
advanced_limiter = AdvancedRateLimiter(redis.Redis())
@app.route('/api/advanced')
def advanced_api():
client_id = get_remote_address()
if not advanced_limiter.check_rate_limit(
f"rate_limit:{client_id}",
limit=100,
window=3600
):
return jsonify({"error": "Rate limit exceeded"}), 429
return jsonify({"data": "advanced api response"})
18. How do you implement comprehensive logging in Flask applications?
Answer:
Proper logging is essential for debugging and monitoring production applications.
python
import logging
from logging.handlers import RotatingFileHandler, SMTPHandler
import sys
from flask import Flask, request
import json
import time
app = Flask(__name__)
class JSONFormatter(logging.Formatter):
def format(self, record):
log_entry = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'logger': record.name,
'message': record.getMessage(),
'module': record.module,
'function': record.funcName,
'line': record.lineno,
}
# Add request context if available
if hasattr(request, 'url'):
log_entry.update({
'url': request.url,
'method': request.method,
'ip': request.remote_addr,
'user_agent': request.headers.get('User-Agent')
})
# Add exception info if present
if record.exc_info:
log_entry['exception'] = self.formatException(record.exc_info)
return json.dumps(log_entry)
def setup_logging(app):
# Remove default handlers
for handler in app.logger.handlers[:]:
app.logger.removeHandler(handler)
# Set log level
app.logger.setLevel(logging.INFO)
# Console handler for development
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(JSONFormatter())
app.logger.addHandler(console_handler)
# File handler for production
file_handler = RotatingFileHandler(
'app.log',
maxBytes=1024 * 1024, # 1MB
backupCount=10
)
file_handler.setLevel(logging.INFO)
file_handler.setFormatter(JSONFormatter())
app.logger.addHandler(file_handler)
# Error email handler for critical errors
if app.config.get('MAIL_SERVER'):
mail_handler = SMTPHandler(
mailhost=(app.config['MAIL_SERVER'], app.config['MAIL_PORT']),
fromaddr=app.config['MAIL_DEFAULT_SENDER'],
toaddrs=app.config['ADMINS'],
subject='Application Error',
credentials=(app.config['MAIL_USERNAME'], app.config['MAIL_PASSWORD'])
)
mail_handler.setLevel(logging.ERROR)
mail_handler.setFormatter(JSONFormatter())
app.logger.addHandler(mail_handler)
setup_logging(app)
# Request logging middleware
@app.before_request
def before_request_logging():
request.start_time = time.time()
@app.after_request
def after_request_logging(response):
if hasattr(request, 'start_time'):
duration = time.time() - request.start_time
app.logger.info('Request processed', extra={
'duration': duration,
'status_code': response.status_code,
'url': request.url,
'method': request.method,
'ip': request.remote_addr
})
return response
# Structured logging in views
@app.route('/api/users/<int:user_id>')
def get_user(user_id):
app.logger.info('Fetching user', extra={
'user_id': user_id,
'action': 'get_user'
})
try:
user = User.query.get_or_404(user_id)
app.logger.info('User found', extra={
'user_id': user_id,
'user_email': user.email
})
return jsonify(user.to_dict())
except Exception as e:
app.logger.error('Error fetching user', extra={
'user_id': user_id,
'error': str(e),
'stack_trace': traceback.format_exc()
})
return jsonify({'error': 'User not found'}), 404
# Custom loggers for different components
database_logger = logging.getLogger('app.database')
api_logger = logging.getLogger('app.api')
auth_logger = logging.getLogger('app.auth')
# Usage in different modules
def database_operation():
database_logger.debug('Executing database query', extra={
'query': 'SELECT * FROM users',
'params': {'limit': 10}
})
# Logging configuration in production
LOGGING_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'json': {
'()': JSONFormatter,
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'json',
'stream': 'ext://sys.stdout'
},
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'formatter': 'json',
'filename': '/var/log/app/app.log',
'maxBytes': 10485760, # 10MB
'backupCount': 5
}
},
'root': {
'level': 'INFO',
'handlers': ['console', 'file']
}
}
19. How do you implement health checks and monitoring in Flask?
Answer:
Health checks are crucial for containerized deployments and monitoring.
python
from flask import Flask, jsonify
import psutil
import redis
from sqlalchemy import text
from prometheus_flask_exporter import PrometheusMetrics
app = Flask(__name__)
metrics = PrometheusMetrics(app)
# Basic health check
@app.route('/health')
def health_check():
return jsonify({"status": "healthy"})
# Comprehensive health check with dependencies
@app.route('/health/detailed')
def detailed_health_check():
health_status = {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"checks": {}
}
# Database health check
try:
db.session.execute(text('SELECT 1'))
health_status["checks"]["database"] = {
"status": "healthy",
"response_time": measure_response_time(
lambda: db.session.execute(text('SELECT 1'))
)
}
except Exception as e:
health_status["checks"]["database"] = {
"status": "unhealthy",
"error": str(e)
}
health_status["status"] = "unhealthy"
# Redis health check
try:
redis_client = redis.Redis()
redis_client.ping()
health_status["checks"]["redis"] = {
"status": "healthy",
"response_time": measure_response_time(redis_client.ping)
}
except Exception as e:
health_status["checks"]["redis"] = {
"status": "unhealthy",
"error": str(e)
}
health_status["status"] = "unhealthy"
# External service health check
try:
response = requests.get('https://api.external-service.com/health', timeout=5)
health_status["checks"]["external_service"] = {
"status": "healthy" if response.status_code == 200 else "unhealthy",
"response_time": response.elapsed.total_seconds()
}
except Exception as e:
health_status["checks"]["external_service"] = {
"status": "unhealthy",
"error": str(e)
}
health_status["status"] = "unhealthy"
# System metrics
health_status["system"] = {
"cpu_percent": psutil.cpu_percent(),
"memory_percent": psutil.virtual_memory().percent,
"disk_usage": psutil.disk_usage('/').percent
}
status_code = 200 if health_status["status"] == "healthy" else 503
return jsonify(health_status), status_code
def measure_response_time(func):
start_time = time.time()
func()
return time.time() - start_time
# Readiness and liveness probes for Kubernetes
@app.route('/health/ready')
def readiness_probe():
"""Check if the application is ready to receive traffic"""
try:
# Check database connection
db.session.execute(text('SELECT 1'))
# Check Redis connection
redis_client = redis.Redis()
redis_client.ping()
# Check if all required services are available
if not all_required_services_available():
return jsonify({"status": "not ready"}), 503
return jsonify({"status": "ready"})
except Exception as e:
return jsonify({"status": "not ready", "error": str(e)}), 503
@app.route('/health/live')
def liveness_probe():
"""Check if the application is running"""
try:
# Basic check - is the application responsive?
return jsonify({"status": "alive"})
except Exception:
return jsonify({"status": "dead"}), 503
# Custom metrics for Prometheus
from prometheus_client import Counter, Histogram, Gauge
REQUEST_COUNT = Counter(
'app_requests_total',
'Total number of requests',
['method', 'endpoint', 'status_code']
)
REQUEST_DURATION = Histogram(
'app_request_duration_seconds',
'Request duration in seconds',
['method', 'endpoint']
)
ACTIVE_USERS = Gauge(
'app_active_users',
'Number of active users'
)
@app.before_request
def before_request_metrics():
request.start_time = time.time()
@app.after_request
def after_request_metrics(response):
if hasattr(request, 'start_time'):
duration = time.time() - request.start_time
REQUEST_COUNT.labels(
method=request.method,
endpoint=request.endpoint,
status_code=response.status_code
).inc()
REQUEST_DURATION.labels(
method=request.method,
endpoint=request.endpoint
).observe(duration)
return response
# Background health check scheduler
import threading
import schedule
class HealthMonitor:
def __init__(self):
self.healthy = True
self.last_check = None
def check_services(self):
try:
# Perform comprehensive health checks
db.session.execute(text('SELECT 1'))
redis_client = redis.Redis()
redis_client.ping()
self.healthy = True
self.last_check = datetime.utcnow()
except Exception as e:
self.healthy = False
app.logger.error(f"Health check failed: {e}")
def start_monitoring(self):
def run_scheduler():
schedule.every(30).seconds.do(self.check_services)
while True:
schedule.run_pending()
time.sleep(1)
thread = threading.Thread(target=run_scheduler, daemon=True)
thread.start()
health_monitor = HealthMonitor()
20. How do you implement feature flags in Flask applications?
Answer:
Feature flags enable gradual rollouts, A/B testing, and quick feature toggling.
python
from flask import Flask, request, jsonify
import redis
import json
app = Flask(__name__)
class FeatureFlagManager:
def __init__(self, redis_client):
self.redis = redis_client
self.cache_key = "feature_flags"
def set_feature(self, name, enabled, percentage=100, users=None):
"""Set a feature flag with optional percentage rollout or specific users"""
flags = self.get_all_flags()
flags[name] = {
'enabled': enabled,
'percentage': percentage,
'users': users or [],
'updated_at': datetime.utcnow().isoformat()
}
self.redis.set(self.cache_key, json.dumps(flags))
def get_all_flags(self):
"""Get all feature flags"""
flags_data = self.redis.get(self.cache_key)
return json.loads(flags_data) if flags_data else {}
def is_enabled(self, name, user_id=None):
"""Check if a feature is enabled for the current user/request"""
flags = self.get_all_flags()
if name not in flags:
return False # Feature is disabled by default
flag = flags[name]
if not flag['enabled']:
return False
# Check if user is in specific allowlist
if user_id and user_id in flag.get('users', []):
return True
# Percentage-based rollout
if flag['percentage'] < 100 and user_id:
user_hash = hash(f"{name}_{user_id}")
return (user_hash % 100) < flag['percentage']
return flag['enabled']
def get_flags_for_user(self, user_id):
"""Get all features enabled for a specific user"""
flags = self.get_all_flags()
user_flags = {}
for name, config in flags.items():
user_flags[name] = self.is_enabled(name, user_id)
return user_flags
# Initialize feature flag manager
feature_manager = FeatureFlagManager(redis.Redis())
# Feature flag decorator
def feature_flag(flag_name):
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
user_id = getattr(request, 'user_id', None)
if not feature_manager.is_enabled(flag_name, user_id):
return jsonify({
"error": "Feature not available",
"feature": flag_name
}), 403
return f(*args, **kwargs)
return decorated_function
return decorator
# Context processor to make feature flags available in templates
@app.context_processor
def inject_feature_flags():
user_id = getattr(request, 'user_id', None)
return {
'feature_flags': feature_manager.get_flags_for_user(user_id)
}
# Usage in views
@app.route('/new-feature')
@feature_flag('new_ui_design')
def new_feature():
return render_template('new_design.html')
@app.route('/old-feature')
def old_feature():
return render_template('old_design.html')
# A/B testing with feature flags
@app.route('/marketing-page')
def marketing_page():
user_id = get_user_id(request)
if feature_manager.is_enabled('new_marketing_copy', user_id):
return render_template('marketing_v2.html')
else:
return render_template('marketing_v1.html')
# Admin endpoints to manage feature flags
@app.route('/admin/features', methods=['GET', 'POST'])
def manage_features():
if request.method == 'POST':
data = request.get_json()
for feature, config in data.items():
feature_manager.set_feature(
name=feature,
enabled=config['enabled'],
percentage=config.get('percentage', 100),
users=config.get('users', [])
)
return jsonify({"status": "updated"})
else:
return jsonify(feature_manager.get_all_flags())
# Gradual rollout example
def enable_feature_gradually(feature_name, duration_hours=24):
"""Gradually roll out a feature over specified duration"""
start_time = datetime.utcnow()
interval_minutes = 10 # Update every 10 minutes
def rollout_step():
elapsed = (datetime.utcnow() - start_time).total_seconds()
total_duration = duration_hours * 3600
percentage = min(100, int((elapsed / total_duration) * 100))
feature_manager.set_feature(feature_name, True, percentage)
if percentage < 100:
# Schedule next rollout step
threading.Timer(
interval_minutes * 60,
rollout_step
).start()
rollout_step()
# Usage
enable_feature_gradually('new_payment_system', duration_hours=48)
These advanced questions cover sophisticated patterns and real-world scenarios that experienced Django and Flask developers encounter in production environments. They demonstrate deep understanding of web application architecture, performance optimization, and production readiness.
Comments (0)
No comments yet. Be the first to comment!