### **Top 100 Flask Developer Interview Questions & Answers (Part 3: Questions 201–300)**
---
#### **201. How do you implement distributed tracing in Flask?**
**Answer:** Use **OpenTelemetry** with Jaeger/Datadog:
```python
from opentelemetry.instrumentation.flask import FlaskInstrumentor
FlaskInstrumentor().instrument_app(app)
```
*Configure via environment variables:*
`OTEL_SERVICE_NAME=flask-app OTEL_EXPORTER_JAEGER_ENDPOINT=http://jaeger:14268/api/traces`
---
#### **202. What is the "Zope Transaction" pattern? When is it useful in Flask?**
**Answer:** Manages **multi-database transactions** (e.g., SQL + NoSQL). Use `transaction` library:
```python
import transaction
with transaction.manager:
db.session.add(user)
redis.set("user", user.id)
# Both commit or rollback together
```
---
#### **203. How do you handle database sharding in Flask-SQLAlchemy?**
**Answer:** Configure multiple binds and route queries:
```python
app.config["SQLALCHEMY_BINDS"] = {
"users": "postgresql://user1@shard1/db",
"orders": "postgresql://user2@shard2/db"
}
# In model:
class Order(db.Model):
__bind_key__ = "orders"
```
---
#### **204. How do you implement a circuit breaker for database connections?**
**Answer:** Use `tenacity` with SQLAlchemy:
```python
from tenacity import retry, stop_after_attempt, retry_if_exception_type
from sqlalchemy.exc import OperationalError
@retry(stop=stop_after_attempt(3), retry=retry_if_exception_type(OperationalError))
def get_db():
return db.session.execute("SELECT 1")
```
---
#### **205. How do you prevent SQL injection when using raw queries?**
**Answer:** **Never** concatenate inputs. Use parameterized queries:
```python
# SAFE:
result = db.session.execute("SELECT * FROM users WHERE name = :name", {"name": user_input})
# UNSAFE (AVOID):
result = db.session.execute(f"SELECT * FROM users WHERE name = '{user_input}'")
```
---
#### **206. How do you debug a memory leak in a Gunicorn worker?**
**Answer:** Steps:
1. Identify leaking worker: `gunicorn --max-requests 1 --preload`
2. Capture heap: `tracemalloc` + `objgraph`
3. Analyze:
```python
import tracemalloc
tracemalloc.start()
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
```
---
#### **207. How do you implement exponential backoff for external API calls?**
**Answer:** With `tenacity`:
```python
from tenacity import retry, wait_exponential, stop_after_attempt
@retry(wait=wait_exponential(multiplier=1, min=2, max=10), stop=stop_after_attempt(5))
def call_external():
requests.get("https://api.example.com")
```
---
#### **208. What is the "cache stampede" problem? How to prevent it?**
**Answer:**
- **Problem**: Many requests simultaneously regenerate an expired cache.
- **Fix**: Use **cache lock** + **background refresh**:
```python
with cache.lock("data_lock"):
if not cache.get("data"):
cache.set("data", generate_data(), timeout=60)
```
---
#### **209. How do you validate XML requests in Flask?**
**Answer:** Use `lxml` with XSD:
```python
from lxml import etree
def validate_xml(xml_data, xsd_path):
schema = etree.XMLSchema(etree.parse(xsd_path))
parser = etree.XMLParser(schema=schema)
etree.fromstring(xml_data, parser)
```
---
#### **210. How do you serve gRPC endpoints alongside Flask?**
**Answer:** Run gRPC server in a separate thread:
```python
from concurrent.futures import ThreadPoolExecutor
def serve_grpc():
server = grpc.server(ThreadPoolExecutor(max_workers=10))
my_pb2_grpc.add_MyServiceServicer_to_server(MyServicer(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()
# In app startup:
Thread(target=serve_grpc, daemon=True).start()
```
---
#### **211. How do you implement idempotency keys for API requests?**
**Answer:**
1. Client sends `Idempotency-Key: <uuid>` header
2. Server stores key + response in Redis:
```python
def create_resource():
key = request.headers.get("Idempotency-Key")
if key and redis.get(key):
return redis.get(key) # Return cached response
# Process request...
redis.setex(key, 3600, response)
return response
```
---
#### **212. How do you handle "slowloris" attacks in Flask?**
**Answer:** Mitigate at **reverse proxy level**:
- Nginx: `client_header_timeout 10s; client_body_timeout 10s;`
- Gunicorn: `--timeout 30` (kills slow requests)
---
#### **213. What is "header injection"? How to prevent in Flask?**
**Answer:**
- **Attack**: Injecting `\r\n` into headers to add malicious headers.
- **Prevention**: Flask auto-validates header names/values. Never use `make_response().headers.add()` with user input.
---
#### **214. How do you implement A/B testing in Flask?**
**Answer:** Use `g` to assign test groups:
```python
@app.before_request
def assign_test_group():
if "test_group" not in session:
session["test_group"] = random.choice(["A", "B"])
g.test_group = session["test_group"]
# In template:
{% if g.test_group == "A" %}
<!-- Variant A -->
{% endif %}
```
---
#### **215. How do you log SQL queries in production without performance impact?**
**Answer:**
- **Staging**: `app.config["SQLALCHEMY_RECORD_QUERIES"] = True`
- **Production**: Sample 1% of requests:
```python
if random.random() < 0.01:
app.logger.info(f"Slow query: {query} ({duration}ms)")
```
---
#### **216. How do you handle timezone conversions for user-specific timezones?**
**Answer:** Store user timezone in DB, convert in templates:
```python
# In context processor:
@app.context_processor
def inject_timezone():
if current_user.is_authenticated:
return {"user_tz": pytz.timezone(current_user.timezone)}
return {"user_tz": pytz.utc}
# In template (with Flask-Babel):
{{ datetime | to_user_timezone(user_tz) | format_datetime("short") }}
```
---
#### **217. How do you implement "password rotation" policies?**
**Answer:**
1. Track `password_changed_at` in user model
2. Force reset if older than 90 days:
```python
@app.before_request
def check_password_age():
if current_user.is_authenticated and current_user.password_age > 90:
return redirect(url_for("auth.change_password"))
```
---
#### **218. How do you secure Flask against deserialization attacks (e.g., pickle)?**
**Answer:**
- **NEVER** use `pickle` with untrusted data
- For sessions: Use `Flask-Session` with `redis` backend (not filesystem)
- For APIs: Prefer JSON over pickle/YAML
---
#### **219. How do you implement dynamic rate limiting (e.g., per user)?**
**Answer:** With `Flask-Limiter` + custom key:
```python
limiter = Limiter(
app,
key_func=lambda: f"user_{g.user.id if g.user else 'anon'}"
)
@limiter.limit("5/minute; 100/hour")
@app.route("/api/data")
def api_data():
...
```
---
#### **220. How do you handle "hot partitions" in DynamoDB with Flask?**
**Answer:** For high-write tables:
1. Use **hash+sort keys** (e.g., `user_id` + `timestamp`)
2. Add **random suffix** to partition key:
```python
partition_key = f"{user_id}#{random.randint(1, 100)}"
```
---
#### **221. How do you validate incoming webhooks from GitHub?**
**Answer:** Verify signature with secret:
```python
def verify_github_webhook(data, signature):
secret = app.config["GITHUB_WEBHOOK_SECRET"]
hmac_hash = hmac.new(secret, data, "sha256")
return hmac.compare_digest(f"sha256={hmac_hash.hexdigest()}", signature)
```
---
#### **222. What is the "Billion Laughs" attack? How to prevent in Flask XML?**
**Answer:**
- **Attack**: XML bomb with recursive entities
- **Prevention**: Disable entity expansion:
```python
parser = etree.XMLParser(resolve_entities=False, no_network=True)
```
---
#### **223. How do you implement "dark launches" for new features?**
**Answer:**
1. Wrap feature in feature flag:
```python
if feature_flags.is_active("new_checkout", user=current_user):
return new_checkout_flow()
else:
return legacy_checkout()
```
2. Gradually enable for internal users → 1% of traffic → 100%
---
#### **224. How do you debug a deadlock in SQLAlchemy transactions?**
**Answer:** Steps:
1. Enable PostgreSQL deadlock logs: `log_lock_waits = on`
2. Use `pg_blocking_pids()` to find blockers
3. Reduce transaction scope + add `FOR UPDATE SKIP LOCKED`
---
#### **225. How do you serve ML models in Flask without blocking the event loop?**
**Answer:** Offload to Celery + cache predictions:
```python
@app.route("/predict")
def predict():
result = predict_task.apply_async(args=[input_data])
return jsonify({"task_id": result.id})
@celery.task
def predict_task(data):
return model.predict(data)
```
---
#### **226. How do you implement "time travel" for data (point-in-time queries)?**
**Answer:** Use **temporal tables** (PostgreSQL):
```sql
CREATE TABLE users (
id SERIAL PRIMARY KEY,
name TEXT,
valid_from TIMESTAMPTZ DEFAULT NOW(),
valid_to TIMESTAMPTZ DEFAULT 'infinity'
);
```
Query historical data:
```python
User.query.filter(
User.valid_from <= target_time,
User.valid_to > target_time
).first()
```
---
#### **227. How do you handle "phantom reads" in Flask transactions?**
**Answer:** Use **SERIALIZABLE isolation level**:
```python
with db.session.begin(isolation_level="SERIALIZABLE"):
# Critical section
```
*Trade-off*: Higher conflict rate → retry logic required.
---
#### **228. How do you integrate Flask with AWS Lambda?**
**Answer:** Use `Zappa` or `AWS WSGI`:
```python
# handler.py
from your_app import app
def lambda_handler(event, context):
from mangum import Mangum
return Mangum(app)(event, context)
```
*Deploy with:* `zappa deploy production`
---
#### **229. How do you validate OpenAPI schemas automatically?**
**Answer:** Use `flask-openapi-validator`:
```python
from flask_openapi_validator import validate
app = Flask(__name__)
validator = OpenAPIValidator("openapi.yaml").validate(app)
@validate("get", "/users/{user_id}")
def get_user(user_id):
...
```
---
#### **230. How do you implement "feature flags" in Flask?**
**Answer:** Use `ldclient` (LaunchDarkly):
```python
from ldclient import get
flag = get().variation("new-checkout", {"key": current_user.id}, False)
if flag:
# Use new feature
```
---
#### **231. How do you handle "cache penetration" (bogus requests)?**
**Answer:**
1. **Bloom filters** for known keys
2. **Cache empty results** with short TTL:
```python
user = cache.get(f"user:{id}")
if user is None:
user = db.query(...).first()
cache.setex(f"user:{id}", 60, user or "NOT_FOUND")
```
---
#### **232. How do you implement "chained microservices" with Flask?**
**Answer:** Use **OpenTelemetry** + **asyncio.gather**:
```python
async def get_user_data(user_id):
profile = await fetch_profile(user_id)
orders = await fetch_orders(user_id)
return {**profile, "orders": orders}
```
*Configure trace context propagation* for service boundaries.
---
#### **233. How do you secure Flask against SSRF attacks?**
**Answer:**
1. Block internal IPs:
```python
def is_internal_ip(url):
hostname = urlparse(url).hostname
return ipaddress.ip_address(hostname).is_private
```
2. Use allowlists for outbound requests
---
#### **234. How do you implement "compensating transactions" for distributed systems?**
**Answer:** Saga pattern with Celery:
```python
@celery.task
def create_order(order_data):
try:
charge_payment(order_data)
reserve_inventory(order_data)
except Exception:
cancel_payment.delay(order_data) # Compensating action
raise
```
---
#### **235. How do you handle "time skew" in JWT tokens?**
**Answer:** Configure leeway:
```python
app.config["JWT_LEEWAY"] = 120 # 2 minutes tolerance
```
*Critical for distributed systems with unsynchronized clocks.*
---
#### **236. How do you implement "query bursting" for high-load endpoints?**
**Answer:** Batch similar requests:
```python
@batch_requests(max_size=10, timeout=0.1)
def get_user(user_ids):
return User.query.filter(User.id.in_(user_ids)).all()
```
*Use `aiocache` for async batching.*
---
#### **237. How do you validate multipart/form-data file uploads?**
**Answer:** Check content type + file signature:
```python
def validate_image(file):
# Check MIME type
if file.mimetype not in ["image/jpeg", "image/png"]:
return False
# Check magic bytes
header = file.read(4)
file.seek(0)
return header in [b"\xff\xd8\xff\xe0", b"\x89PNG"]
```
---
#### **238. How do you implement "circuit breakers" for Celery tasks?**
**Answer:** Use `celery.contrib.abortable`:
```python
from celery.contrib.abortable import AbortableTask
@app.task(bind=True, base=AbortableTask)
def process_data(self, data):
if self.is_aborted():
return "ABORTED"
# Process...
```
---
#### **239. How do you debug "worker timeout" in Gunicorn?**
**Answer:** Steps:
1. Increase `--timeout` temporarily
2. Profile slow requests: `gunicorn --preload --log-level debug`
3. Check for blocking calls (DB, network)
---
#### **240. How do you implement "data masking" for PII in logs?**
**Answer:** Custom log filter:
```python
class MaskFilter(logging.Filter):
def filter(self, record):
record.msg = re.sub(r"\b\d{4}[- ]?\d{4}[- ]?\d{4}\b", "****-****-****", record.msg)
return True
app.logger.addFilter(MaskFilter())
```
---
#### **241. How do you handle "cross-database joins" in Flask?**
**Answer:** Avoid if possible. If required:
1. Use **application-level joins** (fetch data from both DBs → merge in Python)
2. For PostgreSQL: `postgres_fdw` extension for foreign tables
---
#### **242. How do you implement "real-time config reload" without restarts?**
**Answer:** Watch config file + reload:
```python
import watchfiles
def watch_config():
for _ in watchfiles.watch("config.yaml"):
app.config.from_yaml("config.yaml")
Thread(target=watch_config, daemon=True).start()
```
---
#### **243. How do you secure Flask against "HTTP Request Smuggling"?**
**Answer:**
1. Set `proxy_set_header Connection "";` in Nginx
2. Use `gunicorn --proxy-allow-from 127.0.0.1`
3. Validate `Content-Length` vs `Transfer-Encoding`
---
#### **244. How do you implement "geo-distributed sessions" for global apps?**
**Answer:** Use **Redis Cluster** with Gossip protocol:
```python
app.config["SESSION_REDIS"] = redis.RedisCluster(
startup_nodes=[{"host": "redis1", "port": "6379"}]
)
```
---
#### **245. How do you validate JSON-LD requests?**
**Answer:** Use `pyld` library:
```python
from pyld import jsonld
def validate_jsonld(data):
try:
jsonld.expand(data)
return True
except jsonld.JsonLdError:
return False
```
---
#### **246. How do you implement "zero-downtime deployments" with Flask?**
**Answer:**
1. Blue/green deployments via load balancer
2. Gunicorn hot reload: `kill -HUP <master_pid>`
3. Health checks to drain old instances
---
#### **247. How do you handle "timezone ambiguity" during DST transitions?**
**Answer:** Store all datetimes in **UTC**, convert to local only for display:
```python
# Incoming request (with timezone)
dt = datetime.fromisoformat(request.json["time"]).astimezone(timezone.utc)
# Outgoing response
local_tz = pytz.timezone("America/New_York")
response_time = dt.astimezone(local_tz).isoformat()
```
---
#### **248. How do you implement "idempotent database migrations"?**
**Answer:** Alembic revision scripts should be:
1. **Re-runnable**: `CREATE TABLE IF NOT EXISTS`
2. **Idempotent**: Check state before applying changes
```python
def upgrade():
if not table_exists("users"):
op.create_table(...)
```
---
#### **249. How do you debug "connection pool exhaustion"?**
**Answer:**
1. Log pool status: `app.logger.info(f"Pool: {db.engine.pool.status()}")`
2. Increase pool size: `SQLALCHEMY_ENGINE_OPTIONS={"pool_size": 20}`
3. Fix leaks: Ensure `session.close()` in `teardown_request`
---
#### **250. How do you implement "multi-region failover" for Flask apps?**
**Answer:**
1. Active-active setup with global load balancer (AWS Global Accelerator)
2. Synchronous DB replication (AWS Aurora Global Database)
3. Health checks to redirect traffic on failure
---
#### **251. How do you secure Flask against "WebSocket hijacking"?**
**Answer:**
1. Validate `Origin` header in WebSocket handshake
2. Use `wss://` (WebSocket Secure)
3. Set `SameSite=Strict` on session cookies
---
#### **252. How do you implement "request tracing" for async tasks?**
**Answer:** Propagate trace context to Celery:
```python
# In Flask request
current_span = trace.get_current_span()
task.apply_async(..., headers={"traceparent": current_span.context})
# In Celery task
context = TraceContextTextMapPropagator().extract(
carrier={"traceparent": headers["traceparent"]}
)
with tracer.start_as_current_span("celery_task", context=context):
# Task logic
```
---
#### **253. How do you handle "caching stale data" during deploys?**
**Answer:**
1. **Cache versioning**: `cache.set(f"v2:{key}", value)`
2. **Graceful cache invalidation**:
- Deploy new version → keep old cache keys
- After traffic shift → delete old keys
---
#### **254. How do you implement "dynamic CORS origins" based on user roles?**
**Answer:** Custom CORS decorator:
```python
def cors_required(f):
@wraps(f)
def decorated(*args, **kwargs):
origin = request.headers.get("Origin")
if origin not in current_user.allowed_origins:
return "Forbidden", 403
response = f(*args, **kwargs)
response.headers["Access-Control-Allow-Origin"] = origin
return response
return decorated
```
---
#### **255. How do you validate "signed URLs" for private S3 access?**
**Answer:** Use AWS SDK:
```python
def verify_s3_url(url, secret_key):
parsed = urlparse(url)
query = parse_qs(parsed.query)
signature = query.pop("Signature")[0]
# Recreate string to sign
string_to_sign = f"GET\n\n\n{query['Expires'][0]}\n{parsed.path}"
computed = hmac.new(secret_key, string_to_sign, "sha1").hexdigest()
return hmac.compare_digest(signature, computed)
```
---
#### **256. How do you implement "database connection multiplexing"?**
**Answer:** Use `pgbouncer` in transaction pooling mode:
```ini
[databases]
mydb = host=localhost port=5432 dbname=mydb
[pgbouncer]
pool_mode = transaction
server_reset_query = DISCARD ALL
```
---
#### **257. How do you handle "time-based race conditions" in Flask?**
**Answer:** Use **optimistic locking**:
```python
class Order(db.Model):
version_id = db.Column(db.Integer, nullable=False)
# Update logic:
order = Order.query.get(1)
if order.version_id != request.json["version"]:
abort(409, "Stale data")
order.total = new_total
order.version_id += 1
```
---
#### **258. How do you implement "real-time config validation" for feature flags?**
**Answer:** Use JSON Schema + webhook:
```python
@app.route("/flags/validate", methods=["POST"])
def validate_flags():
try:
jsonschema.validate(request.json, flag_schema)
return "", 204
except ValidationError as e:
return str(e), 400
```
---
#### **259. How do you secure Flask against "HTTP Desync Attacks"?**
**Answer:**
1. Set `Connection: close` in Nginx
2. Use `gunicorn --proxy-allow-from 127.0.0.1`
3. Validate `Content-Length` headers strictly
---
#### **260. How do you implement "distributed rate limiting" across instances?**
**Answer:** Use **Redis** with sliding window:
```python
def is_rate_limited(user_id, limit=100, window=3600):
key = f"rate_limit:{user_id}"
now = time.time()
pipe = redis.pipeline()
pipe.zremrangebyscore(key, 0, now - window)
pipe.zadd(key, {now: now})
pipe.expire(key, window)
count, _ = pipe.execute()
return count[1] > limit
```
---
#### **261. How do you handle "time travel testing" for timezone-dependent logic?**
**Answer:** Use `freezegun` in tests:
```python
from freezegun import freeze_time
@freeze_time("2023-01-01 12:00:00")
def test_summer_time():
assert is_summer_time() == False
```
---
#### **262. How do you implement "circuit breakers" for database connections?**
**Answer:** With `pybreaker`:
```python
class DBBreaker(CircuitBreaker):
def __init__(self):
super().__init__(fail_max=5, reset_timeout=60)
def is_failed(self):
return super().is_failed() or self.db_unavailable
breaker = DBBreaker()
@breaker
def get_db():
if not check_db_health():
breaker.db_unavailable = True
raise DBUnavailable
return db.session
```
---
#### **263. How do you debug "memory bloat" from large query results?**
**Answer:**
1. Use `yield_per()` for chunked results:
```python
for user in User.query.yield_per(100).enable_eagerloads(False):
process(user)
```
2. Avoid `all()` → iterate directly on query
---
#### **264. How do you implement "secure cookie attributes" for mobile apps?**
**Answer:**
```python
app.config.update(
SESSION_COOKIE_SECURE=True,
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SAMESITE="None", # Required for mobile WebView
SESSION_COOKIE_DOMAIN=".yourdomain.com" # Subdomain sharing
)
```
---
#### **265. How do you handle "cross-service authentication" in microservices?**
**Answer:** Use **mTLS** or **JWT service tokens**:
```python
# Service A calls Service B
token = jwt.encode(
{"iss": "service-a", "aud": "service-b"},
private_key,
algorithm="RS256"
)
requests.get("https://service-b/api", headers={"Authorization": f"Bearer {token}"})
```
---
#### **266. How do you implement "dynamic blueprint registration" based on config?**
**Answer:**
```python
def register_blueprints(app):
for feature in app.config["ENABLED_FEATURES"]:
module = import_module(f"features.{feature}.views")
app.register_blueprint(module.bp)
```
---
#### **267. How do you validate "signed cookies" for stateless auth?**
**Answer:** Use `itsdangerous`:
```python
from itsdangerous import URLSafeTimedSerializer
serializer = URLSafeTimedSerializer(app.secret_key)
def create_token(user_id):
return serializer.dumps(user_id, salt="auth")
def verify_token(token, max_age=3600):
return serializer.loads(token, salt="auth", max_age=max_age)
```
---
#### **268. How do you handle "time-based security tokens" (e.g., TOTP)?**
**Answer:** With `pyotp`:
```python
def verify_totp(token, secret):
totp = pyotp.TOTP(secret)
return totp.verify(token, valid_window=1) # 1 interval leeway
```
---
#### **269. How do you implement "database encryption at rest" with Flask?**
**Answer:**
1. **Application-level**: Encrypt fields with `cryptography.fernet`
2. **Database-level**: Use PostgreSQL `pgcrypto` or AWS RDS encryption
---
#### **270. How do you debug "flaky tests" in Flask?**
**Answer:**
1. Isolate randomness (fix seeds)
2. Check time dependencies (`freezegun`)
3. Use `pytest-repeat`: `pytest --count=100 test_flaky.py`
---
#### **271. How do you implement "request coalescing" for duplicate queries?**
**Answer:** Cache pending requests:
```python
_pending_requests = {}
def get_user(user_id):
if user_id in _pending_requests:
return _pending_requests[user_id].result()
future = concurrent.futures.Future()
_pending_requests[user_id] = future
def fetch():
try:
result = db.query(user_id)
future.set_result(result)
finally:
del _pending_requests[user_id]
Thread(target=fetch).start()
return future.result()
```
---
#### **272. How do you handle "time zone transitions" for recurring events?**
**Answer:** Store events in UTC, calculate local times on demand:
```python
def get_event_local_time(event, user_tz):
utc_time = event.start_time # Stored in UTC
local_time = utc_time.astimezone(pytz.timezone(user_tz))
# Handle DST transitions automatically
return local_time
```
---
#### **273. How do you implement "secure headers" for modern browsers?**
**Answer:** Use `Flask-Talisman` with strict policy:
```python
csp = {
'default-src': "'self'",
'script-src': ["'self'", "'sha256-...'", "trusted.cdn.com"],
'style-src': ["'self'", "'unsafe-inline'"],
'img-src': ["'self'", "data:"],
'connect-src': ["'self'", "api.trusted.com"]
}
Talisman(app, content_security_policy=csp, feature_policy="camera 'none'; microphone 'none'")
```
---
#### **274. How do you validate "OAuth2 token binding"?**
**Answer:** Bind tokens to client TLS certificates:
```python
# In token validation
if request.certificate_fingerprint != token["cnf"]["x5t#S256"]:
abort(401, "Token binding mismatch")
```
---
#### **275. How do you implement "database sharding" for multi-tenancy?**
**Answer:**
1. **Shard key**: `tenant_id % 10` → 10 shards
2. **Router**:
```python
def get_shard(tenant_id):
shard_id = tenant_id % app.config["SHARD_COUNT"]
return db.get_engine(app, f"shard_{shard_id}")
```
---
#### **276. How do you handle "time-based access policies"?**
**Answer:** Use policy engine like **Casbin**:
```python
from casbin import Enforcer
enforcer = Enforcer("model.conf", "policy.csv")
def check_access(user, resource):
return enforcer.enforce(user.role, resource, "read", current_time.isoformat())
```
---
#### **277. How do you implement "secure file downloads" with expiring links?**
**Answer:** Signed S3 URLs:
```python
def generate_download_url(file_key):
return s3.generate_presigned_url(
"get_object",
Params={"Bucket": "my-bucket", "Key": file_key},
ExpiresIn=300 # 5 minutes
)
```
---
#### **278. How do you debug "connection leaks" in SQLAlchemy?**
**Answer:** Enable echo and log pool status:
```python
app.config["SQLALCHEMY_ECHO"] = True
@app.before_request
def log_pool():
app.logger.debug(f"Pool: {db.engine.pool.status()}")
```
---
#### **279. How do you implement "real-time feature flag evaluation"?**
**Answer:** Use LaunchDarkly streaming API:
```python
from ldclient import get()
def is_feature_enabled(feature, user):
return get().variation(feature, user, False)
# Auto-updates when flags change
get().start()
```
---
#### **280. How do you handle "time zone ambiguity" in recurring events?**
**Answer:** Store recurrence rules in UTC:
```python
# Event repeats every Monday at 9 AM UTC
rrule = "FREQ=WEEKLY;BYDAY=MO;BYHOUR=9;BYMINUTE=0"
```
*Convert to local time only when rendering the calendar.*
---
#### **281. How do you implement "secure password recovery"?**
**Answer:**
1. One-time token with short TTL (15 mins)
2. Invalidate token after use
3. Rate limit recovery attempts
```python
token = serializer.dumps(user.id, salt="recovery", max_age=900)
```
---
#### **282. How do you validate "WebAuthn" (FIDO2) authentication?**
**Answer:** Use `flask_webauthn`:
```python
from flask_webauthn import WebAuthn
webauthn = WebAuthn(app,
rp_name="My App",
rp_id="example.com",
url_prefix="/webauthn"
)
# Registration flow
@app.route("/register/begin", methods=["POST"])
def register_begin():
return webauthn.registration_begin(user)
```
---
#### **283. How do you implement "database row versioning"?**
**Answer:** Use SQLAlchemy **Versioning** extension:
```python
from sqlalchemy_continuum import make_versioned
make_versioned()
class Article(db.Model):
__versioned__ = {}
# Query history:
Article.versions.filter(Article.id == 1).all()
```
---
#### **284. How do you handle "time-based rate limiting" (e.g., 1000 requests/day)?**
**Answer:** Redis counter with sliding window:
```python
def is_rate_limited(user_id, limit=1000, window=86400):
key = f"rate_limit:{user_id}"
now = time.time()
pipe = redis.pipeline()
pipe.zremrangebyscore(key, 0, now - window)
pipe.zadd(key, {now: now})
pipe.expire(key, window)
count = pipe.execute()[1]
return count > limit
```
---
#### **285. How do you implement "secure session invalidation" on password change?**
**Answer:** Bump session salt:
```python
@app.route("/change-password", methods=["POST"])
def change_password():
current_user.session_salt = secrets.token_hex(16)
db.session.commit()
logout_user() # Invalidates all sessions
```
---
#### **286. How do you debug "asyncio event loop conflicts" in Flask?**
**Answer:**
1. Isolate async code: `asyncio.run_in_executor()`
2. Never share event loops between threads
3. Use `async-exit-stack` for cleanup
---
#### **287. How do you implement "multi-factor authentication" with WebAuthn?**
**Answer:** Combine password + WebAuthn:
```python
@app.route("/login", methods=["POST"])
def login():
if not verify_password():
return "Invalid password", 401
# Proceed to WebAuthn challenge
return webauthn.authentication_begin()
```
---
#### **288. How do you handle "time-based data retention policies"?**
**Answer:** Celery task for nightly cleanup:
```python
@celery.task
def delete_old_data():
cutoff = datetime.utcnow() - timedelta(days=365)
db.session.query(Logs).filter(Logs.created_at < cutoff).delete()
db.session.commit()
```
---
#### **289. How do you implement "secure API key rotation"?**
**Answer:**
1. Store hashed keys (bcrypt)
2. Support multiple active keys per user
3. Deprecate old keys gradually:
```python
@app.before_request
def check_api_key():
key = request.headers.get("X-API-KEY")
if key in user.deprecated_keys:
return "Key deprecated. Use new key.", 401
```
---
#### **290. How do you validate "JWT claims" for custom business rules?**
**Answer:** With `Flask-JWT-Extended` claims loader:
```python
@jwt.token_in_blocklist_loader
def check_blocklist(jwt_header, jwt_payload):
return jwt_payload["jti"] in BLOCKLIST
@jwt.additional_claims_loader
def add_claims(identity):
return {"is_admin": User.query.get(identity).is_admin}
```
---
#### **291. How do you implement "database connection failover"?**
**Answer:** Use SQLAlchemy **read replicas** + `sqlalchemy-utils`:
```python
app.config["SQLALCHEMY_BINDS"] = {
"primary": "postgresql://primary/db",
"replica1": "postgresql://replica1/db"
}
class RoutingSession(Session):
def get_bind(self, mapper=None, clause=None):
if self._flushing or request.method == "POST":
return engine.binds["primary"]
return engine.binds[random.choice(["replica1", "replica2"])]
```
---
#### **292. How do you handle "time-based access revocation"?**
**Answer:** Store access policies with end times:
```python
class AccessGrant(db.Model):
user_id = db.Column(db.Integer, db.ForeignKey("user.id"))
expires_at = db.Column(db.DateTime)
def has_access(user, resource):
grant = AccessGrant.query.filter(
AccessGrant.user_id == user.id,
AccessGrant.resource == resource,
AccessGrant.expires_at > datetime.utcnow()
).first()
return grant is not None
```
---
#### **293. How do you implement "secure file uploads" with virus scanning?**
**Answer:** Offload to Celery:
```python
@app.route("/upload", methods=["POST"])
def upload():
file = request.files["file"]
file_path = save_temp_file(file)
scan_file.delay(file_path) # Async scan
return "Uploaded", 202
@celery.task
def scan_file(file_path):
result = clamd.scan(file_path)
if result["infected"]:
os.remove(file_path)
```
---
#### **294. How do you debug "memory leaks in C extensions"?**
**Answer:**
1. `tracemalloc` + `objgraph` for Python objects
2. `valgrind` for C-level leaks:
```bash
valgrind --leak-check=full gunicorn app:app
```
---
#### **295. How do you implement "real-time feature flag analytics"?**
**Answer:** Track flag evaluations:
```python
@ldclient.track("new-checkout")
def checkout():
# ...
```
*Send events to analytics pipeline (e.g., Snowflake).*
---
#### **296. How do you handle "time-based data partitioning" in databases?**
**Answer:** PostgreSQL declarative partitioning:
```sql
CREATE TABLE logs (
id SERIAL,
message TEXT,
created_at TIMESTAMPTZ NOT NULL
) PARTITION BY RANGE (created_at);
CREATE TABLE logs_2023q1 PARTITION OF logs
FOR VALUES FROM ('2023-01-01') TO ('2023-04-01');
```
---
#### **297. How do you implement "secure session regeneration" on privilege changes?**
**Answer:**
```python
def elevate_privilege(user):
logout_user()
login_user(user, fresh=True) # New session
session.regenerate() # Flask-Login method
```
---
#### **298. How do you validate "OAuth2 PKCE" flows in Flask?**
**Answer:** Use `Authlib`:
```python
@app.route("/authorize")
def authorize():
return current_app.auth_client.authorize_redirect(
redirect_uri=url_for("callback", _external=True),
code_verifier=session["code_verifier"] # Stored during login
)
```
---
#### **299. How do you implement "database encryption for specific columns"?**
**Answer:** SQLAlchemy **TypeDecorator**:
```python
from cryptography.fernet import Fernet
class EncryptedText(TypeDecorator):
impl = Text
def __init__(self, key, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cipher = Fernet(key)
def process_bind_param(self, value, dialect):
return self.cipher.encrypt(value.encode())
def process_result_value(self, value, dialect):
return self.cipher.decrypt(value).decode()
```
---
#### **300. How do you debug "asyncio timeouts" in ASGI apps?**
**Answer:**
1. Increase timeout: `hypercorn --timeout 120`
2. Profile async tasks:
```python
import asyncio
async def debug_timeout():
task = asyncio.create_task(slow_operation())
try:
await asyncio.wait_for(task, timeout=30)
except asyncio.TimeoutError:
task.cancel()
app.logger.error("Task timed out")
```
---
**Part 4 (Questions 301–400) is ready!**
Sign in with Wallet
Connect another wallet