Flask Interview Questions and Answers

Answer

Flask() returns an object that is, at its core, a WSGI application: it implements __call__(environ, start_response), so any WSGI server (gunicorn, uWSGI, Waitress) can serve it without knowing anything about Flask. Around that callable it holds the url_map (a Werkzeug Map of routing rules), the view_functions dictionary, the config dict, the Jinja2 environment, the error handler registry, and the before/after/teardown hook lists. Every decorator you write is just mutating one of those registries at import time.

The first positional argument, import_name, is almost always __name__. Flask uses it to compute root_path by locating the module on disk, and root_path is what determines where templates/ and static/ are looked up, where the instance folder sits, and how extensions resolve relative resources. If you pass a wrong or dynamic value you get confusing TemplateNotFound errors on a machine where the tutorial worked.

If your app lives inside a package, __name__ evaluates to the package name and Flask resolves the package directory, which is exactly what you want. You can override the defaults explicitly with Flask(__name__, template_folder='../templates', static_folder='assets', instance_relative_config=True), and in larger projects that explicitness is worth it. Interviewers ask this because a candidate who says '__name__ is just a convention' has never debugged a packaging problem, whereas the correct answer connects a one-line constructor to how the framework finds files at runtime.

from flask import Flask

app = Flask(
    __name__,
    static_folder='assets',
    static_url_path='/cdn',
    instance_relative_config=True,
)

print(app.root_path)        # absolute dir of the module/package
print(app.instance_path)    # <root>/instance, for local secrets and sqlite
print(app.url_map)          # Map([<Rule '/cdn/<filename>' ...>])

@app.get('/health')
def health():
    return {'status': 'ok'}

Answer

Flask has two separate contexts and mixing them up is the single most common source of RuntimeError in production code. The application context is bound to one Flask app instance and exposes current_app and g. The request context is bound to one incoming HTTP request and exposes request and session.

When a WSGI request arrives, Flask pushes an application context (if one is not already active) and then a request context, runs the view, and pops both in reverse order. That is why current_app works inside a view even though you never imported your app object: the proxy resolves to whatever app owns the currently pushed context. Outside a request, for example in a Celery task, a CLI script, or a background thread, no context exists, so touching current_app.config or db.session raises 'Working outside of application context'.

The fix is to push one explicitly with `with app.app_context():`. For code that needs a fake request (URL building with url_for, testing a form parser) you push `with app.test_request_context('/x?y=1'):`, which pushes both contexts. A detail that trips people up: g is tied to the application context, not the request, so it is cleared when the app context pops, and in the normal request lifecycle that happens once per request. Since Flask 2.3 the old _app_ctx_stack and _request_ctx_stack objects are gone, so any StackOverflow snippet referencing them is dead code.

from flask import current_app, g, url_for

# Outside a request this raises RuntimeError
def warm_cache(app):
    with app.app_context():
        ttl = current_app.config['CACHE_TTL']
        g.started = True
        return ttl

# url_for needs a request context (or SERVER_NAME + app context)
def build_link(app):
    with app.test_request_context('/'):
        return url_for('health', _external=True)

Answer

All four are werkzeug.local.LocalProxy objects. A LocalProxy holds a lookup function and forwards every attribute access, item access, and dunder call to whatever object that function returns right now. Since Werkzeug 2.0 the underlying storage is contextvars.ContextVar, not a thread-local, which matters because ContextVars are also correct under asyncio tasks and greenlets, whereas plain thread-locals were not.

Practically this means each worker thread, each gevent greenlet, and each async task sees its own request object with no locking and no leakage, so yes, they are safe for concurrent requests within one process. The failure modes are subtle. First, because they are proxies, `isinstance(request, Request)` is False and `type(request)` is LocalProxy, so code that type-checks needs `request._get_current_object()`.

Second, capturing a proxy in a closure that outlives the request (a thread you spawn, a Celery task argument, a cached lambda) leaves you holding a proxy that resolves to nothing later and raises RuntimeError. Always call _get_current_object() or copy the plain values you need before handing work to another thread. Third, g is not a global cache.

It is reset per application context, so using it to memoise an expensive lookup across requests silently does nothing in production and looks like it works in a single-request test. Interviewers love that last one because it separates people who read the docs from people who copied a pattern.

import threading
from flask import request, g, current_app

@app.post('/webhook')
def webhook():
    # WRONG: the proxy will be unbound when the thread runs
    # threading.Thread(target=handle, args=(request,)).start()

    payload = request.get_json(silent=True) or {}
    real_app = current_app._get_current_object()

    def handle(app, data):
        with app.app_context():
            app.logger.info('async handle %s', data.get('id'))

    threading.Thread(target=handle, args=(real_app, payload), daemon=True).start()
    return '', 202

Answer

Each @app.route call adds a werkzeug.routing.Rule to app.url_map. At request time Werkzeug matches the path against that Map, which is compiled into a sorted structure, not scanned linearly, so route count is not a performance concern until you reach thousands of rules. A converter in angle brackets both matches and coerces: <int:user_id> matches only digits and hands the view an int, <float:>, <uuid:>, <path:> (matches slashes, useful for file paths), and the default <string:> (any text without a slash) round out the built-ins.

Converters accept arguments, so <int(min=1):page> rejects page=0 with a 404 rather than letting bad input reach your handler. If no rule matches you get 404; if the path matches but the method does not, Werkzeug returns 405 with an Allow header, which is why you should never write a catch-all route that swallows method errors. Two behaviours regularly cause bugs.

First, strict_slashes: a rule defined as '/items/' will 308-redirect '/items' to '/items/', but a rule defined as '/items' returns 404 for '/items/'. That asymmetry breaks API clients that append slashes, so most teams either standardise on no trailing slash or set app.url_map.strict_slashes = False. Second, endpoint names default to the function name and must be unique across the app, so two blueprints with a view called index are fine (they become 'admin.index' and 'shop.index') but two same-named views on the app object raise AssertionError at import time.

from flask import Flask

app = Flask(__name__)
app.url_map.strict_slashes = False

@app.get('/users/<int(min=1):user_id>')
def get_user(user_id: int):
    return {'id': user_id, 'type': type(user_id).__name__}  # int

@app.get('/files/<path:relpath>')
def get_file(relpath: str):
    return {'path': relpath}   # matches a/b/c.txt

@app.route('/legacy', methods=['GET', 'POST'], endpoint='legacy_v1')
def legacy():
    return ''

for rule in app.url_map.iter_rules():
    print(rule.endpoint, rule.rule, sorted(rule.methods))

Answer

url_for('endpoint', **values) reverses the routing table: it takes an endpoint name and produces the URL. That decoupling matters for three concrete reasons. Changing a URL prefix (moving /api/users to /api/v2/users, or mounting the app under a subpath with SCRIPT_NAME behind nginx) updates every generated link automatically.

It handles quoting, so url_for('search', q='hello world & more') escapes correctly while string concatenation does not. And it respects APPLICATION_ROOT and the blueprint prefix, so a blueprint registered at url_prefix='/admin' generates '/admin/users' without you repeating the prefix. Any keyword argument that is not part of the rule becomes a query parameter, which is the cleanest way to build filtered links.

BuildError, raised as werkzeug.routing.BuildError, has three usual causes: the endpoint name is wrong (remember it is the function name, or 'blueprint_name.function_name' for blueprint views, not the URL); a required converter argument is missing, for example calling url_for('get_user') without user_id; or you are calling it outside a request context without SERVER_NAME configured, since Flask cannot know the host to build an absolute URL. Inside a blueprint view you can use a leading dot for a relative reference, url_for('.detail', id=3), which resolves within the current blueprint and survives renaming the blueprint. For external links pass _external=True, and set PREFERRED_URL_SCHEME='https' so emails and webhooks do not go out with http:// links.

from flask import Blueprint, url_for

bp = Blueprint('users', __name__, url_prefix='/api/users')

@bp.get('/<int:user_id>')
def detail(user_id):
    return {
        'self': url_for('.detail', user_id=user_id),          # /api/users/7
        'next': url_for('.detail', user_id=user_id + 1),
        'search': url_for('.search', q='data engineer', page=2),
        'absolute': url_for('.detail', user_id=user_id, _external=True),
    }

@bp.get('/search')
def search():
    return []

# app.config['SERVER_NAME'] = 'api.example.com'
# app.config['PREFERRED_URL_SCHEME'] = 'https'

Answer

Flask converts whatever a view returns into a Response via make_response. A str or bytes becomes a 200 text/html response. A dict or a list (list support arrived in Flask 2.2) is serialised to JSON with the app's JSON provider and gets application/json automatically, which is why most modern Flask APIs simply `return {'id': 1}`.

A tuple can be (body, status), (body, headers), or (body, status, headers), so `return {'error': 'not found'}, 404` is idiomatic. A Response object is returned as-is, and a WSGI callable is invoked. jsonify still has a job: it applies the app's JSONProvider (app.json) which handles datetime, UUID, Decimal, and dataclasses through your configured serialiser, it lets you pass keyword arguments as a shortcut (jsonify(status='ok', count=3)), and it returns a real Response you can mutate before returning, for example to set a Cache-Control header or a cookie. Since Flask 2.2 the JSON layer is pluggable through app.json, so setting app.json.sort_keys = False or swapping in an orjson-backed provider changes serialisation everywhere including bare dict returns. One production detail: returning a top-level list used to be blocked for JSON hijacking reasons, that concern is obsolete for modern browsers and Flask now allows it, but many API style guides still wrap collections in an object so the payload can grow pagination metadata later without a breaking change.

from flask import jsonify, make_response, Response
import json

@app.get('/a')
def a():
    return {'ok': True}                      # 200 application/json

@app.get('/b')
def b():
    return {'error': 'gone'}, 410            # tuple with status

@app.get('/c')
def c():
    resp = jsonify(items=[], total=0)
    resp.headers['Cache-Control'] = 'public, max-age=60'
    resp.set_cookie('seen', '1', httponly=True, samesite='Lax')
    return resp

@app.get('/d')
def d():
    body = json.dumps({'raw': 1})
    return Response(body, status=201, mimetype='application/json')

Answer

The request object exposes each input source separately. request.args is a MultiDict of query string values, request.form is a MultiDict of application/x-www-form-urlencoded or multipart form fields, request.files is a MultiDict of FileStorage objects, request.get_json() parses a JSON body, request.data gives raw bytes, and request.values chains args and form (avoid it, the merge order surprises people). Because these are MultiDicts, .get('k') returns the first value and .getlist('k') returns all of them, which is how you handle ?tag=python&tag=flask. Type coercion is built in: request.args.get('page', 1, type=int) returns 1 when the parameter is absent or unparseable rather than raising, which is safer than int(request.args['page']).

The JSON path has two traps. request.json (the property) raises a 415 UnsupportedMediaType if the Content-Type header is not application/json, and a 400 BadRequest if the body will not parse, so clients that post JSON with the wrong header get a confusing error. Use request.get_json(silent=True) to get None instead of an exception, or get_json(force=True) to parse regardless of the header when you control the client. And indexing directly, request.get_json()['email'], raises a KeyError that becomes a 500, not a 400.

Every real API validates through marshmallow, pydantic, or an explicit schema check instead. For files, always run werkzeug.utils.secure_filename on the client-supplied name and never trust the reported content type.

from flask import request
from werkzeug.utils import secure_filename

@app.post('/jobs')
def create_job():
    page = request.args.get('page', 1, type=int)
    tags = request.args.getlist('tag')          # ?tag=python&tag=flask

    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        return {'error': 'json body required'}, 400

    title = (body.get('title') or '').strip()
    if not title:
        return {'error': 'title is required'}, 422

    jd = request.files.get('jd')
    name = secure_filename(jd.filename) if jd else None

    return {'page': page, 'tags': tags, 'title': title, 'file': name}, 201

Answer

A Blueprint is a deferred registration object. Decorating a view with @bp.get does not create a route, it records an operation that gets replayed onto a real app when you call app.register_blueprint(bp). That indirection is what makes the application factory pattern work: your modules import blueprints, not the app, so there is no circular import and one blueprint can be registered onto multiple apps (a public app and an internal admin app) or registered twice under different prefixes.

Blueprints carry their own url_prefix, subdomain, static_folder, template_folder, error handlers, and before_request hooks, which is how you scope authentication to a whole section of the API with a single @bp.before_request. Endpoint names are namespaced, so a view named list_items in blueprint 'jobs' becomes 'jobs.list_items' and url_for('.list_items') resolves relatively. Flask 2.0 added nested blueprints: parent.register_blueprint(child) composes prefixes, so an api blueprint at /api containing a v1 blueprint at /v1 containing users at /users produces /api/v1/users, and the endpoint becomes 'api.v1.users.detail'.

Two gotchas matter in interviews. Registration order is fixed at register time, so mutating a blueprint after registering it has no effect and Flask raises an error if you try to add routes to an already-registered blueprint in recent versions. And blueprint-level errorhandler for 404 does not fire for unmatched URLs, because Flask has not decided which blueprint owns an unrouted path, so a global 404 handler is still required.

from flask import Blueprint, abort, g

api = Blueprint('api', __name__, url_prefix='/api')
v1 = Blueprint('v1', __name__, url_prefix='/v1')
jobs = Blueprint('jobs', __name__, url_prefix='/jobs')

@v1.before_request
def require_key():
    if not request.headers.get('X-Api-Key'):
        abort(401)

@jobs.get('/<int:job_id>')
def detail(job_id):
    return {'id': job_id}

v1.register_blueprint(jobs)
api.register_blueprint(v1)
# app.register_blueprint(api)
# -> GET /api/v1/jobs/7, endpoint 'api.v1.jobs.detail'

Answer

The factory pattern replaces a module-level `app = Flask(__name__)` with a function `create_app(config=None)` that builds, configures, and returns an app. Three problems disappear. Circular imports: with a global app, views must import app and app must import views, so people resort to bottom-of-file imports and fragile ordering.

With a factory, views live in blueprints that import nothing from the app module. Testing: a global app is configured once at import time, so you cannot spin up a second app with a different database URI in the same process. With a factory, every test gets a fresh app pointed at a throwaway database.

Configuration timing: extensions that read config at construction time misbehave when config is loaded after import, whereas a factory guarantees config is set before init_app runs. The rule that makes it work is the two-phase extension pattern. Extension objects (SQLAlchemy, Migrate, Cache, Limiter, JWTManager) are constructed at module scope with no app, then bound inside the factory with ext.init_app(app).

That is why every serious Flask extension exposes init_app. In recent versions you run it with `flask --app 'myapp:create_app' run --debug`, and the CLI also auto-discovers create_app in app.py or wsgi.py. For gunicorn you expose a module-level `app = create_app()` in wsgi.py and point gunicorn at 'wsgi:app'. Interviewers ask for this because virtually every Flask codebase past 500 lines converges on it, and candidates who cannot explain init_app usually have global-state bugs in their tests.

# extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

# __init__.py
from flask import Flask
from .extensions import db, migrate
from .jobs.routes import jobs_bp

def create_app(config_object='config.Production'):
    app = Flask(__name__, instance_relative_config=True)
    app.config.from_object(config_object)
    app.config.from_prefixed_env()          # FLASK_SQLALCHEMY_DATABASE_URI=...

    db.init_app(app)
    migrate.init_app(app, db)
    app.register_blueprint(jobs_bp)

    @app.get('/healthz')
    def healthz():
        return {'status': 'ok'}

    return app

# wsgi.py
# from myapp import create_app
# app = create_app()

Answer

The modern entry point is the flask CLI with an explicit --app. `flask --app app run --debug` starts the Werkzeug development server with the reloader and the interactive debugger. FLASK_APP still works as an environment variable, but FLASK_ENV was removed in Flask 2.3 and setting it now does nothing, which catches out anyone copying older tutorials. Debug mode is controlled by the --debug flag or FLASK_DEBUG=1.

The --app value can be a module ('app'), a module and attribute ('myapp:app'), or a factory call ('myapp:create_app()' with arguments if needed), and Flask auto-discovers app.py or wsgi.py containing app, application, or create_app. Other useful CLI pieces: `flask routes` prints the full url_map with endpoints and methods and is the fastest way to debug a 404, `flask shell` opens a REPL with an application context already pushed, and `flask --app app run --host 0.0.0.0 --port 8000 --cert=adhoc` gives you a self-signed HTTPS dev server for testing secure cookies. Two safety points that interviewers listen for.

First, the Werkzeug development server is single-process, not hardened, and must never serve production traffic, Flask itself prints a warning saying so. Second, debug mode enables the Werkzeug interactive debugger, which lets anyone who can reach the page execute arbitrary Python in your process. It is PIN-protected by default, but the PIN is derived from predictable machine attributes and has been bypassed in the wild. Running with debug=True on a public host is a full remote code execution vulnerability.

# Development
export FLASK_APP='myapp:create_app'
flask run --debug --port 8000

# Or without env vars
flask --app 'myapp:create_app()' run --debug

# Inspect the routing table (fastest 404 debug)
flask --app myapp routes

# REPL with an application context already pushed
flask --app myapp shell

# Production: never flask run
gunicorn 'myapp:create_app()' \
  --bind 0.0.0.0:8000 \
  --workers 5 --worker-class gthread --threads 4 \
  --timeout 30 --graceful-timeout 30 \
  --access-logfile - --error-logfile -

Answer

app.config is a dict subclass with loaders attached. from_object('config.Production') imports a module or uses a class and copies its uppercase attributes, which is the most common pattern because config classes can inherit from a Base. from_pyfile('local.cfg') executes a Python file, useful for a file outside version control. from_envvar('APP_SETTINGS') reads a path from an environment variable and delegates to from_pyfile. from_mapping(**kwargs) sets values directly. from_file('config.toml', tomllib.load) loads any format you have a parser for. And from_prefixed_env(), added in Flask 2.2, pulls every FLASK_-prefixed environment variable, strips the prefix, and parses the value as JSON when possible, so FLASK_MAX_PAGE_SIZE=50 becomes an int and FLASK_FEATURE_FLAGS='{"beta": true}' becomes a dict. Nested keys work through double underscores.

The instance folder is a directory next to your package (or next to the app module) that is deliberately not part of your importable code and should be gitignored. Passing instance_relative_config=True makes from_pyfile paths resolve relative to app.instance_path, which is the standard place for a local SQLite file, dev certificates, or a secrets file on a single-server deployment. In real deployments in India most teams pull secrets from AWS Secrets Manager, Vault, or Infisical into environment variables at container start and rely on from_object for defaults plus from_prefixed_env for overrides.

The key discipline: validate at boot. A missing SECRET_KEY or DATABASE_URI should crash the container immediately, not produce a 500 an hour into the traffic peak.

import os

class Base:
    JSON_SORT_KEYS = False
    SQLALCHEMY_ENGINE_OPTIONS = {'pool_pre_ping': True, 'pool_recycle': 280}
    MAX_CONTENT_LENGTH = 8 * 1024 * 1024

class Production(Base):
    DEBUG = False
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_SAMESITE = 'Lax'

def load_config(app):
    app.config.from_object('config.Production')
    app.config.from_prefixed_env()   # FLASK_SECRET_KEY, FLASK_DATABASE_URI...

    required = ('SECRET_KEY', 'SQLALCHEMY_DATABASE_URI')
    missing = [k for k in required if not app.config.get(k)]
    if missing:
        raise RuntimeError('missing config: ' + ', '.join(missing))

Answer

Flask reuses Werkzeug's exception hierarchy. Every HTTP error is a subclass of werkzeug.exceptions.HTTPException with a code and description, and abort(404) or abort(403, description='...') raises the right one. Because they are exceptions, they propagate cleanly out of service functions, so a repository layer can abort(404) without knowing about HTTP responses, though most teams prefer raising a domain exception and mapping it at the edge. @app.errorhandler accepts either a status code, an exception class, or HTTPException itself for a catch-all.

Registration is specificity-based: Flask picks the handler for the most specific matching class in the MRO, so a handler for ValueError wins over a handler for Exception. A handler for HTTPException catches every 4xx and 5xx that Werkzeug raises, which is exactly how you produce uniform JSON errors for an API instead of Werkzeug's default HTML page. Register a handler for Exception too, and it will catch unhandled exceptions and turn them into a 500, but only when PROPAGATE_EXCEPTIONS is off, in debug mode and in tests Flask re-raises so you can see the traceback.

A subtlety worth knowing: returning a value from an error handler follows the same rules as a view, so `return {'error': e.description}, e.code` works. And errors raised inside after_request or teardown handlers do not go through your error handlers the same way, teardown functions receive the exception as an argument and must not raise, or you lose the original error entirely.

from flask import jsonify
from werkzeug.exceptions import HTTPException

class DomainError(Exception):
    status = 422
    def __init__(self, code, message):
        self.code, self.message = code, message

@app.errorhandler(HTTPException)
def on_http_error(e):
    return jsonify(error=e.name, detail=e.description, status=e.code), e.code

@app.errorhandler(DomainError)
def on_domain_error(e):
    return jsonify(error=e.code, detail=e.message), e.status

@app.errorhandler(Exception)
def on_unhandled(e):
    app.logger.exception('unhandled error')
    return jsonify(error='internal_error'), 500

Answer

before_request functions run in registration order before the view. If any of them returns a non-None value, Flask short-circuits: the view never runs and that value becomes the response, which is how auth guards and maintenance-mode switches are implemented. after_request functions receive the response object, must return it (a very common bug is forgetting the return, which produces a cryptic TypeError), and run in reverse registration order so wrapping behaves like a stack. Critically, after_request is skipped when an unhandled exception occurs, so it is the wrong place to release resources. teardown_request always runs, exception or not, and receives the exception as its argument, but it runs after the response has been created so it cannot modify it. teardown_appcontext runs when the application context pops, which for a normal request is just after teardown_request, and is the correct hook for closing database sessions, returning connections to a pool, or flushing a per-request buffer.

The order for a successful request is: before_request handlers, the view, after_request handlers in reverse, teardown_request, teardown_appcontext. Blueprint-scoped hooks (@bp.before_request) run only for requests routed to that blueprint and execute after app-level before_request handlers. There is also before_app_request on a blueprint, which registers an app-wide hook from inside a blueprint module. Interviewers ask this because getting it wrong produces the two classic Flask leaks: a database session never closed because cleanup lived in after_request, and a response object silently replaced by None.

import time, uuid
from flask import g, request

@app.before_request
def start_timer():
    g.t0 = time.perf_counter()
    g.request_id = request.headers.get('X-Request-Id') or uuid.uuid4().hex

@app.after_request
def add_headers(response):
    took = (time.perf_counter() - g.get('t0', 0)) * 1000
    response.headers['X-Request-Id'] = g.get('request_id', '')
    response.headers['Server-Timing'] = 'app;dur=%.1f' % took
    return response                      # forgetting this breaks everything

@app.teardown_appcontext
def close_session(exc=None):
    sess = g.pop('db_session', None)
    if sess is not None:
        sess.rollback() if exc else sess.commit()
        sess.close()

Answer

Flask configures Jinja2 with select_autoescape so that templates ending in .html, .htm, .xml and .xhtml escape every variable by default. Rendering {{ user.bio }} converts <, >, &, quotes and single quotes to entities, which neutralises stored XSS. The escaping is done by markupsafe: any object that implements __html__ (a Markup instance) is inserted verbatim, everything else is escaped.

That is the whole security model, and it fails in exactly four ways. First, the |safe filter and {% autoescape false %} blocks explicitly disable it, so any user-controlled value passed through |safe is an XSS hole. Second, files with an extension outside the autoescape list, for example a .txt or .j2 template, are not escaped at all unless you extend the list or force it.

Third, render_template_string with user input is server-side template injection, not just XSS: an attacker who controls the template text can reach Python objects through Jinja's attribute traversal and in many configurations achieve code execution, so never build a template string from request data. Fourth, escaping is HTML-context aware only in the sense that it escapes HTML, it does not make a value safe inside a <script> block, inside a URL attribute (javascript: URLs survive escaping), or inside inline CSS. For JSON embedded in a page use the tojson filter, which escapes for a script context correctly. In Flask 3.x, markupsafe.Markup and escape are imported from markupsafe directly, the flask.Markup and flask.escape re-exports were removed.

from markupsafe import Markup, escape
from flask import render_template

@app.get('/profile/<username>')
def profile(username):
    # safe: autoescaped in profile.html
    return render_template('profile.html', username=username)

# Building trusted markup explicitly
def badge(label, count):
    return Markup('<span class=badge>{}: {}</span>').format(
        escape(label), int(count)
    )

# profile.html
# <h1>Hello {{ username }}</h1>            {# escaped #}
# <div>{{ badge_html }}</div>              {# Markup, inserted as-is #}
# <script>const cfg = {{ config_dict|tojson }};</script>

Answer

Flask's default session is client-side. The session dict is serialised to JSON, signed with itsdangerous using SECRET_KEY and an HMAC, and stored in a cookie. The critical point candidates get wrong: it is signed, not encrypted.

Anyone can base64-decode the cookie and read every value, so never put anything confidential in it. What signing guarantees is integrity, a user cannot change their own role to admin because the HMAC will not verify, and Flask will silently discard a tampered cookie rather than raise. The practical limits follow from being a cookie.

Browsers cap cookies at roughly 4KB, and Werkzeug logs a warning when you exceed it, but a session that grows past the limit is silently dropped by the browser, which presents as users randomly getting logged out. There is no server-side invalidation: rotating SECRET_KEY logs out every user at once, and you cannot revoke a single session because there is nothing on the server to delete. PERMANENT_SESSION_LIFETIME only sets the cookie expiry and the signature max_age, so an old cookie stops being accepted after that window.

For anything needing revocation, large session payloads, or a shared session across services, move to a server-side store with Flask-Session backed by Redis, keeping only an opaque session id in the cookie. Whatever you choose, set SESSION_COOKIE_SECURE=True, SESSION_COOKIE_HTTPONLY=True (default) and SESSION_COOKIE_SAMESITE='Lax' or 'Strict' in production, and set session.permanent plus a lifetime rather than relying on browser session cookies.

from datetime import timedelta
from flask import session

app.config.update(
    SECRET_KEY=os.environ['SECRET_KEY'],
    SESSION_COOKIE_SECURE=True,
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SAMESITE='Lax',
    PERMANENT_SESSION_LIFETIME=timedelta(days=7),
)

@app.post('/login')
def login():
    session.clear()                 # prevent session fixation
    session['uid'] = 42             # never store PII or secrets here
    session.permanent = True
    return {'ok': True}

@app.post('/logout')
def logout():
    session.clear()
    return {'ok': True}

Answer

By default Flask registers a /static/<path:filename> route pointing at the static folder next to your app module, served by send_from_directory. You change the location with static_folder and the URL with static_url_path, and blueprints can have their own static folders. send_from_directory is the safe API: it resolves the requested path against a base directory and refuses anything that escapes it, which protects against ../../etc/passwd traversal. Building a path manually with os.path.join and calling send_file is the classic Flask path-traversal vulnerability, because os.path.join with an absolute second argument discards the first.

Flask does handle caching reasonably, it sets Last-Modified and an ETag from the file stat and honours conditional requests, and SEND_FILE_MAX_AGE_DEFAULT controls Cache-Control max-age (None by default in recent versions, meaning conditional requests rather than blind caching). Even so, serving static assets from Flask in production wastes a worker. Every asset request occupies a gunicorn worker slot for the duration of the transfer, so a page with forty assets on a slow mobile connection can starve a pool sized for API traffic.

Put nginx, CloudFront, or S3 in front and let Flask handle only dynamic routes. In containerised deployments the common pattern in India is to build assets at image build time, push them to S3 with a content hash in the filename, and serve them through CloudFront, with Flask generating asset URLs from a manifest. If you must serve files through Flask, use send_file with conditional=True and consider X-Sendfile or X-Accel-Redirect so nginx does the actual transfer.

from flask import send_from_directory, send_file, abort

# Safe: rejects traversal outside /srv/reports
@app.get('/reports/<path:name>')
def report(name):
    if not name.endswith('.pdf'):
        abort(404)
    return send_from_directory('/srv/reports', name, max_age=300)

# Private file served by nginx after Flask authorises it
@app.get('/private/<int:doc_id>')
def private(doc_id):
    if not current_user_can_read(doc_id):
        abort(403)
    resp = app.response_class()
    resp.headers['X-Accel-Redirect'] = '/protected/%d.pdf' % doc_id
    resp.headers['Content-Type'] = 'application/pdf'
    return resp

Answer

Flask embeds Click, and app.cli is a Click group. Decorating a function with @app.cli.command('name') registers a subcommand that runs with an application context already pushed, so current_app, the database session, and extensions all work without you managing contexts. Click decorators compose normally, so @click.argument, @click.option, and @click.confirmation_option give you typed arguments, help text, and prompts for free.

For blueprint-scoped commands use @bp.cli.command, which nests them under the blueprint name, giving you `flask users create-admin`. Commands are the right place for seed data, one-off backfills, index rebuilds, cache warming, and report exports, because they reuse the same models and config as the web app rather than duplicating connection logic in a script. Use @with_appcontext explicitly only when registering a bare Click command onto app.cli with add_command, since that path does not push the context automatically. flask shell is the second half of the story: it opens a Python REPL with the app context pushed and the shell context populated.

By default it injects the app object, and @app.shell_context_processor lets you add anything else, so you can drop into a shell and immediately query models without ten lines of imports. On a production box this is the fastest way to inspect real data during an incident, though the disciplined version is a read-only replica connection. Both features are also testable: app.test_cli_runner() invokes commands in-process so you can assert on exit codes and output in pytest.

import click
from flask.cli import with_appcontext
from .extensions import db
from .models import User

@app.cli.command('create-admin')
@click.argument('email')
@click.option('--force', is_flag=True, help='Overwrite if the user exists')
def create_admin(email, force):
    existing = db.session.scalar(db.select(User).filter_by(email=email))
    if existing and not force:
        raise click.ClickException('user already exists: %s' % email)
    user = existing or User(email=email)
    user.is_admin = True
    db.session.add(user)
    db.session.commit()
    click.secho('admin ready: %s' % email, fg='green')

@app.shell_context_processor
def shell_ctx():
    return {'db': db, 'User': User}

Answer

Pick Django when the product is content-and-admin heavy and you want batteries included: ORM, migrations, auth, permissions, forms, and a generated admin panel that non-engineers can use. Most Indian marketplaces and CMS-driven products start there because the Django admin alone saves a quarter of internal tooling work. Pick FastAPI for a greenfield high-concurrency API where the workload is I/O bound (calling model endpoints, third-party APIs, or many small database queries), you want async/await natively, and you want Pydantic validation plus generated OpenAPI documentation without extra libraries.

Pick Flask when you want a small, explicit WSGI service and full control over every dependency: a synchronous internal API, an ML inference wrapper where the heavy work releases the GIL inside NumPy or PyTorch anyway, a webhook receiver, a legacy service that must integrate with an existing SQLAlchemy codebase, or anything you want to keep readable by an engineer who does not know the framework. Flask is also the pragmatic answer inside large enterprises in India where the deployment platform is standardised on WSGI and gunicorn, where the ops team knows how to tune it, and where adding an ASGI stack means new runbooks. The honest counterpoint you should volunteer in an interview: for a brand-new public API with heavy concurrency and a schema-first contract, FastAPI is usually the better default in 2026, and saying so shows judgement rather than framework loyalty. Flask's real edge is longevity, stability, and the fact that the whole framework is small enough to reason about completely.

Answer

That error means a LocalProxy (current_app, g, or anything an extension resolves through them, such as Flask-SQLAlchemy's db.session or Flask-Mail's mail.send) was accessed while no application context was pushed. Four situations produce it in real codebases. A Celery task: the worker process imports your models but never runs through the WSGI stack, so db.session has no engine bound.

A background thread or an APScheduler job spawned from a view: the context belongs to the request thread and is popped when that request finishes. A module-level statement, for example `DEFAULT_TTL = current_app.config['TTL']` evaluated at import time, before any app exists. And a script or migration helper run with plain `python manage.py`.

The fix is always the same shape: obtain the app object and push a context with `with app.app_context():`. For Celery the clean pattern is a custom Task base class whose __call__ wraps super().__call__ in an app context, created inside your factory so the app is the configured one. For threads, pass current_app._get_current_object() into the thread rather than the proxy.

For module-level constants, move the lookup inside a function or a factory. There is a sibling error, 'Working outside of request context', raised when you touch request or session outside a request, and the fix there is app.test_request_context() in tests or restructuring so the value is passed as an argument. A useful diagnostic: has_app_context() and has_request_context() let library code branch safely instead of crashing.

from celery import Celery, Task

def celery_init_app(app):
    class FlaskTask(Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery = Celery(app.name, task_cls=FlaskTask)
    celery.config_from_object(app.config['CELERY'])
    celery.set_default()
    app.extensions['celery'] = celery
    return celery

# Now this works inside a worker
@shared_task
def send_digest(user_id):
    user = db.session.get(User, user_id)      # needs app context
    current_app.logger.info('digest for %s', user.email)

Answer

Flask-SQLAlchemy 3.x realigned the extension with SQLAlchemy 2.0's select() API and tightened its context rules. The biggest behavioural change is that db.session now requires an active application context for every operation, including in scripts, so code that used to work at module level or inside a bare Celery task now raises RuntimeError. SQLALCHEMY_DATABASE_URI (or SQLALCHEMY_BINDS) is required at init_app time and the extension no longer silently falls back to SQLite.

SQLALCHEMY_TRACK_MODIFICATIONS defaults to False and the noisy warning is gone. Model.query still exists as a legacy accessor, but the recommended style is db.session.execute(db.select(Model)) with .scalars() to get model instances, matching upstream SQLAlchemy 2.0 so your knowledge transfers to non-Flask projects. The extension adds Flask-specific conveniences: db.get_or_404(User, id), db.first_or_404(select), db.one_or_404(select), and db.paginate(select, per_page=20) which reads page and per_page from the query string and returns a Pagination object.

Typed declarative models with Mapped and mapped_column are supported and are the direction to move in for new code, since they give real type checking on attributes. Practical migration notes: engine options moved into SQLALCHEMY_ENGINE_OPTIONS as a dict, multiple binds are configured with a dict of URLs and a model's __bind_key__, and db.create_all() must be called inside an app context. Interviewers at product companies use this question to check whether you have upgraded a real codebase or only ever used tutorial-era 2.x syntax.

from sqlalchemy.orm import Mapped, mapped_column
from .extensions import db

class Candidate(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(db.String(254), unique=True, index=True)
    city: Mapped[str | None] = mapped_column(db.String(80))

# Modern query style
stmt = (db.select(Candidate)
          .where(Candidate.city == 'Bengaluru')
          .order_by(Candidate.id.desc()))

rows = db.session.execute(stmt).scalars().all()
one = db.session.scalar(db.select(Candidate).filter_by(email='a@b.com'))
user = db.get_or_404(Candidate, 7)
page = db.paginate(stmt, per_page=20, error_out=False)
print(page.items, page.total, page.has_next)

Answer

Flask-SQLAlchemy wraps a scoped_session whose scope function is tied to the application context, so every request gets its own Session and db.session inside a view always refers to that request's session. The extension registers a teardown_appcontext handler that calls session.remove() when the context pops, which closes the session and returns the connection to the engine's pool. That machinery is what keeps a normal request clean.

Leaks come from stepping outside it. If you push an app context manually in a long-running loop and never pop it, the session lives forever and holds a connection, so a script that iterates ten thousand records inside one context can hold one connection and an ever-growing identity map until it OOMs. If you spawn a thread inside a request and use db.session there, the scope function resolves to the same context and two threads share one Session, which is not thread-safe and produces intermittent InvalidRequestError or corrupted state.

If an exception escapes mid-transaction and nothing rolls back, the connection returns to the pool in a failed transaction state and the next user of it sees 'current transaction is aborted' on Postgres. The correct patterns are: commit or rollback explicitly around unit-of-work boundaries, use `with db.session.begin():` for automatic commit and rollback, call db.session.remove() at the end of each iteration in a batch job or use a fresh app context per chunk, and never share a Session across threads or processes. Under gunicorn with --preload, also dispose the engine after fork so children do not inherit a parent's sockets.

from .extensions import db

# Batch job: fresh context and session per chunk
def backfill(app, ids, chunk=500):
    for start in range(0, len(ids), chunk):
        with app.app_context():
            batch = ids[start:start + chunk]
            try:
                with db.session.begin():          # commit or rollback automatically
                    for row in db.session.scalars(
                        db.select(Candidate).where(Candidate.id.in_(batch))
                    ):
                        row.city = row.city.strip().title()
            finally:
                db.session.remove()               # return the connection

# After gunicorn --preload fork
def post_fork(server, worker):
    with app.app_context():
        db.engine.dispose()

Answer

Pool settings go in SQLALCHEMY_ENGINE_OPTIONS. pool_size is the number of persistent connections per engine (default 5), max_overflow is how many extra connections may be opened under load before callers block (default 10), pool_timeout is how long a caller waits for a free connection before raising TimeoutError (default 30 seconds), pool_recycle discards connections older than N seconds, and pool_pre_ping issues a cheap SELECT 1 before handing out a connection and transparently reconnects if it is dead. The arithmetic that matters in production: total connections equals workers times threads times (pool_size plus max_overflow), per process. Four gunicorn workers with 4 threads and the default pool can open 4 times 15, sixty connections, from one container, and Postgres on a small RDS instance caps out around a hundred.

Teams routinely exhaust the database because nobody multiplied. 'MySQL server has gone away' (error 2006) and Postgres 'server closed the connection unexpectedly' are the same underlying issue: something between your app and the database closed an idle connection while SQLAlchemy still believed it was live. The something is usually MySQL's wait_timeout (often 300 or 600 seconds), an AWS NLB or ALB idle timeout (350 seconds by default), or a firewall.

The fix is pool_recycle set safely below the shortest timeout in the path (280 seconds is a common choice) plus pool_pre_ping=True for correctness. If you sit behind PgBouncer in transaction pooling mode, set pool_size low, disable SQLAlchemy-side prepared statement caching where needed, and never rely on session-level state such as temporary tables.

app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'pool_size': 5,
    'max_overflow': 5,
    'pool_timeout': 10,        # fail fast instead of piling up
    'pool_recycle': 280,       # below MySQL wait_timeout / ALB 350s idle
    'pool_pre_ping': True,
    'connect_args': {'connect_timeout': 5},
}

# Budget check before you deploy
workers, threads = 5, 4
per_proc = 5 + 5
print('max db connections:', workers * per_proc)   # threads share the pool

Answer

N+1 happens when you load a collection with one query and then trigger a separate query per row by touching a lazily loaded relationship. Serialising fifty jobs and reading job.company.name inside the loop issues one query for the jobs and fifty for the companies. On a local SQLite database the endpoint feels fine, on production Postgres with two milliseconds of network round trip per query it becomes a hundred extra milliseconds and scales with page size.

It is the single most common performance defect in Flask APIs and interviewers ask it because the fix requires understanding SQLAlchemy loader strategies rather than adding a cache. The fixes are eager loading strategies applied per query: selectinload issues one additional IN query per relationship and is the right default for one-to-many, joinedload does a LEFT OUTER JOIN and suits many-to-one and one-to-one, subqueryload is mostly legacy, and raiseload('*') is the enforcement tool. Setting a relationship to lazy='raise' or applying raiseload turns an accidental lazy load into a loud InvalidRequestError instead of a silent extra query, which is how you stop regressions from creeping back in.

Detection is straightforward: enable SQLALCHEMY_RECORD_QUERIES and log get_recorded_queries() count per request, or use Flask-DebugToolbar locally, or add an after_request assertion in tests that fails when a single endpoint exceeds a query budget. In production, an OpenTelemetry SQLAlchemy instrumentation shows span counts per trace, and a trace with sixty database spans for one list endpoint is unmistakable.

from sqlalchemy.orm import selectinload, joinedload, raiseload

# N+1: one query + one per job
jobs = db.session.scalars(db.select(Job).limit(50)).all()
data = [{'title': j.title, 'company': j.company.name} for j in jobs]

# Fixed: 2 queries total
stmt = (db.select(Job)
          .options(joinedload(Job.company), selectinload(Job.skills))
          .limit(50))
jobs = db.session.scalars(stmt).unique().all()

# Enforce it
strict = db.select(Job).options(joinedload(Job.company), raiseload('*')).limit(50)

@app.after_request
def warn_on_query_count(resp):
    from flask_sqlalchemy.record_queries import get_recorded_queries
    n = len(get_recorded_queries())
    if n > 10:
        app.logger.warning('%s issued %d queries', request.path, n)
    return resp

Answer

Flask-Migrate is a thin Flask CLI wrapper around Alembic. `flask db init` creates the migrations directory with env.py wired to your Flask-SQLAlchemy metadata, `flask db migrate -m 'add city index'` compares your model metadata against the current database and writes a candidate revision, `flask db upgrade` applies it, `flask db downgrade` reverses it, and `flask db current` / `flask db history` show where you are. Each revision file has a revision id and a down_revision, forming a linked list, and the applied head is stored in the alembic_version table. Autogenerate is a comparison, not a source of truth, and it reliably misses several categories: column type changes on some backends unless you enable compare_type, server default changes unless compare_server_default is on, renamed columns (it emits a drop plus an add, which destroys data), constraints and indexes created outside the models, CHECK constraints, enum value additions on Postgres, and anything in a schema Alembic is not told to include.

It also cannot know your data, so a NOT NULL column added to a populated table needs a three-step migration: add nullable, backfill in a data migration, then alter to NOT NULL. Always read the generated file before committing. Two production disciplines separate seniors here: naming conventions on constraints (set them in MetaData so Alembic can find and drop them deterministically instead of relying on database-generated names), and never editing a migration that has already run in production, add a new one instead. For zero-downtime deploys, make migrations backward compatible with the currently running code, since the old pods keep serving during the rollout.

from sqlalchemy import MetaData
from flask_sqlalchemy import SQLAlchemy

convention = {
    'ix': 'ix_%(column_0_label)s',
    'uq': 'uq_%(table_name)s_%(column_0_name)s',
    'ck': 'ck_%(table_name)s_%(constraint_name)s',
    'fk': 'fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s',
    'pk': 'pk_%(table_name)s',
}
db = SQLAlchemy(metadata=MetaData(naming_convention=convention))

# migrations/env.py
# context.configure(..., compare_type=True, compare_server_default=True)

# Backward-compatible NOT NULL rollout
# 1) op.add_column('job', sa.Column('slug', sa.String(120), nullable=True))
# 2) op.execute("UPDATE job SET slug = lower(title) WHERE slug IS NULL")
# 3) op.alter_column('job', 'slug', nullable=False)

Answer

Flask deliberately ships no validation, so you choose. Three options dominate in 2026. Marshmallow is the classic: declare a Schema with typed fields and validators, call schema.load(request.get_json()) which raises ValidationError with a per-field error dict, and schema.dump(obj) to serialise models out.

Pydantic v2 is increasingly used even in Flask codebases because it is fast (the core is Rust), gives real type hints, and produces JSON Schema, so you validate with Model.model_validate(request.get_json()) and catch ValidationError. Flask-WTF plus WTForms remains the right answer for server-rendered HTML forms because it also handles CSRF and field rendering. Whichever you pick, the pattern is the same: validate at the boundary, convert the library's error into your API's error contract with a single @app.errorhandler, and never let raw request data reach a model constructor, because passing **payload into a SQLAlchemy model is a mass-assignment vulnerability that lets a caller set is_admin.

Add explicit length and range limits on every string and number, since an unbounded string field is a memory and storage attack. If you want generated OpenAPI documentation, flask-smorest builds on marshmallow and gives you a Swagger UI plus request and response validation through decorators, which is the closest Flask gets to FastAPI's developer experience. A neat trick for teams migrating gradually: write a small decorator that validates the body against a schema and passes the parsed object as a keyword argument, so views stay clean and the validation is visible in the signature.

from functools import wraps
from flask import request, jsonify
from pydantic import BaseModel, EmailStr, Field, ValidationError

class CreateCandidate(BaseModel):
    email: EmailStr
    full_name: str = Field(min_length=2, max_length=80)
    experience_years: int = Field(ge=0, le=50)

def validate(model):
    def outer(fn):
        @wraps(fn)
        def inner(*args, **kwargs):
            try:
                payload = model.model_validate(request.get_json(silent=True) or {})
            except ValidationError as e:
                return jsonify(error='validation_failed', detail=e.errors()), 422
            return fn(*args, payload=payload, **kwargs)
        return inner
    return outer

@app.post('/candidates')
@validate(CreateCandidate)
def create(payload: CreateCandidate):
    return payload.model_dump(), 201

Answer

Set MAX_CONTENT_LENGTH first. Without it, Werkzeug will read an arbitrarily large body, and a single 2GB upload can exhaust a worker's memory. With it, exceeding the limit raises RequestEntityTooLarge and Flask returns 413, which you should catch with an errorhandler to return JSON instead of an HTML page.

Recent Flask 3.x releases add MAX_FORM_MEMORY_SIZE (a cap on non-file form field data, so an attacker cannot send a hundred megabytes of text fields) and MAX_FORM_PARTS (a cap on the number of multipart parts, which blocks a part-flooding denial of service). Note that MAX_CONTENT_LENGTH is enforced by your app, not by nginx, so also set client_max_body_size in nginx or the equivalent on your ALB, otherwise the proxy buffers the whole body before Flask ever sees it. On the file itself: always run secure_filename on the client-supplied name (it strips directory components and non-ASCII), never use that name as the storage key, generate a UUID or content hash instead, and store outside the web root.

Never trust the reported content type or the extension; sniff the real type with python-magic or by reading the header bytes, and for images re-encode through Pillow rather than serving the original bytes, which defuses polyglot files. FileStorage exposes a stream, so for large files you should stream directly to S3 with boto3's upload_fileobj rather than calling .save() to local disk, and for very large uploads the right architecture is a presigned S3 POST or PUT so the bytes never touch your Python process at all. Finally, add antivirus scanning (ClamAV) for anything users will download again.

import uuid, boto3
from werkzeug.utils import secure_filename
from werkzeug.exceptions import RequestEntityTooLarge

app.config['MAX_CONTENT_LENGTH'] = 10 * 1024 * 1024
app.config['MAX_FORM_MEMORY_SIZE'] = 512 * 1024

ALLOWED = {'application/pdf', 'image/png', 'image/jpeg'}
s3 = boto3.client('s3')

@app.errorhandler(RequestEntityTooLarge)
def too_large(e):
    return {'error': 'file_too_large', 'limit_mb': 10}, 413

@app.post('/resume')
def upload_resume():
    f = request.files.get('file')
    if f is None or f.mimetype not in ALLOWED:
        return {'error': 'unsupported_media_type'}, 415
    key = 'resumes/%s' % uuid.uuid4().hex
    s3.upload_fileobj(f.stream, 'gs-uploads', key,
                      ExtraArgs={'ContentType': f.mimetype})
    return {'key': key, 'original': secure_filename(f.filename)}, 201

Answer

CSRF is only possible when the browser attaches credentials automatically, which in practice means cookies (or HTTP Basic auth). Flask-WTF's CSRFProtect generates a random token, signs it with SECRET_KEY through itsdangerous, stores the raw value in the session, and expects the signed token back on every state-changing request, either in a form field named csrf_token or in the X-CSRFToken header. Because an attacker's page cannot read your session cookie or your rendered page, it cannot produce a matching token, so the forged POST fails with 400 CSRFError.

CSRFProtect(app) protects every POST, PUT, PATCH and DELETE app-wide, and you opt individual views out with @csrf.exempt, typically for third-party webhooks that authenticate with their own signature. WTF_CSRF_TIME_LIMIT controls token expiry (3600 seconds by default), and long-lived single-page forms need it raised or refreshed, otherwise users see spurious failures after an hour on the page. For a pure JSON API authenticated with an Authorization: Bearer header, CSRF protection is unnecessary, because the browser does not attach that header automatically, so a cross-site request simply arrives unauthenticated.

The nuance interviewers probe: if you store the JWT in a cookie for convenience, CSRF is back, and you need either SameSite=Strict plus a double-submit token or the CSRF protection that flask-jwt-extended provides through JWT_COOKIE_CSRF_PROTECT. Also remember SameSite is defence in depth, not a substitute, older browsers and some navigation flows do not enforce it consistently.

from flask_wtf import CSRFProtect
from flask_wtf.csrf import CSRFError, generate_csrf

csrf = CSRFProtect()
csrf.init_app(app)
app.config['WTF_CSRF_TIME_LIMIT'] = 8 * 3600

@app.errorhandler(CSRFError)
def on_csrf(e):
    return {'error': 'csrf_failed', 'detail': e.description}, 400

# Hand the token to a SPA on the same origin
@app.get('/csrf-token')
def csrf_token():
    return {'token': generate_csrf()}

# Razorpay-style webhook: verify HMAC instead of CSRF
@app.post('/webhooks/payments')
@csrf.exempt
def payments_webhook():
    verify_signature(request.headers.get('X-Signature'), request.get_data())
    return '', 204

Answer

Flask-Login is cookie-session authentication for server-rendered apps. You give it a user_loader callback, call login_user(user) after verifying credentials, and it stores the user id in the Flask session cookie. @login_required then guards views, current_user is a proxy to the loaded user, and remember-me cookies extend the login beyond the session. Its strengths are simplicity and instant revocation: because the identity lives in the session, deleting or rotating the server-side state (or the secret) logs the user out immediately.

Its limits are that it is browser-shaped, awkward for mobile clients and cross-domain SPAs, and it inherits every cookie-session constraint including the 4KB limit and CSRF exposure. flask-jwt-extended issues signed JSON Web Tokens: create_access_token(identity=user.id) returns a short-lived token, @jwt_required() validates it, and get_jwt_identity() reads the subject. Tokens are stateless, so any service with the public key or shared secret can verify them without a database lookup, which is why they suit mobile apps, microservices, and third-party API access. The cost is revocation.

A valid token stays valid until it expires, so logout, password change, and account suspension need a denylist, typically JWT ids in Redis with a TTL matching the token expiry, which reintroduces the state you were trying to avoid. The standard production shape is a fifteen-minute access token plus a longer refresh token that is stored server-side and rotated on use, so compromise windows stay small. Set token expiry deliberately, and never put anything sensitive in the payload, JWT claims are base64, not encrypted.

from datetime import timedelta
from flask_jwt_extended import (JWTManager, create_access_token,
                                jwt_required, get_jwt, get_jwt_identity)

app.config['JWT_SECRET_KEY'] = os.environ['JWT_SECRET_KEY']
app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(minutes=15)
app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)
jwt = JWTManager(app)

@jwt.token_in_blocklist_loader
def is_revoked(jwt_header, jwt_payload):
    return redis_client.get('revoked:' + jwt_payload['jti']) is not None

@app.post('/logout')
@jwt_required()
def logout():
    claims = get_jwt()
    ttl = claims['exp'] - int(time.time())
    redis_client.setex('revoked:' + claims['jti'], max(ttl, 1), '1')
    return {'ok': True}

Answer

The base fixtures are an app built by your factory with TESTING=True and a throwaway database, and a client from app.test_client(). The test client speaks WSGI directly, so no server runs and no port is bound, making tests fast and deterministic. client.get, client.post(json={...}), client.post(data={...}) for forms, and follow_redirects=True cover most cases, and the returned response exposes status_code, headers, data, and get_json(). Setting TESTING=True makes Flask propagate exceptions instead of converting them to 500s, so a bug surfaces as a real traceback in the test output.

Use client.session_transaction() to read or seed the session without going through a login endpoint, which keeps auth setup fast. app.test_request_context('/path?x=1', method='POST', json={...}) pushes a request context without dispatching, which is what you need to unit test a helper that reads request or calls url_for, and app.test_cli_runner() invokes CLI commands in process. For database isolation, the fastest reliable pattern is to run each test inside a transaction that is rolled back afterwards: bind the session to a connection with an open transaction and roll it back in fixture teardown, so no test sees another's rows and you avoid recreating tables per test. Use a real Postgres in CI rather than SQLite, because SQLite silently accepts things Postgres rejects and misses constraint and type behaviour you will hit in production. Add one test per endpoint that asserts on the error path, not only the happy path, that is where interviewers look.

import pytest
from myapp import create_app
from myapp.extensions import db

@pytest.fixture
def app():
    app = create_app('config.Testing')
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

def test_create_requires_email(client):
    r = client.post('/candidates', json={'full_name': 'Asha'})
    assert r.status_code == 422
    assert r.get_json()['error'] == 'validation_failed'

def test_logged_in_dashboard(client):
    with client.session_transaction() as sess:
        sess['uid'] = 42
    assert client.get('/dashboard').status_code == 200

Answer

When Werkzeug cannot match a URL to any rule, there is no blueprint associated with the request, because blueprint association comes from the matched endpoint. So Flask has nothing to dispatch a blueprint-scoped 404 handler on, and only the application-level handler runs. This is not a bug, it is a direct consequence of routing order, and it surprises teams who mount an API blueprint at /api and expect unmatched /api/anything to return their JSON error shape.

The workaround that actually works is a single app-level 404 handler that inspects request.path (or request.blueprint, which is set for matched routes) and returns JSON for API prefixes and HTML otherwise. Alternatively, negotiate on the Accept header, which is cleaner if your API and site share a domain. Handlers registered with @bp.errorhandler do work for exceptions raised inside that blueprint's views, including 403, 500, and custom exception classes, so blueprint scoping is still useful for everything except unrouted paths.

There is a second scoping subtlety: @bp.app_errorhandler registers an application-wide handler from inside a blueprint module, which is how you keep error handling code near the feature that owns it while still catching global errors. And handler lookup for exception classes walks the MRO and prefers the blueprint handler over the app handler for the same class, so a blueprint handler for Exception shadows the app one within that blueprint. Interviewers use this question to see whether you have actually shipped a multi-blueprint app or only read the tutorial.

from werkzeug.exceptions import NotFound
from flask import request, jsonify, render_template

@app.errorhandler(404)
def not_found(e: NotFound):
    wants_json = (request.path.startswith('/api/')
                  or request.accept_mimetypes.best == 'application/json')
    if wants_json:
        return jsonify(error='not_found', path=request.path), 404
    return render_template('404.html'), 404

# Works: raised inside the blueprint's own views
@api_bp.errorhandler(PermissionError)
def on_permission(e):
    return jsonify(error='forbidden'), 403

# App-wide handler declared inside the blueprint module
@api_bp.app_errorhandler(500)
def on_500(e):
    app.logger.exception('unhandled')
    return jsonify(error='internal_error'), 500

Answer

Installing flask[async] pulls in asgiref and lets you write async def view functions, error handlers, and before_request or after_request hooks. What Flask does with them is narrow: when the dispatcher sees a coroutine function it hands it to asgiref.sync.async_to_sync, which creates an event loop, runs the coroutine to completion, and returns the result, all inside the same worker thread that was already handling the request. Flask remains a WSGI application throughout.

That single fact decides the whole answer. What you gain is the ability to await async libraries inside a view, and in particular to fan out concurrent I/O with asyncio.gather, so three upstream calls of 200ms each complete in 200ms instead of 600ms. That is a genuine latency win for aggregation endpoints.

What you do not gain is throughput: a sync gunicorn worker still processes exactly one request at a time, so async def will not let one process hold a thousand concurrent connections. For that you need gthread, gevent, or a real ASGI stack. Three things bite in production.

A fresh event loop is created per request, so a module-level httpx.AsyncClient or asyncpg pool built at import time is attached to a dead loop and raises 'Event loop is closed' or 'attached to a different loop' on the second request. The loop setup costs roughly a hundred microseconds, irrelevant on a 200ms endpoint and visible on a 2ms one. And most Flask extensions are synchronous, so calling db.session inside an async view still blocks the thread and buys you nothing. The honest answer in an interview is that async views are a compatibility feature for calling async libraries, not a concurrency model.

import asyncio
import httpx
from flask import current_app

# pip install 'flask[async]'  ->  brings in asgiref

BASE = 'https://internal.example.com'

@app.get('/candidates/<int:cid>/card')
async def card(cid):
    # Build the client INSIDE the view: async_to_sync creates a new event
    # loop per request, so a module-level AsyncClient dies on request two.
    async with httpx.AsyncClient(timeout=httpx.Timeout(3.0, connect=1.0)) as c:
        profile, scores = await asyncio.gather(
            c.get(f'{BASE}/profile/{cid}'),
            c.get(f'{BASE}/scores/{cid}'),
            return_exceptions=True,
        )

    if isinstance(profile, Exception):
        current_app.logger.warning('profile fetch failed: %s', profile)
        return {'error': 'upstream_unavailable'}, 503

    return {
        'profile': profile.json(),
        'scores': None if isinstance(scores, Exception) else scores.json(),
    }

Answer

Returning a generator from a view makes Flask build a streaming response: it iterates the generator lazily and, because no Content-Length is known, the server switches to chunked transfer encoding. The trap is lifecycle. Flask pops the request context as soon as the view function returns, which is before the generator body runs, so touching request, session, or g inside the generator raises 'Working outside of request context'.

Wrapping the generator in stream_with_context keeps the context alive for the duration of the iteration and is the fix. The same applies to the application context, so a generator that streams rows out of db.session needs the context to survive, which stream_with_context also handles because the request context implies an app context. Deployment matters more than the code here.

A gunicorn sync worker is fully occupied for as long as the stream is open, so a hundred concurrent SSE clients need a hundred workers, which is why long-lived streams belong on gthread, gevent, or a dedicated service. gunicorn's --timeout is a worker liveness check, not a request timeout, and a sync worker sitting in a long stream can be SIGKILLed mid-response, so raise it or pick a worker class that heartbeats. In front of the app, nginx buffers responses by default and will hold your SSE events until the buffer fills, so send X-Accel-Buffering: no or set proxy_buffering off, and disable gzip for the stream because compression buffers too. On the data side, use SQLAlchemy execution_options(yield_per=1000) so the result set is not fully materialised in memory, otherwise you have replaced a slow response with a 2GB worker.

import csv, io, json, time
from flask import Response, stream_with_context

@app.get('/export.csv')
def export_csv():
    def rows():
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(['id', 'email', 'city'])
        yield buf.getvalue(); buf.seek(0); buf.truncate(0)
        stmt = db.select(Candidate).execution_options(yield_per=1000)
        for c in db.session.scalars(stmt):
            w.writerow([c.id, c.email, c.city])
            yield buf.getvalue(); buf.seek(0); buf.truncate(0)

    return Response(
        stream_with_context(rows()),
        mimetype='text/csv',
        headers={'Content-Disposition': 'attachment; filename=export.csv'},
    )

@app.get('/events')
def events():
    def sse():
        while True:
            yield 'event: tick\ndata: %s\n\n' % json.dumps({'ts': time.time()})
            time.sleep(15)   # keep-alive, proxies drop idle connections

    return Response(
        stream_with_context(sse()),
        mimetype='text/event-stream',
        headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
    )

Answer

Flask-Caching is the usual choice: configure CACHE_TYPE ('SimpleCache' for local development, 'RedisCache' in production), CACHE_REDIS_URL, and CACHE_DEFAULT_TIMEOUT, then bind it with cache.init_app(app). Two decorators do most of the work and both have sharp edges. @cache.cached(timeout=60) keys on the request path only, so ?page=2 and ?page=3 return the same cached body until you pass query_string=True or a custom key_prefix callable. That is the single most common caching bug in Flask code review. @cache.memoize hashes the function arguments into the key, which means memoising a bound method includes self, and if the object has no stable repr you get a new key every call and a cache that never hits while quietly filling Redis.

The bigger risk is authorisation: caching a view that renders per-user or per-tenant data with a path-only key serves one customer's data to another, and in a multi-tenant Indian SaaS product that is a reportable data incident, not a bug. Always include the tenant or user id in the key, or do not cache the view at all and cache the expensive query underneath it instead. Use response_filter so error responses are not cached, otherwise a transient 500 gets pinned for the full TTL.

Add jitter to TTLs, because a thousand keys expiring in the same second produce a stampede that hits the database harder than having no cache. Alongside the object cache, HTTP caching is nearly free: set an ETag from a cheap version marker such as updated_at and call make_conditional(request) so repeat clients get a 304 without you serialising anything.

import os, random
from flask import g, request, jsonify
from flask_caching import Cache

cache = Cache(config={
    'CACHE_TYPE': 'RedisCache',
    'CACHE_REDIS_URL': os.environ['REDIS_URL'],
    'CACHE_DEFAULT_TIMEOUT': 60,
})
cache.init_app(app)

# WRONG: default key is the path, so ?page=2 serves page 1's body
# @app.get('/jobs')
# @cache.cached(timeout=60)
# def jobs(): ...

def jobs_key():
    return 'jobs:%s:%s' % (g.tenant_id, request.full_path)

@app.get('/jobs')
@cache.cached(timeout=60 + random.randint(0, 15),
              key_prefix=jobs_key,
              response_filter=lambda r: r.status_code == 200)
def jobs():
    return jsonify(items=list_jobs(g.tenant_id))

@app.get('/jobs/<int:job_id>')
def job(job_id):
    j = db.get_or_404(Job, job_id)
    resp = jsonify(id=j.id, title=j.title)
    resp.set_etag('job-%d-%d' % (j.id, int(j.updated_at.timestamp())))
    resp.cache_control.max_age = 30
    return resp.make_conditional(request)

Answer

Flask-Limiter is the standard extension. You give it a key_func (get_remote_address by default), a storage_uri, default_limits, and optionally a strategy: fixed-window is cheapest, moving-window is fairer and costs more Redis operations. Two configuration mistakes account for almost every production complaint.

The first is storage. The default in-memory backend is per process, so five gunicorn workers each enforce their own counter and a '100 per minute' limit becomes 500 per minute, distributed unpredictably. Flask-Limiter warns about this at startup, and the fix is a shared Redis or Memcached storage_uri.

The second is the key. Behind an ALB, nginx, or Cloudflare, request.remote_addr is the proxy's address, so every client in the world shares one bucket: either nobody is limited because the bucket resets constantly, or a single abusive client throttles all your users. The correct fix is werkzeug.middleware.proxy_fix.ProxyFix with x_for set to the exact number of trusted proxies in the chain, so Werkzeug reads the correct entry from X-Forwarded-For counting from the right.

Setting x_for too high lets a client forge the header and impersonate any IP, which defeats both rate limiting and audit logging. Even with the IP correct, IP is a poor key in India: mobile carriers use carrier-grade NAT and a large office sits behind a single egress IP, so thousands of legitimate users share an address. For authenticated traffic, key on the API key, user id, or tenant instead, and reserve IP limits for unauthenticated endpoints such as OTP and login. Exempt health checks with @limiter.exempt, enable headers_enabled so clients see X-RateLimit-Remaining, and return a JSON 429 with Retry-After rather than Werkzeug's HTML page.

import os
from flask import request
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from werkzeug.middleware.proxy_fix import ProxyFix

# One ALB plus one nginx in front, so trust exactly two hops
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=2, x_proto=1, x_host=1)

def limit_key():
    api_key = request.headers.get('X-Api-Key')
    return 'key:' + api_key if api_key else 'ip:' + get_remote_address()

limiter = Limiter(
    key_func=limit_key,
    storage_uri=os.environ['REDIS_URL'],   # shared across gunicorn workers
    strategy='moving-window',
    default_limits=['1000 per hour'],
    headers_enabled=True,
)
limiter.init_app(app)

@app.get('/healthz')
@limiter.exempt
def healthz():
    return {'status': 'ok'}

@app.post('/auth/otp')
@limiter.limit('5 per minute; 30 per day',
               key_func=lambda: (request.get_json(silent=True) or {}).get('phone', ''))
def send_otp():
    return {'sent': True}

@app.errorhandler(429)
def on_429(e):
    return {'error': 'rate_limited', 'detail': e.description}, 429

Answer

app.logger is an ordinary logging.Logger named after the app's import name, and Flask attaches a default StreamHandler to it only if the root logger has no handlers configured when the first record is emitted. That lazy, conditional behaviour is exactly why logs vanish. If a library calls logging.basicConfig at import, or you call dictConfig after the app has already logged, or gunicorn's own logging config takes over, your handler is either never added or shadowed, and lines end up formatted differently, duplicated, or dropped.

The reliable pattern from the Flask documentation is to call logging.config.dictConfig before you construct the Flask object, configure the root logger there, and let app.logger propagate to it. Do not add handlers directly to app.logger in a factory that may be called twice in tests, you will get duplicate lines per call. Under gunicorn, pass --access-logfile - and --error-logfile - so both streams go to stdout and stderr where a container runtime can collect them, and remember gunicorn's access log is separate from your application log, so a request that never reaches a view still appears in the access log only.

For anything you intend to query later, emit structured JSON with python-json-logger or structlog and attach a correlation id: a logging.Filter that reads g.request_id when has_request_context() is true keeps every line joinable to a trace and to the load balancer's request id. Guard against the obvious compliance problem, never log request bodies wholesale, since that is how OTPs, tokens, Aadhaar numbers, and card details end up in a log index that a much larger group of people can read. Log at INFO in production, sample high-volume debug lines, and always use app.logger.exception inside an error handler so the traceback is attached.

import logging, logging.config
from flask import g, request, has_request_context

class RequestContextFilter(logging.Filter):
    def filter(self, record):
        ctx = has_request_context()
        record.request_id = g.get('request_id', '-') if ctx else '-'
        record.path = request.path if ctx else '-'
        return True

# Runs BEFORE Flask() is constructed
logging.config.dictConfig({
    'version': 1,
    'disable_existing_loggers': False,
    'filters': {'ctx': {'()': RequestContextFilter}},
    'formatters': {
        'json': {
            '()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
            'format': '%(asctime)s %(levelname)s %(name)s %(request_id)s %(path)s %(message)s',
        },
    },
    'handlers': {
        'stdout': {
            'class': 'logging.StreamHandler',
            'stream': 'ext://sys.stdout',
            'formatter': 'json',
            'filters': ['ctx'],
        },
    },
    'root': {'level': 'INFO', 'handlers': ['stdout']},
})

app = create_app()
app.logger.info('boot complete')   # propagates to the root handler

Answer

Use flask-cors and scope it per resource rather than switching it on globally: CORS(app, resources={r'/api/*': {'origins': [...]}}) keeps your server-rendered pages out of the policy. supports_credentials=True is what allows cookies and Authorization headers on cross-origin calls, and it cannot be combined with origins='*', because the spec requires an exact origin in Access-Control-Allow-Origin when credentials are involved; flask-cors will echo the request origin only if it matches your list. Preflight is where most debugging time goes. For any request that is not a simple GET or POST with a basic content type, the browser first sends OPTIONS carrying Access-Control-Request-Method and Access-Control-Request-Headers, and critically it sends no cookies and no Authorization header.

So a global @app.before_request auth guard that returns 401 for unauthenticated requests will reject the preflight, and the browser reports only a generic CORS failure with no status code visible to your frontend engineer. The fix is a single early return for request.method == 'OPTIONS'. A second frequent failure is duplicated headers: if nginx also adds Access-Control-Allow-Origin, the browser sees two values and refuses the response, so pick one layer and remove the other.

Third, custom response headers are invisible to fetch unless you list them in expose_headers, which is why a frontend that cannot read X-Total-Count or X-Request-Id usually has a working API and a missing config line. Set max_age so the browser caches the preflight and you halve request volume on chatty SPAs. Finally, remember CORS is enforced by browsers only, it stops a malicious page from reading your responses, it does not authenticate anything, and curl or a server-side attacker ignores it entirely.

from flask import request
from flask_cors import CORS

CORS(
    app,
    resources={r'/api/*': {'origins': ['https://app.example.com',
                                       'https://admin.example.com']}},
    supports_credentials=True,
    expose_headers=['X-Request-Id', 'X-Total-Count', 'Retry-After'],
    max_age=86400,          # cache the preflight for a day
)

@app.before_request
def require_token():
    # Preflight arrives without credentials: let it through or every
    # cross-origin call fails with an unexplained CORS error.
    if request.method == 'OPTIONS':
        return None
    if not request.path.startswith('/api/'):
        return None
    if not request.headers.get('Authorization'):
        return {'error': 'unauthorized'}, 401

Answer

gunicorn runs a pre-fork master that spawns N worker processes, and the worker class decides how each one handles concurrency. The sync worker serves one request at a time with no patching and no shared state, which makes it the most predictable and the most memory hungry, because concurrency equals worker count. gthread gives each worker a thread pool of --threads size, so concurrency is workers times threads while memory stays amortised, and that is the right default for an I/O bound Flask API since the GIL is released during socket and database waits. gevent and eventlet swap blocking I/O for greenlets and can hold thousands of connections per worker, at the cost of monkey patching and library cooperation. Sizing: (2 x cores) + 1 is a starting heuristic, not a rule.

The number that matters comes from Little's law, concurrency equals throughput times latency, so 300 requests per second at 80ms needs roughly 24 slots, which is three workers with eight threads plus headroom for p99. Then check memory: workers times RSS must fit inside the container limit or the kernel OOM-kills the pod mid-request. The most misunderstood flag is --timeout.

It is not a request deadline. It is how long the master waits for a worker heartbeat before sending SIGKILL and forking a replacement, and when it fires every in-flight request on that worker dies with no response and no traceback. 'Random 502s under load' is very often a 30 second --timeout against an endpoint that occasionally takes 35. Raising it hides the symptom; the fix is a real timeout on the slow dependency and moving long work into Celery. --graceful-timeout governs shutdown, --keepalive matters only when clients connect directly rather than through a connection-pooling load balancer, and --max-requests with jitter recycles workers to bound slow leaks.

# gunicorn.conf.py
import os

bind = '0.0.0.0:8000'

# I/O bound Flask API on a 2 vCPU container
workers = int(os.getenv('WEB_CONCURRENCY', '3'))
worker_class = 'gthread'
threads = 4                  # 3 x 4 = 12 concurrent requests

timeout = 30                 # worker LIVENESS, not a request deadline
graceful_timeout = 30
keepalive = 5

max_requests = 1000          # recycle workers to bound slow leaks
max_requests_jitter = 100

accesslog = '-'
errorlog = '-'
preload_app = False

def post_fork(server, worker):
    from myapp.wsgi import app
    from myapp.extensions import db
    with app.app_context():
        db.engine.dispose()  # children must not inherit the parent's sockets

# gunicorn -c gunicorn.conf.py 'myapp:create_app()'

Answer

First separate a leak from fragmentation. CPython returns freed objects to its own allocator, and glibc keeps per-thread malloc arenas, so RSS is a high-water mark that rarely falls even when Python memory is genuinely free. Setting MALLOC_ARENA_MAX=2 often flattens a mysterious sawtooth on threaded workers without changing a line of code.

If RSS still grows monotonically, look for the five usual causes: a manually pushed application context that never pops, so the SQLAlchemy identity map accumulates every object the process has ever loaded; a module-level dict or list used as a cache with no eviction; buffering whole uploads or upstream responses into memory instead of streaming; a logging handler with a queue that grows faster than it drains; and objects that keep references to request or response objects, for example a metrics dictionary keyed by full URL, which is unbounded when the URL contains ids. Diagnose with tracemalloc: start it at boot with a depth of 25, take a snapshot, take another an hour later, and compare_to on 'lineno' gives you the exact source lines that grew. objgraph.show_growth() is a quicker first pass for counting object types, and py-spy dump attaches to a running production process without restarting it, which matters when the leak takes six hours to reproduce. memray gives allocation-level flame graphs when you can run the workload offline. Fixes follow the cause: bound caches with cachetools.LRUCache instead of a bare dict, call db.session.remove() between batch chunks, use a fresh app context per chunk, stream uploads straight to S3, and normalise metric labels so cardinality is bounded. Add --max-requests 1000 --max-requests-jitter 100 as a safety net, not as the fix, and set a container memory limit so a regression restarts one pod instead of degrading the whole node.

import gc, tracemalloc
from flask import Blueprint

ops = Blueprint('ops', __name__, url_prefix='/internal')
_baseline = None

@ops.post('/mem/start')
def mem_start():
    tracemalloc.start(25)
    return {'tracing': True}

@ops.post('/mem/diff')
def mem_diff():
    global _baseline
    gc.collect()
    snap = tracemalloc.take_snapshot()
    if _baseline is None:
        _baseline = snap
        return {'baseline': True}
    top = snap.compare_to(_baseline, 'lineno')[:15]
    _baseline = snap
    return {'top': ['%s +%.1f KB' % (s.traceback.format()[-1], s.size_diff / 1024)
                    for s in top]}

# Bound the cache instead of using a bare module-level dict
from cachetools import LRUCache, TTLCache
SCORE_CACHE = TTLCache(maxsize=10_000, ttl=300)

# Environment mitigations
#   MALLOC_ARENA_MAX=2
#   gunicorn --max-requests 1000 --max-requests-jitter 100

Answer

gevent achieves concurrency by monkey patching socket, ssl, select, time, and threading so that blocking calls yield to a hub instead of blocking the OS thread. The patch only affects modules that have not already been imported and bound to the original functions, so monkey.patch_all() must be the first thing that runs in the first module the interpreter loads, before Flask, SQLAlchemy, requests, or boto3 are imported. Doing it inside create_app is too late and you get MonkeyPatchWarning about ssl already being imported, followed by hangs that only appear under load.

With gunicorn's gevent worker the patching happens in the worker before it loads the app, which is correct, but combining -k gevent with --preload breaks that: the master imports the application unpatched and forks it, so half your library stack is using real blocking sockets. Beyond patching, three classes of code do not cooperate. Native extensions that block inside C never yield, so mysqlclient, some crypto operations, and heavy NumPy or Pillow loops freeze every greenlet in the worker, not just their own. psycopg2 needs psycogreen's patch_psycopg() to become cooperative; psycopg3 handles this natively.

And any CPU-bound stretch is a full stop for the worker, so a 500ms JSON serialisation stalls a thousand connections at once. The other trap is resource budgeting. With --worker-connections 1000 and three workers, three thousand greenlets can each want a database connection while your pool is sized for a handful, so you either exhaust Postgres or every greenlet blocks in pool_timeout.

Size the pool for greenlet concurrency and put PgBouncer in front. Turn on gevent's monitoring thread (GEVENT_MONITOR_THREAD_ENABLE=true with GEVENT_MAX_BLOCKING_TIME) so blocked-hub warnings name the offending stack rather than leaving you with unexplained latency spikes.

# wsgi.py  -  these must be the first executable lines in the process
from gevent import monkey
monkey.patch_all()

from psycogreen.gevent import patch_psycopg
patch_psycopg()

from myapp import create_app
app = create_app()

# 3 workers x 1000 connections could demand 3000 DB connections.
# Size the pool for greenlets and front it with PgBouncer.
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
    'pool_size': 10,
    'max_overflow': 10,
    'pool_timeout': 5,
    'pool_pre_ping': True,
}

# gunicorn -k gevent --worker-connections 1000 --workers 3 wsgi:app
# (never add --preload with the gevent worker)
#
# Find greenlets that block the hub:
#   GEVENT_MONITOR_THREAD_ENABLE=true GEVENT_MAX_BLOCKING_TIME=0.1

Answer

Behind a proxy, the WSGI environ describes the connection from the proxy, not from the user: REMOTE_ADDR is the proxy's IP, wsgi.url_scheme is http even though the client used https, and HTTP_HOST may be the internal service name. The visible symptoms are rate limits that bucket every user together, audit logs full of one IP, url_for(_external=True) emitting http:// links in password reset emails, and infinite redirect loops when a security extension forces https. The fix is werkzeug.middleware.proxy_fix.ProxyFix, which rewrites the environ from X-Forwarded-For, X-Forwarded-Proto, X-Forwarded-Host, X-Forwarded-Port and X-Forwarded-Prefix.

The critical parameter is the hop count: x_for=2 means take the second value from the right of X-Forwarded-For, matching exactly two trusted proxies in the chain. Set it too high and a client can prepend a forged address that Werkzeug will treat as the real client, defeating rate limiting, IP allowlists, and fraud checks; set it too low and you record your own proxy. Count the hops in your actual topology (CloudFront plus ALB plus nginx is three) and re-check it whenever infrastructure changes.

Assign it as app.wsgi_app = ProxyFix(app.wsgi_app, ...) rather than wrapping the Flask object, so Flask's own test client and error handling still work. The same middleware layer is useful for two other jobs. Plain WSGI middleware runs before routing and sees every request including ones that 404, which makes it the right place for a hard body-size guard or a request id that must exist even for unmatched paths. And DispatcherMiddleware mounts multiple WSGI apps under different prefixes in one process, which is how teams expose a Prometheus /metrics endpoint or a separate admin app without a second container.

from werkzeug.middleware.proxy_fix import ProxyFix
from werkzeug.middleware.dispatcher import DispatcherMiddleware
from prometheus_client import make_wsgi_app

# Topology: ALB -> nginx -> gunicorn  =>  two trusted hops
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=2, x_proto=1, x_host=1, x_prefix=1)

class BodySizeGuard:
    """Runs before routing, so it also guards unmatched paths."""

    def __init__(self, wsgi_app, limit):
        self.wsgi_app, self.limit = wsgi_app, limit

    def __call__(self, environ, start_response):
        try:
            size = int(environ.get('CONTENT_LENGTH') or 0)
        except ValueError:
            size = 0
        if size > self.limit:
            start_response('413 Payload Too Large',
                           [('Content-Type', 'application/json')])
            return [b'{"error": "too_large"}']
        return self.wsgi_app(environ, start_response)

app.wsgi_app = BodySizeGuard(app.wsgi_app, 10 * 1024 * 1024)

# One process, three mounted WSGI apps
application = DispatcherMiddleware(app, {'/metrics': make_wsgi_app()})

Answer

The contract exists because one extension instance may serve several application objects: a test suite creates a new app per test, and some deployments run a public app and an internal app in one process. So the rule is that the extension object holds no app-specific state. __init__(self, app=None) stores only configuration-independent defaults and calls init_app(app) if an app was passed as a convenience. init_app(app) does the real work: call app.config.setdefault for every key under a single namespace prefix so users can see and override them, register hooks with app.before_request or app.teardown_appcontext, add CLI commands via app.cli, and store per-app state in app.extensions['your_name'], which is the dictionary Flask maintains exactly for this purpose. At runtime, look state up through current_app.extensions rather than self, so the correct app's configuration is used no matter which context is active.

Lazily create expensive resources (a client, a connection pool) on first access and cache them in that per-app state, and register a teardown_appcontext callback that releases whatever you attached to g. Two more things separate a good extension from a fragile one. Emit blinker signals for interesting events, so consumers can observe without subclassing you, which is the pattern Flask itself uses for request_started and template_rendered.

And never raise at import time for missing configuration; raise inside init_app so the failure is attributable to the app that is misconfigured, and make the message name the exact config key. Interviewers ask this question less to hear you build an extension and more to check that you understand why every mature Flask extension has an init_app method, which is the same reason the application factory pattern exists.

import click
from blinker import Namespace
from flask import current_app, g

_signals = Namespace()
report_generated = _signals.signal('report-generated')

class ReportEngine:
    def __init__(self, app=None):
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        app.config.setdefault('REPORTS_BUCKET', 'gs-reports')
        app.config.setdefault('REPORTS_TIMEOUT', 30)
        if not app.config.get('REPORTS_ENDPOINT'):
            raise RuntimeError('REPORTS_ENDPOINT is required by ReportEngine')

        app.extensions['report_engine'] = {'client': None}
        app.teardown_appcontext(self._teardown)
        app.cli.add_command(self._cli)

    @property
    def client(self):
        state = current_app.extensions['report_engine']
        if state['client'] is None:
            state['client'] = build_client(current_app.config)
        return state['client']

    def render(self, template, **ctx):
        out = self.client.render(template, ctx)
        report_generated.send(current_app._get_current_object(), name=template)
        return out

    def _teardown(self, exc=None):
        handle = g.pop('report_handle', None)
        if handle is not None:
            handle.close()

    @staticmethod
    @click.command('warm-reports')
    def _cli():
        click.echo('warmed')

reports = ReportEngine()   # module scope; bound inside create_app

Answer

Flask 2.2 replaced the old app.json_encoder and JSONEncoder hooks with the JSONProvider interface. app.json is an instance of DefaultJSONProvider, and every JSON path in the framework goes through it: jsonify, bare dict and list returns from views, flask.json.dumps and loads while an app context is active, and the tojson Jinja filter. You customise by subclassing JSONProvider or DefaultJSONProvider and assigning app.json = MyProvider(app) inside the factory, which changes serialisation everywhere at once rather than in the handful of places you remembered. Two details of the default provider surprise people.

It serialises datetime and date with werkzeug.http.http_date, producing an HTTP date string rather than ISO 8601, so that is the format your existing clients have been parsing for years. And it sorts keys by default, which costs measurable time on wide objects and can be turned off with app.json.sort_keys = False. Swapping in orjson is the standard performance move: it is implemented in Rust, returns bytes, and is roughly an order of magnitude faster than the standard library on large payloads.

The catch is compatibility. orjson emits ISO 8601 for datetimes natively, so unless you keep the old behaviour in a default= hook you have shipped a silent breaking change to every mobile client and integration partner. orjson also has no sort_keys keyword (use OPT_SORT_KEYS) and rejects non-string dict keys unless you pass OPT_NON_STR_KEYS. Whether it is worth it depends entirely on your profile. On a typical CRUD API the database dominates and serialisation is a few percent of the request, so the change buys nothing. On a list endpoint returning several megabytes of nested rows at high request rates it is often the single largest item in the flame graph, and there it pays for itself the day you deploy it.

import dataclasses, datetime, uuid
import orjson
from flask import Flask
from flask.json.provider import JSONProvider
from werkzeug.http import http_date

def _default(obj):
    if isinstance(obj, (datetime.datetime, datetime.date)):
        return http_date(obj)        # keep Flask's legacy wire format
    if isinstance(obj, uuid.UUID):
        return str(obj)
    if dataclasses.is_dataclass(obj) and not isinstance(obj, type):
        return dataclasses.asdict(obj)
    if hasattr(obj, '__html__'):
        return str(obj.__html__())
    raise TypeError(type(obj).__name__)

class OrjsonProvider(JSONProvider):
    option = orjson.OPT_NON_STR_KEYS

    def dumps(self, obj, **kwargs):
        return orjson.dumps(obj, default=_default, option=self.option).decode()

    def loads(self, s, **kwargs):
        return orjson.loads(s)

def create_app():
    app = Flask(__name__)
    app.json = OrjsonProvider(app)   # jsonify, dict returns and tojson all use it
    return app

Answer

OpenTelemetry is the default answer in 2026. Install opentelemetry-instrumentation-flask plus the instrumentations for whatever the app talks to: sqlalchemy, requests or httpx, redis, celery, boto3. FlaskInstrumentor().instrument_app(app) wraps the WSGI layer, creates a server span per request tagged with the route template rather than the raw path (important, otherwise every id becomes a distinct metric series), and reads an inbound W3C traceparent header so a trace started at the edge continues through your service.

The Celery instrumentation is the piece that matters for the crossing: it injects the trace context into the task's message headers when you call delay or apply_async, and extracts it in the worker, so the task span becomes a child of the request span and you can see a webhook, its database calls, its enqueue, and the worker's downstream calls in one waterfall. Without it, your traces stop at the queue boundary and every asynchronous failure becomes a manual log hunt. Correlation IDs sit alongside traces, not instead of them: accept X-Request-Id from the load balancer or generate one in before_request, store it in g, add it to every log line with a logging filter, echo it in the response header, and set it as a span attribute so a support ticket quoting one id lands you on the exact trace.

Two deployment details decide whether this works. Set instrumentation up in gunicorn's post_fork hook, because BatchSpanProcessor runs an exporter thread and threads do not survive fork, so configuring it in the master with --preload gives you spans that are never flushed. And exclude health and readiness probes with excluded_urls, or a Kubernetes cluster probing every five seconds will dominate your trace volume and your bill. Export over OTLP to Jaeger, Tempo, or SigNoz, which many Indian teams self-host to keep observability spend predictable.

# tracing.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.instrumentation.celery import CeleryInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

def setup_tracing(app, db):
    provider = TracerProvider()
    provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
    trace.set_tracer_provider(provider)

    FlaskInstrumentor().instrument_app(app, excluded_urls='healthz,readyz')
    SQLAlchemyInstrumentor().instrument(engine=db.engine)
    RequestsInstrumentor().instrument()
    CeleryInstrumentor().instrument()

# gunicorn.conf.py
def post_fork(server, worker):
    from myapp.wsgi import app
    from myapp.extensions import db
    with app.app_context():
        setup_tracing(app, db)      # exporter thread must be created per worker

@app.before_request
def correlate():
    ctx = trace.get_current_span().get_span_context()
    g.request_id = (request.headers.get('X-Request-Id')
                    or format(ctx.trace_id, '032x'))

Answer

On SIGTERM, the gunicorn master stops accepting new connections and forwards SIGTERM to the workers. Each worker finishes the request it is serving and exits; if it is still busy after --graceful-timeout (30 seconds by default) it is SIGKILLed and those requests die without a response. So a clean rollout requires that graceful_timeout comfortably exceeds your p99 latency and that the orchestrator waits at least that long before force-killing the container.

In Kubernetes, terminationGracePeriodSeconds must be greater than graceful_timeout, or the kubelet kills the pod while gunicorn is still draining. The subtler problem is that pod termination and endpoint removal are concurrent: the kubelet sends SIGTERM at the same moment the endpoints controller starts propagating the removal to every kube-proxy and ingress. For a few seconds, traffic still arrives at a pod that has stopped accepting, which shows up as a burst of 502s on every deploy.

The standard fix is a preStop hook that sleeps five to ten seconds, so the pod keeps serving while the removal propagates and only then begins draining. Probes need to be different endpoints doing different things: readiness may check the database and fail temporarily to take a pod out of rotation, while liveness must be cheap and dependency-free, because a liveness probe that queries Postgres turns a brief database blip into a rolling restart of your entire fleet, converting a degradation into an outage. Migrations run as a separate release step or an init container, never inside the app process where several workers would race, and they must be backward compatible with the currently deployed code because both versions serve traffic during the rollout. That means expand then contract: add a nullable column, deploy code that writes both, backfill, then drop the old column in a later release.

# gunicorn.conf.py
graceful_timeout = 30          # must be < terminationGracePeriodSeconds
timeout = 30

# deployment.yaml (excerpt)
# spec:
#   terminationGracePeriodSeconds: 60
#   containers:
#     - lifecycle:
#         preStop:
#           exec: { command: ['sh', '-c', 'sleep 10'] }
#       readinessProbe:
#         httpGet: { path: /readyz, port: 8000 }
#         periodSeconds: 5
#       livenessProbe:
#         httpGet: { path: /healthz, port: 8000 }
#         periodSeconds: 10
#         failureThreshold: 6

@app.get('/healthz')          # liveness: never touches a dependency
def healthz():
    return {'status': 'ok'}

@app.get('/readyz')           # readiness: may fail without causing a restart
def readyz():
    try:
        db.session.execute(db.text('SELECT 1'))
    except Exception:
        return {'status': 'degraded'}, 503
    return {'status': 'ready'}

Answer

Start by refusing to guess. Get p50, p95 and p99 per endpoint and a trace waterfall showing where the time goes, because the fix is completely different if it is database time, upstream time, serialisation, or Python overhead. In most Flask services that fall over at ten times the load, the ordering is: database first, then worker saturation, then everything else.

On the database, look for N+1 loads, missing indexes on the exact filter and sort columns, OFFSET pagination that gets slower with every page (switch to keyset pagination on an indexed column), and the connection budget, since workers times threads times pool size can exceed what your Postgres or MySQL instance will accept. PgBouncer in transaction mode plus read replicas for list endpoints buys a lot of headroom cheaply. On the app tier, move to gthread with a thread count derived from measured concurrency, then scale horizontally behind the load balancer with autoscaling driven by p95 latency or requests per pod rather than CPU, which lags badly for I/O bound work.

Take everything that does not need to be in the request off the request path: email, PDFs, webhook fan-out, ML inference, and third-party calls all belong in Celery or RQ, with idempotency keys so retries are safe. Add cache-aside in Redis for hot reads with jittered TTLs and a stampede lock, and put a CDN in front of anything cacheable including public GET endpoints. Every outbound call needs an explicit timeout, a retry budget, and a circuit breaker; one slow partner API with no timeout is the classic way a Flask fleet locks up entirely.

Then protect the system with rate limits and load shedding, because a queue that grows without bound is worse than a fast 429. Finally, know Flask's own ceiling: per-request framework overhead is real but small, and if profiling genuinely shows the framework as the bottleneck, move only the hottest endpoints to an ASGI service rather than rewriting a working system.

# 1) Keyset pagination: constant time regardless of depth
stmt = (db.select(Job)
          .where(Job.id < cursor)
          .order_by(Job.id.desc())
          .options(joinedload(Job.company))
          .limit(50))

# 2) Cache-aside with a stampede lock and jittered TTL
def hot_jobs(tenant_id):
    key = 'hot:%s' % tenant_id
    cached = r.get(key)
    if cached:
        return orjson.loads(cached)
    if not r.set(key + ':lock', '1', nx=True, ex=10):
        return orjson.loads(r.get(key) or b'[]')   # serve stale, do not stampede
    data = expensive_query(tenant_id)
    r.setex(key, 300 + random.randint(0, 60), orjson.dumps(data))
    r.delete(key + ':lock')
    return data

# 3) Every upstream call bounded, no exceptions
session = requests.Session()
adapter = HTTPAdapter(max_retries=Retry(total=2, backoff_factor=0.2,
                                        status_forcelist=[502, 503, 504]))
session.mount('https://', adapter)
resp = session.post(PARTNER_URL, json=payload, timeout=(1.0, 3.0))