Hands-on Quiz: AI for Data Engineering
Test your ability to design and write production-grade AI-powered data pipelines.
Challenge 1: Safe SQL Extraction
Write a Python function using sqlglot that verifies an incoming SQL query generated by an LLM is strictly a read-only SELECT query, contains no mutating statements (DROP, DELETE, UPDATE), and caps the result with LIMIT 100.
import sqlglot
from sqlglot.expressions import Select, Drop, Delete, Update
def ensure_safe_query(raw_sql: str) -> str:
# Your implementation here
pass
View Solution
import sqlglot
from sqlglot.expressions import Select, Drop, Delete, Update
def ensure_safe_query(raw_sql: str) -> str:
parsed = sqlglot.parse_one(raw_sql)
for expr in parsed.find_all((Drop, Delete, Update)):
raise ValueError("Mutating statement detected!")
if not isinstance(parsed, Select):
raise ValueError("Only SELECT queries permitted.")
if not parsed.args.get("limit"):
parsed = parsed.limit(100)
return str(parsed)
Challenge 2: Cache Key Generation for LLM Transforms
To prevent re-running costly LLM inferences on identical input rows across batch pipeline runs, implement a deterministic SHA-256 cache key function.
import hashlib
import json
def get_row_cache_key(model_name: str, row_dict: dict) -> str:
# Your implementation here
pass
View Solution
import hashlib
import json
def get_row_cache_key(model_name: str, row_dict: dict) -> str:
canonical_json = json.dumps(row_dict, sort_keys=True)
payload = f"{model_name}:{canonical_json}".encode("utf-8")
return hashlib.sha256(payload).hexdigest()
Find this content helpful?
☕ Buy me a coffee