Forem: osman uygar köse

This AI Reads Your Data Like a Senior Analyst - And It Works in Milliseconds

osman uygar köse — Mon, 26 Jan 2026 15:36:39 +0000

I Built an AI That Actually Understands Your Data (And It's Smarter Than Most Data Analysts) 🤯

You know what drives me crazy?

Data tools that show you numbers but leave you hanging. They're like that friend who says "I've got news" and then just... stares at you.

We built SQLatte to convert natural language into SQL queries. Cool, right? Users loved it. But then came the existential crisis:

User: "Show me today's sales"

SQLatte: Returns a table with numbers

User: "Okay... so is this good or bad?"

SQLatte: cricket sounds 🦗

That's when we realized: Raw data is useless without context.

🎯 The Real Problem: Numbers Don't Tell Stories

Imagine you're running an e-commerce platform. Your dashboard shows:

Sales Today: $45,000
Orders: 23

Now answer these:

✅ Is this good? (Compared to yesterday)
⚠️ Is the data complete? (It's 11 AM - day isn't over)
📈 What's the trend?
💡 What should I do about it?

Spoiler alert: Most BI tools can't answer these questions. They just dump data and run away.

💡 Enter: The AI-Powered Insights Engine

So we built something different. Something that doesn't just query data - it understands it.

Here's what makes it special:

1. Temporal Awareness - It Knows What Time It Is

Our engine tracks:

context = {
    'current_time': {
        'date': '2025-01-26',
        'hour': 11,
        'completion_percent': 45.8,  # 46% of day passed
        'is_business_hours': True
    }
}

Why this matters:

❌ Dumb insight (without context):

"Today's sales are $45,000 - below average!"

✅ Smart insight (with context):

"⏰ Today's sales are $45,000 (46% of day complete). At this pace, you'll hit ~$98,000 by EOD - that's 15% above target! 📈"

See the difference? One makes you panic. The other makes you smile.

2. Incomplete Data Detection - No More False Alarms

The engine automatically detects incomplete data:

if '2025-01-26' in last_date_str:
    context['incomplete_data_warning'] = True
    # Warn the LLM to handle with care

Real-world example:

You run a query at 2 PM: "Compare today vs yesterday"

Without detection:

Today: $50K
Yesterday: $120K
😱 PANIC! Sales crashed 58%!

With detection:

⚠️ WARNING: Today's data is incomplete (58% of day passed)
Projected EOD: ~$86K (28% below yesterday)
💡 Monitor until 6 PM for accurate comparison

No heart attacks. Just facts.

🏗️ Architecture: Three Modes, One Goal

We designed three modes because one size doesn't fit all:

Mode 1: LLM-Only (Maximum Intelligence)

insights:
  enabled: true
  mode: llm_only
  max_insights: 3

Pros:

Richest, most contextual insights
Natural language explanations
Understands business logic

Cons:

Costs money (LLM API calls)
Slower (~1.5s latency)
Rate limit risks

Best for: Premium features, high-value queries

Mode 2: Statistical-Only (Lightning Fast)

insights:
  enabled: true
  mode: statistical_only

What it analyzes:

📊 Trend detection in numeric columns
⚡ Anomaly identification (outliers)
📈 Growth/decline calculations
🎯 Basic statistics

Pros:

⚡ Fast (<100ms)
💰 Free (no API costs)
🔄 Predictable results

Best for: Free tiers, real-time dashboards

Mode 3: Hybrid (Recommended) ⭐

insights:
  enabled: true
  mode: hybrid
  include_statistical: true

The best of both worlds:

Try LLM first → Get contextual analysis
If LLM fails → Fallback to statistical
Combine both → Maximum coverage

The sweet spot: Balance between cost and quality.

💻 Technical Deep Dive: How It Actually Works

The Core Engine

class LLMInsightsEngine:
    def __init__(self, llm_provider, config):
        self.llm_provider = llm_provider
        self.mode = config.get('mode', 'hybrid')
        self.max_insights = config.get('max_insights', 3)

        # Temporal context tracking
        self.today = datetime.now().date()
        self.current_hour = datetime.now().hour

    def generate_insights(
        self,
        columns: List[str],
        data: List[List[Any]],
        user_question: str,
        schema_info: str,
        sql_query: str
    ) -> List[Dict]:
        """Main entry point - the magic happens here"""

        if not self.enabled or not data:
            return []

        # Smart mode selection
        if self.mode == 'llm_only':
            return self._generate_llm_insights(...)
        elif self.mode == 'statistical_only':
            return self._generate_statistical_insights(...)
        else:  # hybrid
            return self._hybrid_analysis(...)

Prompt Engineering: The Secret Sauce

Here's how we talk to the LLM (this took us 47 iterations to get right):

def _build_llm_prompt(self, context: Dict) -> str:
    prompt = f"""You're a data analyst assistant. Analyze query results 
    and generate actionable business insights.

    USER QUESTION: {context['question']}

    SCHEMA: {context['schema']}

    RESULTS:
    - Row count: {context['row_count']}
    - Columns: {', '.join(context['columns'])}
    - Numeric columns: {context['numeric_columns']}

    TIME CONTEXT:
    - Current: {context['current_time']['date']} {context['current_time']['hour']:02d}:00
    - Business hours: {'Yes' if context['current_time']['is_business_hours'] else 'No'}
    """

    if context.get('incomplete_data_warning'):
        prompt += f"""
        ⚠️ CRITICAL: Today's data is incomplete 
        ({context['current_time']['completion_percent']:.0f}% of day passed)
        """

    prompt += """
    TASK:
    1. Generate 2-3 meaningful, actionable insights
    2. Warn about incomplete/misleading data
    3. Return as JSON: {"insights": [...]}

    Focus on BUSINESS VALUE, not just statistics.
    """
    return prompt

Statistical Analysis: The Safety Net

When LLM isn't available or fails, we fall back to rule-based analysis:

def _generate_statistical_insights(
    self, 
    columns: List[str], 
    data: List[List[Any]]
) -> List[Dict]:
    insights = []

    # Analyze numeric columns for trends
    for col_idx, col_name in enumerate(columns):
        values = self._extract_numeric_column(col_idx, data)

        if len(values) >= 3:
            # Trend detection
            first_half = statistics.mean(values[:len(values)//2])
            second_half = statistics.mean(values[len(values)//2:])
            change = ((second_half - first_half) / first_half) * 100

            # Anomaly detection (outliers)
            if abs(change) > 20:
                insights.append({
                    'type': 'trend',
                    'icon': '📈' if change > 0 else '📉',
                    'severity': 'warning' if abs(change) > 50 else 'info',
                    'message': f"{col_name} shows {abs(change):.1f}% "
                              f"{'increase' if change > 0 else 'decrease'}"
                })

    return insights[:self.max_insights]

Thread-Safe Singleton Pattern

In a multi-threaded FastAPI environment, we need one engine instance:

class InsightsEngineSingleton:
    _instance = None
    _initialized = False

    @classmethod
    def initialize(cls, llm_provider, config):
        if cls._initialized:
            return cls._instance

        cls._instance = LLMInsightsEngine(llm_provider, config)
        cls._initialized = True
        return cls._instance

    @classmethod
    def get_instance(cls):
        return cls._instance

# Usage in FastAPI
from src.core.llm_insights_engine import get_insights_engine

engine = get_insights_engine()
insights = engine.generate_insights(columns, data, ...)

🎯 Real-World Impact: The Numbers Don't Lie

Use Case 1: E-Commerce Flash Sale

Query: "Show sales by category today"

Raw Output:

Category      | Sales
--------------|--------
Electronics   | $125,000
Fashion       | $89,000
Home & Living | $45,000

AI Insights:

📈 Electronics performing 230% above average today. 
   Likely a flash sale or campaign effect.

⏰ ALERT: It's 11 AM (46% of day). EOD projections:
   - Electronics: ~$273K
   - Fashion: ~$194K
   - Home & Living: ~$98K

💡 ACTION NEEDED: Check Electronics inventory. 
   High demand may cause stockouts by 3 PM.

Business Impact: Prevented stockout, secured additional $47K in sales.

Use Case 2: Security Incident Detection

Query: "Top attacked endpoints in last 24h"

Raw Output:

Endpoint        | Attack Count
----------------|-------------
/login          | 1,247
/admin          | 892
/api/v1/users   | 234

AI Insights:

⚠️ CRITICAL: /login receiving 8x normal traffic. 
   High probability of brute-force attack!

🔒 /admin endpoint also targeted. Check if rate 
   limiting is active.

💡 IMMEDIATE ACTIONS:
   1. Enable CAPTCHA on /login
   2. Reduce rate limit: 5 → 3 attempts
   3. Alert security team NOW

Business Impact: Stopped brute-force attack within 12 minutes of detection.

🎬 Try It Yourself

GitHub: github.com/osmanuygar/sqlatte

Quick Start:

git clone https://github.com/osmanuygar/sqlatte
cd sqlatte
pip install -r requirements.txt

# Edit config/config.yaml with your API keys
python run.py

# Open http://localhost:8000

🙋‍♂️ Questions? Feedback? Want to Collaborate?

I'm always happy to discuss data engineering, AI, and building better tools.

Email: osmanuygar@gmail.com

LinkedIn: www.linkedin.com/in/osman-uygar-kose-56785820

If you found this useful, give it a ❤️ and share it with your team!

Happy coding! ☕

Prompt Engineering a Barista: How SQLatte's Personality Transforms SQL into Conversations

osman uygar köse — Mon, 12 Jan 2026 16:36:00 +0000

The Problem: SQL Tools Feel Like Robots

Most NL2SQL tools have boring prompts:

# ❌ Generic, robotic
system_prompt = "You are a SQL assistant. Generate accurate queries."

We wanted something different: What if querying databases felt like ordering coffee from your favorite barista?

The Solution: SQLatte's Barista Prompt ☕

Here's our actual system prompt across Claude, Gemini, and VertexAI:

system_prompt = """You are SQLatte ☕ - a friendly AI assistant that helps 
users query their databases with natural language.

Your personality:
- Helpful and friendly, like a barista serving the perfect drink
- Knowledgeable about SQL and databases
- Can have casual conversations too
- Use coffee/brewing metaphors occasionally when appropriate

When users ask general questions (not about data):
- Respond naturally and helpfully
- If they seem lost, guide them on how to use SQLatte
- Be concise but friendly
"""

Why Baristas?

Good baristas:

🎯 Remember your order → Conversation memory
💬 Explain what they're making → Query explanations
🤝 Suggest something new → Follow-up questions
⚠️ Warn if it's too hot → Performance alerts

The metaphor isn't decoration—it's our UX framework.

The Three-Layer Prompt Architecture

1. Intent Detection

First, figure out what the user wants:

prompt = """Analyze this question: is it SQL or general chat?

Rules:
1. Data question → intent: "sql"
2. Greeting/chat → intent: "chat"
3. No tables selected → guide them

Format:
INTENT: sql or chat
CONFIDENCE: 0.0 to 1.0
REASONING: brief explanation
"""

2. SQL Generation (The Main Prompt)

prompt = f"""You are a SQL expert barista ☕. 

Table Schema: {schema_info}
User Question: {question}

🎯 RULES:
1. Valid SQL with JOINs
2. Table aliases (orders o, customers c)
3. LIMIT 100 for safety

⚡ PERFORMANCE:
4. If 'dt' column exists (YYYYMMDD partition):
   - For "recent/today" → dt >= '20251211' (last 2 days)
   - For "yesterday" → dt = '20251212'
   - ⚠️ ALWAYS use dt filters for speed!

📝 RESPONSE:
SQL: <query>
EXPLANATION: <what it does>
SUGGESTIONS: <2-3 follow-ups>

Remember: Be friendly, not robotic! ☕
"""

3. Chat Responses

For non-SQL questions, same warm personality.

Coffee Metaphors: What Works vs What Doesn't

✅ Good (Users Love):

"☕ Here's your data, freshly brewed! I used the date partition 
so this runs in milliseconds."

"⏱️ This might take a moment—like waiting for a French press. 
I'm scanning 2.3M rows."

🚫 Bad (Too Cheesy):

"That's not my cup of tea..." ❌
"Spilling the beans on your data..." ❌  
"Your query needs more grounds..." ❌

The Rule:

1-2 metaphors per conversation, not every message. Clarity over cleverness.

Real-World Impact: Before vs After

❌ Without Personality:

User: "show sales"
Bot: SELECT * FROM sales

User: "more details"
Bot: ERROR: I don't understand

✅ With Barista Prompt:

User: "show sales"
SQLatte: ☕ Here's your sales data! Filtered last 2 days for speed.
Found 1,247 transactions totaling $45,892.

Want to see:
- Top products?
- Sales by region?
- Hourly breakdown?

User: "by region"
SQLatte: Great! Here's the regional breakdown...
[remembers context]

Multi-Provider Implementation

Same personality, different APIs:

# Claude (native system role)
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    system=barista_prompt,
    messages=[{"role": "user", "content": question}]
)

# Gemini (system_instruction)
model = genai.GenerativeModel(
    "gemini-pro",
    system_instruction=barista_prompt
)

# VertexAI (prepend to message)
full_prompt = f"{barista_prompt}\n\n{question}"

Key insight: Different LLMs, same personality—adapt implementation, not the prompt.

Conversation Memory: The Secret Sauce

# Store last 5 exchanges
conversation_history = get_recent_messages(session_id, limit=5)

# Inject into prompt
context_summary = "\n".join([
    f"{msg['role']}: {msg['content']}" 
    for msg in conversation_history
])

enhanced_prompt = f"{question}\n\nRecent context:\n{context_summary}"

Result: "show me more" and "break down by region" just work naturally.

Prompt Evolution: Our Journey

V1: Too Generic ❌

"You are a SQL assistant. Be helpful."
# Result: Boring, unhelpful errors

V2: Too Much ❌

"You are SUPER EXCITED about SQL! Use TONS of coffee puns! ☕☕☕"
# Result: Annoying and unprofessional

V3: Just Right ✅

"Like a barista serving the perfect drink... 
Use metaphors occasionally when appropriate"
# Result: Warm, professional, memorable

Lessons for Your Prompts

✅ Do:

Pick a clear metaphor (barista, coach, teacher)
Set boundaries ("occasionally", "when appropriate")
Structure sections (RULES:, FORMAT:, PERFORMANCE:)
Test across providers (Claude ≠ Gemini ≠ GPT)
Add personality to errors ("Let me help..." not "ERROR")
Include context (conversation history)

❌ Don't:

Be vague ("be friendly" → how exactly?)
Force humor (puns in errors = bad UX)
Over-explain (users want answers, not essays)
One-size-fits-all (different prompts for different needs)

Copy-Paste Starter Code

from anthropic import Anthropic

class PersonalitySQL:
    def __init__(self, api_key: str):
        self.client = Anthropic(api_key=api_key)
        self.personality = """You are MyApp ☕ - like a friendly barista 
        serving perfect data. Helpful, knowledgeable, occasionally uses 
        coffee metaphors when appropriate."""

    def query(self, question: str, schema: str, history: list = []):
        # Add conversation context
        context = "\n".join([f"{m['role']}: {m['content']}" 
                            for m in history[-5:]])

        prompt = f"""Schema: {schema}
        Context: {context}
        Question: {question}

        Generate SQL and explain naturally."""

        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            system=self.personality,
            messages=[{"role": "user", "content": prompt}]
        )

        return response.content[0].text

# Usage
sql = PersonalitySQL("your-api-key")
result = sql.query("show me sales", "sales(id, amount, dt)")

Try SQLatte

GitHub: @osmanuygar
Project Link: https://github.com/osmanuygar/sqlatte

Key Takeaways

Metaphors create UX frameworks - "Like a barista" defines everything
Personality drives metrics - +148% engagement isn't luck
Context is critical - Memory makes AI feel intelligent
Boundaries prevent overuse - "Occasionally" is key
Test and iterate - Track engagement, not just accuracy
Warmth wins - Users remember how you made them feel

What personality have you engineered into your prompts? Share your system prompts in the comments! ☕

Building an AI tool? Remember: users don't want features, they want experiences. Make yours memorable.

The Art of Context Windows: Our AI Had Alzheimer's: Here's How We Taught It To Remember

osman uygar köse — Tue, 30 Dec 2025 15:10:18 +0000

It was 2 AM. Our Slack was on fire with support tickets.

"Your AI is broken."

"It keeps forgetting what I just asked."

"This is unusable."

The irony? Our AI was working perfectly. The problem was something far more fundamental.

🔥 The Problem Nobody Talks About

We built SQLatte - an AI that converts natural language to SQL. The demo looked magical:

User: "Show me top 10 customers by revenue"

AI: ✨ Generates perfect SQL, returns beautiful data

Users loved it. For about 30 seconds.

Then they'd ask a follow-up question:

User: "Now show me their orders from last week"

AI: "Orders from who? Which customers? What table?"

The AI had Alzheimer's. It forgot everything after each response.

And here's the kicker: this is exactly how LLMs are designed to work.

Every major LLM - GPT-4, Claude, Gemini - is stateless. They process your prompt, generate a response, and immediately forget everything. Like a goldfish with a computer science degree.

But users don't know this. They expect a conversation.

So we had two choices:

Educate users about LLM limitations (good luck with that)
Make the AI actually remember

We chose option 2. Here's what happened.

💥 Problem #1: The Goldfish Effect

Let me show you the exact moment users started leaving.

Real conversation from our logs:

9:23 AM - User: "Show me customers in New York"
9:23 AM - AI: [Returns 47 customers]
9:24 AM - User: "How many orders did they place?"
9:24 AM - AI: "Could you specify which customers?"
9:24 AM - User: "THE ONES YOU JUST SHOWED ME"
9:25 AM - User: *closes tab*

This happened 67 times per day.

Why? Because LLMs have zero-length memory. Every request is like talking to someone who just woke up from anesthesia.

# What actually happens inside an LLM
def process_query(prompt):
    response = llm.generate(prompt)  # Process
    memory.clear()  # FORGET EVERYTHING
    return response

The painful metrics:

67% of users abandoned after 2nd question
Average session: 1.2 queries (users gave up)
Support tickets: "Is this thing even AI?"

We needed memory. But here's where it gets interesting...

🧠 The Wrong Solution (That Everyone Tries First)

Our first attempt was embarrassingly naive:

"Let's just save everything!"

# DON'T DO THIS
conversation_history = []

def ask_ai(question):
    conversation_history.append(question)

    # Send ENTIRE history to LLM
    full_context = "\n".join(conversation_history)
    response = llm.generate(full_context)

    conversation_history.append(response)
    return response

Seemed logical. Worked for 3 questions.

Then it exploded.

What went wrong:

Query #10: 2,341 tokens → $0.02 per query
Query #20: 8,924 tokens → $0.09 per query
Query #50: 34,512 tokens → $0.35 per query

Plus, response times went from 800ms to 8 seconds. Users thought the app was broken.

The real problem: We were building an infinite memory system with a finite token budget.

We needed something smarter.

💡 The Breakthrough: The 10-Message Rule

After burning through $847 in API costs testing different approaches, we discovered something surprising.

Humans don't remember everything either.

Think about your own conversations. You remember:

The current topic
Last few exchanges
Maybe something important from earlier

You don't replay the entire conversation word-for-word in your head.

What if AI could do the same?

We ran an experiment. Tested different memory windows:

Messages Saved	Token Cost	User Satisfaction	Success Rate
5 messages	Low	71%	78%
10 messages	Optimal	94%	96%
20 messages	High	91%	94%
50 messages	Insane	89%	93%

The data was clear: 10 messages was the sweet spot.

Beyond 10? Diminishing returns. Below 10? Lost context.

🎯 The Architecture That Changed Everything

Here's the system we built (stolen from how humans actually remember):

┌─────────────────────────────────────────┐
│  User asks: "Show me their orders"      │
│  (Who are "they"? AI doesn't know!)     │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│    SESSION MANAGER                      │
│  "Oh, this is session ABC123"           │
│  "Let me check recent history..."       │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│   LOAD LAST 10 MESSAGES                 │
│                                         │
│  [7 messages ago]                       │
│  User: "Show customers in NY"           │
│  AI: "SELECT * FROM customers..."       │
│                                         │
│  [Current]                              │
│  User: "Show me their orders"           │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│   CONTEXT INJECTION (The Magic)         │
│                                         │
│  System: You're SQLatte, an AI...      │
│  Context: User asked about NY customers │
│  Current: Show me "their" orders        │
│  → "their" = NY customers from before!  │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│         LLM PROCESSES                   │
│  "Ah! They mean NY customers!"          │
│  *generates SQL with proper JOIN*       │
└────────────┬────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────┐
│    SAVE TO HISTORY                      │
│  Keep last 10, drop older ones          │
└─────────────────────────────────────────┘

The key insight: We're not building infinite memory. We're building selective memory.

💻 The Code: How We Actually Built It

Now the fun part. Let me show you exactly how we implemented this, including the mistakes we made.

Problem #2: "How Do We Track Sessions?"

First challenge: How do you track which conversation belongs to which user?

Naive approach (doesn't work):

# Tracking by IP address
sessions = {}

def get_session(ip_address):
    return sessions.get(ip_address)

Why it fails:

Multiple users behind same corporate proxy
User switches devices
IP changes during conversation

The fix: Session IDs + Browser Storage

import uuid
from datetime import datetime

class ConversationSession:
    """One user's conversation thread"""

    def __init__(self, session_id: str):
        self.session_id = session_id  # Unique ID
        self.messages = []
        self.created_at = datetime.now()
        self.last_activity = datetime.now()

    def add_message(self, role: str, content: str):
        """
        Add message to history
        role: 'user' or 'assistant'
        """
        self.messages.append({
            'role': role,
            'content': content,
            'timestamp': datetime.now()
        })
        self.last_activity = datetime.now()

    def get_recent_context(self, max_messages: int = 10):
        """
        Get last N messages for LLM context

        This is the magic number: 10
        - Enough for context
        - Not too expensive
        - Fast inference
        """
        return self.messages[-max_messages:]

Client-side (the secret sauce):

// Store session ID in browser
function getOrCreateSession() {
    let sessionId = localStorage.getItem('sqlatte_session_id');

    if (!sessionId) {
        // Generate new session
        sessionId = crypto.randomUUID();
        localStorage.setItem('sqlatte_session_id', sessionId);
    }

    return sessionId;
}

// Send with every request
fetch('/api/query', {
    body: JSON.stringify({
        question: userQuestion,
        session_id: getOrCreateSession()  // ← Key!
    })
});

Why this works:

Persists across page reloads
User returns tomorrow → same session
No login required
Privacy-friendly (stays in browser)

Problem #3: "How Do We Inject Context Without Confusing The AI?"

This is where most implementations fail. You can't just dump message history. The AI gets confused.

❌ Bad approach:

# This confuses the AI
context = "Previous messages: " + "\n".join(all_messages)
prompt = context + "\n\nCurrent question: " + question

✅ Good approach (structured context):

def build_llm_context(session_id: str, current_question: str):
    """
    Build structured context that AI can understand
    """
    session = get_session(session_id)
    recent = session.get_recent_context(max_messages=10)

    # Structure matters!
    context = []

    # 1. System prompt (who the AI is)
    context.append({
        'role': 'system',
        'content': 'You are SQLatte, an AI that converts natural language to SQL.'
    })

    # 2. Recent conversation (formatted clearly)
    if len(recent) > 1:
        context.append({
            'role': 'system',
            'content': f'''
Context: User is continuing a conversation.

Recent exchange:
{format_conversation_history(recent[-5:])}

Important: The current question may reference previous context.
'''
        })

    # 3. All recent messages (for AI to see flow)
    for msg in recent:
        context.append({
            'role': msg['role'],
            'content': msg['content']
        })

    # 4. Current question
    context.append({
        'role': 'user',
        'content': current_question
    })

    return context

def format_conversation_history(messages):
    """Make history readable for AI"""
    formatted = []
    for msg in messages:
        if msg['role'] == 'user':
            formatted.append(f"User: {msg['content']}")
        else:
            # Summarize AI responses (don't repeat everything)
            summary = msg['content'][:100] + "..."
            formatted.append(f"Assistant: {summary}")

    return "\n".join(formatted)

The difference this makes:

Before (unstructured):

API Error: "Context too ambiguous"
AI: "I don't understand the reference"

After (structured):

AI: "Ah, they're asking about the NYC customers from 3 messages ago"
SQL: SELECT COUNT(*) FROM orders o 
     JOIN customers c ON o.customer_id = c.id
     WHERE c.city = 'New York'

Problem #4: "This Is Getting Expensive!"

Remember our $847 API bill? Here's how we fixed it.

Token usage before optimization:

Message 1:  500 tokens  → $0.001
Message 5:  2,400 tokens → $0.005
Message 10: 5,800 tokens → $0.012
Message 20: 12,300 tokens → $0.025

Three optimizations that saved us:

1. Smart Compression

def compress_old_messages(messages):
    """
    Summarize messages older than 5 turns
    Instead of full text, keep just the essence
    """
    if len(messages) <= 5:
        return messages

    old_messages = messages[:-5]
    recent_messages = messages[-5:]

    # Summarize old ones
    summary = extract_key_points(old_messages)
    compressed = [{
        'role': 'system',
        'content': f'Earlier in conversation: {summary}'
    }]

    return compressed + recent_messages

def extract_key_points(messages):
    """Extract just the table names and operations"""
    tables = set()
    operations = []

    for msg in messages:
        tables.update(extract_tables_from_sql(msg['content']))
        operations.append(extract_operation(msg['content']))

    return f"discussed tables: {', '.join(tables)}, operations: {', '.join(operations)}"

2. Token Budgeting

MAX_CONTEXT_TOKENS = 2000  # Hard limit

def enforce_token_budget(messages):
    """
    If we're over budget, drop oldest messages
    Keep at least last 3 messages (critical context)
    """
    total = estimate_tokens(messages)

    while total > MAX_CONTEXT_TOKENS and len(messages) > 3:
        messages.pop(0)  # Remove oldest
        total = estimate_tokens(messages)

    return messages

3. Lazy Loading

def get_context(session_id, current_question):
    """
    Don't load full history unless needed
    """
    session = get_session(session_id)

    # Start with minimal context
    messages = session.messages[-3:]

    # Does this question need more context?
    if needs_deeper_context(current_question):
        messages = session.messages[-10:]

    return messages

def needs_deeper_context(question):
    """
    Check if question references earlier context
    """
    reference_words = ['that', 'those', 'them', 'their', 'previous', 'earlier']
    return any(word in question.lower() for word in reference_words)

Results:

Average tokens per query: 5,800 → 1,200 (79% reduction)
Monthly API costs: $847 → $178 (79% savings)
Response time: 8s → 1.2s (85% faster)

🎨 Prompt Engineering: The Secret Sauce

The way you inject context into prompts matters. A LOT.

❌ Bad Approach (Naive):

prompt = f"""
Previous messages: {all_messages}

Current question: {question}

Generate SQL.
"""

Problems:

Too verbose
No structure
LLM gets confused
Wastes tokens

✅ Good Approach (Structured):

prompt = f"""
Context: User is continuing a conversation about database queries.

Recent conversation:
User: "Show me top customers"
Assistant: Generated SQL for top 10 customers by revenue
User: "Show orders from last week" [CURRENT]

Task: Generate SQL for the current question, considering:
1. Previously discussed: top customers
2. Implied context: Orders from THE CUSTOMERS discussed above
3. Time range: Last week

Generate SQL:
"""

Why this works:

Clear structure
Explicit context markers
Task-oriented
Inference guidance

📊 Real-World Examples

Example 1: Follow-up Questions

Without Memory:

User: "Show me customers in New York"
AI: SELECT * FROM customers WHERE city = 'New York'

User: "How many orders did they make?"
AI: "Who are 'they'? Please specify."

With Memory:

User: "Show me customers in New York"
AI: SELECT * FROM customers WHERE city = 'New York'

User: "How many orders did they make?"
AI: SELECT COUNT(*) 
    FROM orders o
    JOIN customers c ON o.customer_id = c.id
    WHERE c.city = 'New York'

[AI remembers we're talking about NY customers!]

Example 2: Context-Aware Refinements

Without Memory:

User: "Show revenue by month"
AI: [Returns data]

User: "Now filter for products over $1000"
AI: "Filter what? Please provide complete query."

With Memory:

User: "Show revenue by month"
AI: SELECT 
      DATE_TRUNC('month', order_date) as month,
      SUM(amount) as revenue
    FROM orders
    GROUP BY month

User: "Now filter for products over $1000"
AI: SELECT 
      DATE_TRUNC('month', order_date) as month,
      SUM(amount) as revenue
    FROM orders
    WHERE product_price > 1000  -- Added filter!
    GROUP BY month

[AI understands the refinement applies to previous query!]

⚡ Performance Optimization: The Trade-offs

Building conversational AI isn't free. Here are the trade-offs:

Token Usage

Without Memory: ~500 tokens/request
With Memory (10 msgs): ~2500 tokens/request

Cost increase: 5x
Value increase: 10x+ (UX improvement)

Response Time

Without Context: ~800ms
With Context: ~1200ms (+50%)

Why? More tokens to process
Worth it? Absolutely.

Memory Management Strategy

class ConversationManager:
    def cleanup_expired_sessions(self):
        """
        Remove old sessions to prevent memory bloat

        Strategy:
        - Sessions expire after 60 minutes of inactivity
        - Periodic cleanup (every 10 minutes)
        - Keep favorites even if expired
        """
        cutoff = datetime.now() - timedelta(hours=1)

        for session_id in list(self.sessions.keys()):
            session = self.sessions[session_id]

            if session.is_expired(cutoff):
                # Keep if has favorites
                if not session.has_favorites():
                    del self.sessions[session_id]

🎯 Optimization Strategies We Use

1. Progressive Context Loading

Don't load all history. Start with last 5, extend if needed.

2. Semantic Compression

Summarize old messages instead of keeping full text:

def compress_old_messages(messages: List[Message]) -> str:
    """
    Compress messages older than 10 turns
    """
    if len(messages) > 10:
        old_messages = messages[:-10]
        summary = f"Earlier in conversation: discussed {extract_topics(old_messages)}"
        return summary
    return ""

3. Smart Token Budgeting

MAX_CONTEXT_TOKENS = 2000
MAX_HISTORY_MESSAGES = 10

def get_context(session_id):
    messages = get_messages(session_id)

    # Estimate tokens
    total_tokens = sum(estimate_tokens(m) for m in messages)

    if total_tokens > MAX_CONTEXT_TOKENS:
        # Drop oldest messages
        while total_tokens > MAX_CONTEXT_TOKENS:
            messages.pop(0)
            total_tokens = sum(estimate_tokens(m) for m in messages)

    return messages

4. Client-Side Session Persistence

Store session ID in browser localStorage:

function getOrCreateSession() {
    let sessionId = localStorage.getItem('sqlatte_session_id');

    if (!sessionId) {
        sessionId = generateUUID();
        localStorage.setItem('sqlatte_session_id', sessionId);
    }

    return sessionId;
}

Benefits:

Conversations persist across page reloads
User comes back later → continues same conversation
No server-side user accounts needed

🔒 Edge Cases We Handle

1. Session Timeout

def is_expired(self, timeout_minutes: int = 60) -> bool:
    """Check if session has expired"""
    elapsed = datetime.now() - self.last_activity
    return elapsed > timedelta(minutes=timeout_minutes)

Why 60 minutes?

Users take coffee breaks
Still relevant context
Not too memory-hungry

2. Context Confusion

Sometimes AI misinterprets context. We add disambiguation:

if user_question_is_ambiguous(question, context):
    prompt = f"""
    IMPORTANT: User's question might be ambiguous.

    Question: "{question}"
    Previous context: {context_summary}

    If ambiguous, ask for clarification rather than guessing.
    """

3. Table Switching

User changes topic → clear context:

def detect_topic_change(new_question, old_context):
    """
    Detect if user switched to different tables/topic
    """
    old_tables = extract_tables(old_context)
    new_tables = extract_tables(new_question)

    if not set(old_tables).intersection(new_tables):
        return True  # Topic changed!

    return False

📈 Results: Before vs After

User Satisfaction

Before: 67% completion rate (users gave up)
After: 94% completion rate
Why: Natural conversation flow

Query Complexity

Before: Average 1.2 queries per session
After: Average 4.8 queries per session
Why: Users explore data more when follow-ups work

Error Rate

Before: 23% queries failed (missing context)
After: 8% queries failed
Why: AI understands intent better

🎓 Key Lessons Learned

1. Don't Over-Context

10 messages is the sweet spot. More = diminishing returns.

2. Structure Your History

❌ Dump all messages
✅ Structured summary + recent details

3. Make Sessions Disposable

Don't try to keep everything forever. Cleanup expired sessions.

4. Prompt Engineering > Architecture

How you inject context matters MORE than how much context.

5. Test Follow-Up Scenarios

Write tests for:

"Show me X" → "Now show me Y"
"Filter by A" → "Remove that filter"
"Top 10" → "Show details for #3"

💡 Try It Yourself

Full implementation in SQLatte:
👉 https://github.com/osmanuygar/sqlatte

🎯 Conclusion

Building AI that remembers isn't about storing everything - it's about storing the right things at the right time.

Key takeaways:

✅ Use session-based architecture
✅ Limit context window (10 messages sweet spot)
✅ Structure your prompts carefully
✅ Handle edge cases (timeouts, topic changes)
✅ Monitor token usage

The result? AI that feels intelligent, not just smart.

What's your experience with conversational AI? Drop a comment below! 👇

Found this useful? Star the repo: https://github.com/osmanuygar/sqlatte

How Prompt Engineering Turned Natural Language into Production-Ready SQL Queries

osman uygar köse — Tue, 23 Dec 2025 14:54:16 +0000

TL;DR: We built SQLatte, an AI-powered natural language to SQL platform, and learned that 80% of the magic is in the prompts. Here's how we engineered LLM prompts to generate optimized, production-ready SQL queries with 94% accuracy and 10x faster query execution through intelligent partition filtering.

The Challenge

Imagine you're a system engineer at 2 AM, your monitoring system is screaming, and you need to query your Zabbix database to find which host is causing the issue. Problem? You don't know SQL. Even if you did, Zabbix's schema is complex with dozens of interconnected tables.

This is why we built SQLatte - a platform where you can ask "Which hosts had CPU over 90% in the last hour?" and get instant results without writing a single line of SQL.

The secret sauce? Carefully engineered prompts that turn casual questions into optimized database queries.

🏗️ Architecture Overview

User Question
     ↓
Intent Detection (SQL vs Chat)
     ↓
Schema Context Injection
     ↓
SQL Generation with Optimizations
     ↓
Query Execution
     ↓
Results + Explanation

Let's dive into each step and the prompt engineering behind it.

🧠 Part 1: Intent Detection - The Gatekeeper

The Problem

Not every question needs a SQL query. Users might say:

"Hello!" (chat)
"How do I use this?" (chat)
"Show me top customers" (SQL)
"What's my table schema?" (SQL)

We need to route correctly before attempting SQL generation.

The Solution: Structured Intent Detection Prompt

def determine_intent(self, question: str, schema_info: str) -> Dict[str, any]:
    """
    Determine if the question requires SQL or is general chat
    """
    prompt = f"""Analyze this user question and determine if it requires a SQL query or is general conversation.

Available Tables Schema:
{schema_info if schema_info else "No tables selected"}

User Question: {question}

Rules:
1. If question asks about data, analytics, or queries related to the available tables → intent: "sql"
2. If question is greeting, general chat, help, or unrelated to tables → intent: "chat"
3. If tables are not selected but question seems like SQL query → intent: "chat" (explain they need to select tables)

Respond in this exact format:
INTENT: sql or chat
CONFIDENCE: 0.0 to 1.0
REASONING: brief explanation
"""

Why This Works

1. Structured Output: We enforce a specific format (INTENT: ... CONFIDENCE: ...) which makes parsing deterministic.

2. Context Injection: We pass the available schema so the LLM knows what data exists.

3. Clear Rules: Three explicit rules guide the LLM's decision-making.

4. Confidence Score: Allows us to handle edge cases (confidence < 0.6 → ask for clarification).

Real Results

# Example 1: Clear SQL intent
question = "Show me users who signed up yesterday"
# → INTENT: sql, CONFIDENCE: 0.95

# Example 2: Chat intent
question = "Hello, how are you?"
# → INTENT: chat, CONFIDENCE: 0.98

# Example 3: Edge case
question = "What data do you have?"
# → INTENT: chat, CONFIDENCE: 0.85
# → Response: "Please select tables from the dropdown..."

Accuracy: 97% on our test dataset of 500 queries.

⚡ Part 2: The Performance Game-Changer - Partition Filtering

The Problem

Imagine querying a table with 10 billion rows of monitoring data:

-- ❌ DISASTER: Full table scan
SELECT * FROM metrics 
WHERE cpu_usage > 90
LIMIT 100;

-- Execution time: 347 seconds ⏰
-- Rows scanned: 10,000,000,000

Our Trino/Hive tables are partitioned by date using a dt column (format: YYYYMMDD, e.g., '20251223'). But the LLM doesn't know this by default.

The Solution: Partition-Aware Prompt Engineering

Here's the critical part of our SQL generation prompt:

def generate_sql(self, question: str, schema_info: str) -> Tuple[str, str]:
    prompt = f"""You are a SQL expert. Generate a SQL query based on the user's question.

Table Schema(s):
{schema_info}

User Question: {question}

Rules:
1. Generate ONLY valid SQL syntax
2. If multiple tables are provided, use appropriate JOINs
3. Infer JOIN conditions from table relationships (common column names)
4. Use table aliases for readability (e.g., orders o, customers c)
5. Include LIMIT clause for safety (default 100 rows)
6. For aggregations, use GROUP BY appropriately
7. Use explicit JOIN syntax (INNER JOIN, LEFT JOIN, etc.)

⚡ PERFORMANCE OPTIMIZATION (CRITICAL):
8. **PARTITION COLUMN**: If schema contains a 'dt' column (VARCHAR format YYYYMMDD, e.g., '20251218'), this is a PARTITION KEY
   - ALWAYS add WHERE clause with 'dt' filter when possible
   - Date filters MUST use dt column in format: dt = '20251218' or dt BETWEEN '20251201' AND '20251218'
   - For "recent", "latest", "today", "last" queries → use last 2 days: dt >= '20251222'
   - For "yesterday" → use dt = '20251222' (current date - 1)
   - For specific date range → convert to dt format (e.g., "last week" → dt >= '20251216')
   - ⚠️ NEVER query without dt filter unless explicitly asked for "all time" data

9. If 'datetime' column exists alongside 'dt', use 'dt' for filtering (faster) and 'datetime' for display

10. Example optimized query: 
    SELECT * FROM orders WHERE dt >= '20251201' AND status = 'completed' LIMIT 100

Format your response as:
SQL:

sql
[your SQL query here]


EXPLANATION:
[brief explanation including partition strategy]
"""

Real-World Impact

Before partition optimization:

-- User: "Show me yesterday's failed orders"

-- Generated SQL (naive):
SELECT * FROM orders 
WHERE status = 'failed'
AND datetime >= '2024-12-22 00:00:00'
LIMIT 100;

-- ❌ Performance:
-- Execution: 89 seconds
-- Rows scanned: 10B
-- Partitions scanned: 3,650 (all partitions!)

After prompt engineering:

-- Generated SQL (optimized):
SELECT * FROM orders 
WHERE dt = '20241222'
AND status = 'failed'
AND datetime >= '2024-12-22 00:00:00'
LIMIT 100;

-- ✅ Performance:
-- Execution: 0.8 seconds (111x faster!)
-- Rows scanned: 2.7M
-- Partitions scanned: 1 (only yesterday's partition!)

The Magic: Query Classification

The LLM learns to classify queries and apply appropriate patterns:

Query Type	Pattern	Partition Strategy
"latest N records"	`ORDER BY datetime DESC LIMIT N`	`dt >= today - 2 days`
"yesterday"	Date = yesterday	`dt = yesterday`
"last week"	Date range	`dt >= today - 7 days`
"specific date"	Date = X	`dt = 'YYYYMMDD'`
"all time" (explicit)	No date filter	No dt filter

🎨 Part 3: Schema Context Injection

The Problem

LLMs need to understand your database structure to generate correct SQL. But how do you provide schema information effectively?

Our Approach: Rich Schema Description

def get_table_schema(self, table_name: str) -> str:
    """Get table schema with enhanced information"""

    schema_info = f"""Table: {self.database}.{table_name}
{'=' * 60}

COLUMNS:
──────────────────────────────────────────────────────────
  • order_id  →  BIGINT  [NOT NULL]  [PRIMARY KEY]
  • customer_id  →  BIGINT  [NOT NULL]  [INDEX]
  • dt  →  VARCHAR(8)  [PARTITION KEY]  ⚡ Format: YYYYMMDD
  • datetime  →  TIMESTAMP  [NOT NULL]
  • status  →  VARCHAR(50)  [NOT NULL]
  • amount  →  DECIMAL(10,2)
  • region  →  VARCHAR(50)

PRIMARY KEY: order_id

FOREIGN KEYS:
──────────────────────────────────────────────────────────
  • customer_id → customers.customer_id

INDEXES:
──────────────────────────────────────────────────────────
  • idx_status (status)
  • idx_customer (customer_id)

⚡ OPTIMIZATION HINTS:
──────────────────────────────────────────────────────────
  • This table is PARTITIONED by 'dt' column
  • ALWAYS filter by 'dt' for fast queries
  • Current partitions: 20231201 to 20251223 (755 days)

ROWS: 10,234,567,890
"""
    return schema_info

Key Insights

1. Visual Hierarchy: Clear sections (COLUMNS, INDEXES, etc.) help the LLM parse structure.

2. Explicit Markers: [PARTITION KEY], [PRIMARY KEY] make important columns obvious.

3. Optimization Hints: We literally tell the LLM how to optimize queries.

4. Format Examples: Format: YYYYMMDD prevents the LLM from guessing date formats.

🔗 Part 4: Multi-Table JOINs

The Challenge

User asks: "Show me orders with customer names from last week"

This requires:

Joining orders and customers tables
Inferring the join condition (likely customer_id)
Applying partition filter
Proper aliasing

The Prompt Enhancement

# In the SQL generation prompt:

"""
JOIN RULES (CRITICAL):
1. If multiple tables are provided, use appropriate JOINs
2. Infer JOIN conditions from:
   - Common column names (e.g., customer_id, user_id)
   - Foreign key relationships (see FOREIGN KEYS section in schema)
   - Column name patterns (table_name_id usually joins to table_name.id)

3. Always use explicit JOIN syntax:
   ✅ INNER JOIN customers c ON o.customer_id = c.customer_id
   ❌ FROM orders o, customers c WHERE o.customer_id = c.customer_id

4. Use meaningful table aliases:
   - orders → o
   - customers → c
   - products → p

5. Apply partition filter to ALL partitioned tables in JOIN

Example multi-table query:
SELECT o.order_id, c.customer_name, o.amount
FROM orders o
INNER JOIN customers c ON o.customer_id = c.customer_id
WHERE o.dt >= '20251216'  -- Last week
AND o.status = 'completed'
LIMIT 100;
"""

Results

Input:

"Show me orders with customer names from yesterday"

Generated SQL:

SELECT 
    o.order_id,
    o.datetime,
    c.customer_name,
    c.email,
    o.amount,
    o.status
FROM orders o
INNER JOIN customers c ON o.customer_id = c.customer_id
WHERE o.dt = '20251222'  -- ⚡ Partition filter!
ORDER BY o.datetime DESC
LIMIT 100;

Performance:

Join type: Correct (INNER JOIN)
Join condition: Inferred correctly
Partition filter: Applied ✅
Execution time: 1.2 seconds

Success rate: 91% for 2-table joins, 84% for 3+ table joins

🎭 Part 5: Conversational Context

The Problem

Users ask follow-up questions:

User: "Show me sales by region"
Assistant: [shows results]

User: "What about last month?"  ← How does LLM know this refers to sales?

The Solution: Context Injection

def generate_sql_with_context(
    self,
    question: str, 
    schema_info: str,
    conversation_history: List[Dict]
) -> Tuple[str, str]:
    """Generate SQL with conversation context"""

    # Build context summary
    if len(conversation_history) > 1:
        context_summary = "\n\nRecent conversation:\n"
        for msg in conversation_history[-5:]:  # Last 5 messages
            if msg['role'] == 'user':
                context_summary += f"User: {msg['content']}\n"
            elif msg['role'] == 'assistant':
                # Include previous SQL for reference
                if 'sql' in msg.get('metadata', {}):
                    context_summary += f"Generated SQL: {msg['metadata']['sql']}\n"

        enhanced_question = f"{question}\n\nContext: {context_summary}"
    else:
        enhanced_question = question

    return self.generate_sql(enhanced_question, schema_info)

Example Flow

# Turn 1
User: "Show sales by region"
Generated SQL:
"""
SELECT region, SUM(amount) as total_sales
FROM orders
WHERE dt >= '20251201'  -- Current month
GROUP BY region
ORDER BY total_sales DESC;
"""

# Turn 2 (follow-up)
User: "What about last month?"

Context injected to LLM:
"""
Recent conversation:
User: Show sales by region
Generated SQL: SELECT region, SUM(amount) ... WHERE dt >= '20251201' ...

Current question: What about last month?
"""

Generated SQL (context-aware):
"""
SELECT region, SUM(amount) as total_sales
FROM orders
WHERE dt >= '20251101' AND dt < '20251201'  -- ⚡ Adjusted date range!
GROUP BY region
ORDER BY total_sales DESC;
"""

Context-aware accuracy: 88% (vs 23% without context)

🔧 Part 6: Prompt Engineering Best Practices

What We Learned

1. Be Explicit About Formats

❌ Bad:

"Use the date column for filtering"

✅ Good:

"Use the 'dt' column (VARCHAR format YYYYMMDD, e.g., '20251223') for partition filtering"

2. Provide Examples

❌ Bad:

"Filter by date range"

✅ Good:

"Example: dt BETWEEN '20251201' AND '20251223'"

3. Use Structured Outputs

❌ Bad:

"Determine if this needs SQL"

✅ Good:

"Respond in this exact format:
INTENT: sql or chat
CONFIDENCE: 0.0 to 1.0"

4. Prioritize Critical Information

We use markers like:

⚡ CRITICAL:
⚠️ NEVER:
✅ ALWAYS:

This helps the LLM focus on important rules.

How I Automated 90% of Data Requests Using LLM-Powered SQL Generation

osman uygar köse — Tue, 16 Dec 2025 09:17:49 +0000

How I Automated 90% of Data Requests Using LLM-Powered SQL Generation

https://github.com/osmanuygar/sqlatte

The Problem: You're the SQL Bottleneck

Picture this: You're a data engineer at a mid-sized company. Every day, you get Slack messages like:

"Hey, can you pull last month's revenue by region?"
"Quick question - what's our customer churn rate?"
"Need the top 10 products by sales ASAP for the board meeting"

You spend 30% of your time writing SQL queries for people who can clearly articulate what they need, but don't know how to get it.

The real problem isn't writing SQL - it's being the single point of failure for data access.

What If Non-Technical Users Could Query Databases Directly?

That was my goal: Build a system where anyone can ask questions in natural language and get SQL-generated results - without compromising security or data integrity.

Spoiler: LLMs make this surprisingly straightforward. Here's how I built it.

The Architecture: Solving Three Hard Problems

Building a text-to-SQL system isn't just about calling an LLM API. You need to solve:

Intent Detection - Is this a SQL query or just chat?
Schema Context - How do you give the LLM table information?
Multi-Table Relationships - How do you handle JOINs?

Problem #1: Intent Detection

The Challenge: Not every question needs SQL.

"Hello!" → Don't generate SQL
"Show me users" → Generate SQL
"How do I use this?" → Don't generate SQL

The Solution: Two-stage LLM approach.

def determine_intent(self, question: str, schema_info: str) -> Dict:
    """
    First LLM call: Figure out if we need SQL or not
    """
    prompt = f"""
    Analyze this question:
    Question: {question}
    Available tables: {schema_info}

    Rules:
    1. If asking about DATA in tables → "sql"
    2. If greeting/help/chat → "chat"
    3. If SQL-like but no tables selected → "chat"

    Respond:
    INTENT: sql or chat
    CONFIDENCE: 0.0 to 1.0
    REASONING: why
    """

    response = llm.generate(prompt)

    # Parse structured response
    intent = parse_intent(response)

    # Only generate SQL if confidence > 0.6
    return intent

Why this works:

Prevents hallucinated SQL for non-data questions
Saves API costs (no unnecessary SQL generation)
Better UX (conversational responses for chat)

Problem #2: Schema Context Window

The Challenge: How do you tell the LLM what tables exist?

Large schemas don't fit in context windows. And you can't just dump everything - the LLM gets confused.

The Solution: User-selected table approach.

# User selects tables from dropdown
selected_tables = ["orders", "customers"]

# Fetch only relevant schemas
schema_info = ""
for table in selected_tables:
    schema = db.get_table_schema(table)
    schema_info += f"\nTable: {table}\n{schema}\n"

# Now LLM has focused context
sql = llm.generate_sql(question, schema_info)

Result:

Context stays manageable
LLM focuses on relevant tables
Users think about their data model

Problem #3: Automatic JOINs

The Challenge: Users don't specify JOIN conditions.

Question: "Show me orders with customer names"
Tables: orders, customers

User expects: orders.customer_id = customers.id

The Solution: Let the LLM infer relationships.

prompt = f"""
You are a SQL expert. Generate a query.

Tables:
{schema_info}  # Contains column names

Question: {question}

Rules:
1. Infer JOIN conditions from common column names
2. Use table aliases (orders o, customers c)
3. Use explicit JOIN syntax (INNER JOIN, LEFT JOIN)
4. Add LIMIT 100 for safety

Format:
SQL:

sql
[query]


EXPLANATION:
[your reasoning]
"""

Example Output:

SELECT 
    o.order_id,
    o.total_amount,
    c.customer_name
FROM orders o
INNER JOIN customers c 
    ON o.customer_id = c.id
LIMIT 100

The LLM sees customer_id in both tables and correctly infers the JOIN condition.

The Implementation: Python + FastAPI + LLMs

Here's the actual architecture:

┌──────────────┐
│   FastAPI    │  ← Web server
└──────┬───────┘
       │
       ├─→ LLM Provider (Factory Pattern)
       │   ├─ Anthropic Claude
       │   ├─ Google Gemini
       │   └─ Vertex AI
       │
       └─→ DB Provider (Factory Pattern)
           ├─ Trino
           ├─ PostgreSQL
           └─ MySQL

Factory Pattern for Flexibility

Key Design Decision: Make it easy to swap LLMs and databases.

class ProviderFactory:
    LLM_PROVIDERS = {
        'anthropic': 'AnthropicProvider',
        'gemini': 'GeminiProvider',
        'vertexai': 'VertexAIProvider',
    }

    @staticmethod
    def create_llm_provider(config: dict):
        provider_name = config['llm']['provider']
        provider_class = LLM_PROVIDERS[provider_name]

        # Dynamic import
        module = __import__(f'providers.llm.{provider_name}')
        return module.Provider(config['llm'][provider_name])

To switch LLMs: Change one line in config.yaml:

llm:
  provider: "anthropic"  # or "gemini" or "vertexai"

No code changes needed. ✅

Abstract Base Classes

Each provider implements the same interface:

class LLMProvider(ABC):
    @abstractmethod
    def determine_intent(self, question: str, schema: str) -> Dict:
        pass

    @abstractmethod
    def generate_sql(self, question: str, schema: str) -> Tuple[str, str]:
        pass

    @abstractmethod
    def generate_chat_response(self, question: str) -> str:
        pass

Why this matters:

Test different LLMs without refactoring
Compare performance across providers
Future-proof against API changes

The API Flow: Step by Step

Here's what happens when a user asks: "Show me top 10 customers"

@app.post("/query")
async def process_query(request: QueryRequest):
    # request.question = "Show me top 10 customers"
    # request.table_schema = "Table: customers\nColumns: id, name, revenue..."

    # Step 1: Intent detection (First LLM call)
    intent = llm_provider.determine_intent(
        question=request.question,
        schema_info=request.table_schema
    )
    # → {"intent": "sql", "confidence": 0.95}

    # Step 2: Route based on intent
    if intent["intent"] == "sql" and intent["confidence"] > 0.6:

        # Step 3: Generate SQL (Second LLM call)
        sql, explanation = llm_provider.generate_sql(
            question=request.question,
            schema_info=request.table_schema
        )
        # → sql = "SELECT name, revenue FROM customers ORDER BY revenue DESC LIMIT 10"

        # Step 4: Execute on database
        columns, data = db_provider.execute_query(sql)

        # Step 5: Return structured response
        return SQLQueryResponse(
            sql=sql,
            columns=columns,
            data=data,
            explanation=explanation
        )

    else:
        # Chat mode
        message = llm_provider.generate_chat_response(request.question)
        return ChatResponse(message=message)

Performance:

Intent detection: ~200ms
SQL generation: ~500ms
Query execution: ~100-1000ms (depends on query)
Total: ~800-1700ms

Not instant, but acceptable for analytical queries.

Configuration: One YAML File

Design Goal: No .env files, no complex setup.

# config/config.yaml

# LLM Provider
llm:
  provider: "anthropic"
  anthropic:
    api_key: "sk-ant-your-key"
    model: "claude-sonnet-4-20250514"
    max_tokens: 1000

# Database Provider
database:
  provider: "trino"
  trino:
    host: "trino.company.com"
    port: 443
    user: "readonly_user"
    password: "***"
    catalog: "hive"
    schema: "default"

Supports environment variables too:

llm:
  anthropic:
    api_key: "${ANTHROPIC_API_KEY}"  # Read from env

Run:

pip install -r requirements.txt
python run.py
# → Server starts on http://localhost:8000

That's it. No Docker required (but supported).

The Widget: Embeddable Anywhere

Problem: How do you deploy this to non-technical users?

Solution: JavaScript widget that works on any website.

<!-- Add to any webpage -->
<script src="http://your-sqlatte:8000/static/js/sqlatte-badge.js"></script>

<script>
  // Optional config
  window.SQLatteWidget.configure({
    fullscreen: true,
    position: 'bottom-right'
  });
</script>

Result: A floating badge appears:

                                    ☕
                                   ┌─┐
                                   └─┘

Click it → Chat modal opens → Query your database.

Widget Architecture

Self-contained JavaScript module:

(function() {
  const SQLatteWidget = {
    // Auto-detect backend URL
    apiBase: (function() {
      if (window.location.hostname === 'localhost') {
        return 'http://localhost:8000';
      }
      return window.location.origin;
    })(),

    // Create UI
    init() {
      this.createBadge();      // Floating button
      this.createModal();      // Chat interface
      this.injectStyles();     // External CSS
    },

    // Query backend
    async sendMessage(question) {
      const response = await fetch(`${this.apiBase}/query`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          question: question,
          table_schema: this.currentSchema
        })
      });

      const result = await response.json();

      if (result.response_type === 'sql') {
        this.renderTable(result.columns, result.data);
      } else {
        this.renderChat(result.message);
      }
    }
  };

  // Initialize on page load
  document.addEventListener('DOMContentLoaded', () => {
    SQLatteWidget.init();
  });

  // Expose globally
  window.SQLatteWidget = SQLatteWidget;
})();

Key Design Choices:

External CSS - No inline styles (easier customization)
Namespaced classes - sqlatte-* (avoid conflicts)
CORS configuration - Works cross-domain
Z-index management - Modal always on top

Security Considerations

Rule #1: Never give write access to the AI.

database:
  trino:
    user: "readonly_user"  # ← Read-only DB user

Rule #2: Limit query results.

# In LLM prompt
"""
Rules:
...
5. Include LIMIT clause for safety (default 100 rows)
"""

Rule #3: Validate SQL before execution.

def execute_query(self, sql: str):
    # Basic SQL injection prevention
    dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'TRUNCATE']

    sql_upper = sql.upper()
    for keyword in dangerous_keywords:
        if keyword in sql_upper:
            raise SecurityError(f"Forbidden keyword: {keyword}")

    # Execute
    return self.db.execute(sql)

Rule #4: Rate limiting.

# In FastAPI
from slowapi import Limiter

limiter = Limiter(key_func=get_remote_address)

@app.post("/query")
@limiter.limit("10/minute")  # 10 queries per minute
async def process_query(request: QueryRequest):
    ...

Real-World Results

Before:

20-30 SQL requests per day
30 min average response time
Data team spending 25% time on ad-hoc queries

After (2 months):

<5 SQL requests per day (90% reduction)
Self-service analytics
Data team focuses on pipeline development

ROI: ~15 hours/week saved for 3-person data team.

Challenges & Solutions

Challenge 1: LLM Hallucinations

Problem: Sometimes LLM generates invalid SQL or references non-existent columns.

Solution:

Always show generated SQL to user
Catch database errors gracefully
Let users see the error and rephrase

try:
    columns, data = db.execute_query(sql)
except DatabaseError as e:
    return ErrorResponse(
        message="The generated SQL failed. Try rephrasing your question.",
        sql=sql,  # Show what was attempted
        error=str(e)
    )

Challenge 2: Complex JOINs

Problem: LLM struggles with 3+ table JOINs.

Solution: Encourage users to select only relevant tables.

UI: "Select tables (Ctrl+Click for multiple)"
Tip: "Select 2-3 related tables for best results"

Challenge 3: Cost Control

Problem: LLM API costs can add up.

Solution:

Intent detection with smaller models
Cache common queries
Rate limiting per user

# Use cheaper model for intent
class AnthropicProvider:
    def determine_intent(self, ...):
        # Use Haiku (cheaper) for simple task
        model = "claude-haiku-3-20240307"

    def generate_sql(self, ...):
        # Use Sonnet (better) for complex task
        model = "claude-sonnet-4-20250514"

How to Run It Yourself

1. Clone:

git clone https://github.com/osmanuygar/sqlatte.git
cd sqlatte

2. Install:

pip install -r requirements.txt

3. Configure config/config.yaml:

llm:
  provider: "anthropic"
  anthropic:
    api_key: "your-key-here"

database:
  provider: "trino"  # or postgresql, mysql, etc.
  trino:
    host: "your-db-host"
    # ... connection details

4. Run:

python run.py
# Visit http://localhost:8000

5. (Optional) Docker:

docker-compose up -d

What's Next?

Current roadmap:

[ ] Query history (let users save/reuse queries)
[ ] Data visualization (charts from results)
[ ] Support for ClickHouse, BigQuery
[ ] Query cost estimation
[ ] Multi-language support

Key Takeaways

What worked:

Two-stage LLM approach (intent → SQL) prevents hallucinations
Factory pattern makes swapping providers trivial
User-selected schemas keeps context manageable
Widget architecture enables easy deployment

What I'd do differently:

Add query result caching earlier
Implement user-based rate limiting from day one
Build query history sooner (users want to revisit queries)

Try It / Contribute

GitHub: https://github.com/osmanuygar/sqlatte
Issues: Feature requests and bugs welcome
PRs: Contributions appreciated (especially new DB/LLM providers)

What would you build with text-to-SQL? Let me know in the comments!

Building data tools? Follow for more architecture deep-dives.

Kubernetes Secrets Without the Pain: Meet kcpwd

osman uygar köse — Mon, 01 Dec 2025 18:39:47 +0000

Kubernetes Secrets Without the Pain: Meet kcpwd

kcpwd is a cross-platform password manager that syncs to Kubernetes secrets with zero infrastructure. No Vault servers, no operators, no complexity—just kcpwd k8s sync.

The Problem: Secret Management is Too Complicated

You know the drill. You're deploying to Kubernetes and need to manage secrets. Your options?

Option 1: Hardcode them ❌

# deployment.yaml (please don't do this)
env:
  - name: DB_PASSWORD
    value: "super_secret_123"  # 😱 In git!

Option 2: Setup Vault ⚠️

Install Consul or etcd
Deploy 3+ Vault servers
Configure unsealing
Setup RBAC
Write policies
Time spent: 2-4 hours minimum

Option 3: Use External Secrets Operator ⚠️

Deploy operator
Configure cloud provider
Manage IAM roles
Time spent: 1-2 hours

There has to be a better way for the 90% of teams who just need secure secret management without enterprise complexity.

Enter kcpwd: The Simple Solution

kcpwd is a cross-platform password manager that works as both a CLI and Python library—with native Kubernetes integration. It's designed for developers who want security without complexity.

Installation (30 seconds)

pip install kcpwd

That's it. No servers, no operators, no configuration files.

Basic Usage (5 minutes to production)

Step 1: Store your secrets locally

# Store passwords (uses your OS keyring)
kcpwd set prod_db_password "my_secure_password"
kcpwd set api_key "sk-1234567890"
kcpwd set redis_password "redis_secret"

Step 2: Sync to Kubernetes

# Sync single secret
kcpwd k8s sync prod_db_password --namespace production

# Or sync all at once
kcpwd k8s sync-all --namespace production

Step 3: Use in your deployments

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
  namespace: production
spec:
  template:
    spec:
      containers:
      - name: app
        image: myapp:latest
        env:
        - name: DB_PASSWORD
          valueFrom:
            secretKeyRef:
              name: prod-db-password  # Created by kcpwd
              key: password
        - name: API_KEY
          valueFrom:
            secretKeyRef:
              name: api-key
              key: password

That's it. You're now managing Kubernetes secrets securely.

Why kcpwd?

✅ Zero Infrastructure

No servers to maintain. No databases to backup. kcpwd uses your native OS keyring:

macOS: Keychain
Linux: Secret Service (gnome-keyring, KWallet)
Windows: Credential Locker
CI/CD: Encrypted file storage (AES-256-GCM)

✅ Works Everywhere

# On your Mac
kcpwd set db_password "secret123"

# In GitHub Actions
pip install kcpwd
kcpwd set db_password "${{ secrets.DB_PASSWORD }}"
kcpwd k8s sync-all --namespace production

# In Docker/CI/CD
# No keyring? No problem! Auto-fallback to encrypted storage

✅ Developer Experience

Python Integration:

from kcpwd import require_password

@require_password('database_password')
def connect_to_db(password=None):
    return psycopg2.connect(
        host="postgres.example.com",
        password=password
    )

# Password auto-injected from keyring
connect_to_db()

CLI Power:

# Generate strong passwords
kcpwd generate -l 32 -s api_key

# List all secrets
kcpwd list

# Check what's in Kubernetes
kcpwd k8s list --namespace production

Real-World Use Cases

1. CI/CD Pipeline (GitHub Actions)

name: Deploy to Kubernetes
on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Setup and sync secrets
        run: |
          pip install kcpwd

          # Import from GitHub Secrets
          kcpwd set db_password "${{ secrets.DB_PASSWORD }}"
          kcpwd set api_key "${{ secrets.API_KEY }}"

          # Sync to Kubernetes
          kcpwd k8s sync-all --namespace production

      - name: Deploy
        run: kubectl apply -f k8s/

Time to setup: ~5 minutes

2. Multi-Environment Management

#!/bin/bash
# deploy.sh

ENVIRONMENT=$1  # dev, staging, production

# Sync environment-specific secrets
kcpwd k8s sync-all \
  --namespace $ENVIRONMENT \
  --prefix ${ENVIRONMENT}_ \
  --label environment=$ENVIRONMENT \
  --label managed-by=kcpwd

echo "✓ Secrets synced to $ENVIRONMENT"

# Deploy to staging
./deploy.sh staging

# Deploy to production
./deploy.sh production

3. Helm Integration

kcpwd works seamlessly with Helm charts:

# values.yaml
database:
  host: postgres.example.com
  password: "{{ kcpwd('db_password') }}"

api:
  endpoint: https://api.example.com
  key: "{{ kcpwd('api_key') }}"

# Master-protected for production secrets
production:
  secret: "{{ kcpwd('prod_secret', master=true) }}"

# Process and deploy
kcpwd helm template values.yaml -o values-processed.yaml
helm install myapp ./chart -f values-processed.yaml

4. GitOps with ArgoCD

# argocd-app.yaml
apiVersion: argoproj.io/v1alpha1
kind: Application
metadata:
  name: myapp
  namespace: argocd
spec:
  project: default
  source:
    repoURL: https://github.com/myorg/myapp
    targetRevision: HEAD
    path: k8s
  destination:
    namespace: production
  hooks:
  - name: sync-secrets
    hookType: PreSync
    script: |
      pip install kcpwd
      kcpwd k8s sync-all --namespace production

No secrets in Git. kcpwd syncs them before ArgoCD deploys.

5. Watch Mode (Auto-Sync)

# Keep Kubernetes secrets in sync automatically
kcpwd k8s watch \
  --namespace production \
  --interval 60 \
  --prefix prod_

# Updates K8s secrets every 60 seconds
# Perfect for long-running deployments

Advanced Features

🔐 Master Password Protection

For production secrets, add an extra layer:

# Store with master password
kcpwd set-master prod_db_password "super_secret"
# Enter master password: ********

# Sync to K8s (will prompt for master password)
kcpwd k8s sync prod_db_password --namespace production

🔄 Bi-directional Sync

Import existing Kubernetes secrets into kcpwd:

# Import existing secret
kcpwd k8s import database-credentials \
  --namespace production \
  --key imported_db_password

# Now it's in kcpwd too!
kcpwd get imported_db_password

🏷️ Label Management

Organize secrets with Kubernetes labels:

kcpwd k8s sync api_key \
  --namespace production \
  --label app=myapp \
  --label team=backend \
  --label managed-by=kcpwd

🔁 Secret Rotation

Automate password rotation:

# rotate_secrets.py
from kcpwd import set_password, generate_password
from kcpwd.k8s import sync_to_k8s
import subprocess

def rotate_secret(key, namespace):
    # Generate new password
    new_password = generate_password(length=32)

    # Update in kcpwd
    set_password(key, new_password)

    # Sync to K8s
    sync_to_k8s(key, namespace=namespace)

    # Restart deployment
    subprocess.run([
        "kubectl", "rollout", "restart",
        f"deployment/{key}-app", "-n", namespace
    ])

# Use with cron or K8s CronJob
rotate_secret("api_key", "production")

Comparison: kcpwd vs. Alternatives

Feature	kcpwd	Vault	Sealed Secrets	External Secrets
Setup Time	0 minutes	2-4 hours	30+ minutes	1+ hour
Infrastructure	None	3+ servers	Controller	Operator + Cloud
Learning Curve	Minimal	Steep	Moderate	Moderate
Local Dev	✅ Perfect	⚠️ Complex	❌ No	⚠️ Limited
CI/CD	✅ Simple	✅ Yes	✅ Yes	✅ Yes
GitOps	✅ Yes	✅ Yes	✅ Yes	✅ Yes
Master Password	✅ Yes	❌ No	❌ No	❌ No
Python Library	✅ Yes	⚠️ Complex	❌ No	❌ No
Cost	Free	Free/$$$	Free	Free

Security Features

✅ AES-256-GCM encryption (NIST-approved)
✅ PBKDF2-SHA256 key derivation (600,000 iterations)
✅ Native OS keyrings (hardware-backed on supported devices)
✅ Optional master password (extra protection layer)
✅ No network communication (except kubectl)
✅ Kubernetes RBAC integration
✅ Open source (MIT license)

Platform Support

macOS ✅

Native Keychain integration, full feature support

Linux ✅

Auto-detects system keyring, fallback to encrypted storage

Windows ✅

Native Credential Locker integration

CI/CD ✅

Encrypted file storage, zero dependencies

Docker/Containers ✅

Works in any environment

Getting Started

Installation

# Basic installation
pip install kcpwd

# With Web UI (bonus feature!)
pip install 'kcpwd[ui]'

Quick Start

# 1. Check platform support
kcpwd info

# 2. Store a secret
kcpwd set my_password "secret123"

# 3. Sync to Kubernetes
kcpwd k8s sync my_password --namespace default

# 4. Verify
kubectl get secret my-password -o yaml

Documentation

Web UI Bonus 🎨

kcpwd also includes a beautiful web interface:

kcpwd ui
# Open http://localhost:8765

Features:

Modern dark theme
Password generator
Strength checker
Temporary password sharing
Session management

What's Next?

We're actively developing kcpwd and would love your feedback! Upcoming features:

[ ] Password history tracking
[ ] Terraform provider
[ ] Browser extensions
[ ] 2FA/OTP support
[ ] Kubernetes operator
[ ] Cloud sync options

Try It Today

# Install
pip install kcpwd

# Your first secret in K8s
kcpwd set test_secret "hello_kubernetes"
kcpwd k8s sync test_secret --namespace default

# That's it!
kubectl get secret test-secret

Star us on GitHub ⭐ if you find kcpwd useful!

Why I Built kcpwd

As a data engineer, I was tired of:

Fighting with Vault setup in dev environments
Committing .env files by accident
Managing secrets across local, CI/CD, and K8s
Complex workflows that slow down development

kcpwd is my answer: developer-first secret management that just works.

Community

🐛 Found a bug? Open an issue
💡 Feature request? Start a discussion
🤝 Want to contribute? Check our contributing guide
⭐ Star the repo to show support!

kcpwd - Because secret management shouldn't require a PhD in DevOps.

pip install kcpwd

Made with ❤️ by osmanuygar

kcpwd: A Minimal Python CLI to Keep Your Secrets Safe in Local Development

osman uygar köse — Tue, 11 Nov 2025 10:21:14 +0000

kcpwd

Hi everyone

In local development, it’s common to store credentials in .env files or even hardcode them temporarily during testing 😬.

While tools like Vault handle secrets perfectly in production, setting them up for small local projects can sometimes feel like overkill.

That’s why I built kcpwd — a tiny, native Python CLI and decorator-based tool to handle secrets without needing Vault.

https://github.com/osmanuygar/kcpwd

Why I built it

During local development, I often found myself doing one of the following:

Storing secrets in .env files (and hoping they don’t leak 😅)
Hardcoding temporary credentials into test code
Avoiding Vault setup because it felt too heavy for a small project

So, I wanted something lightweight that would let me store and access secrets securely, using just Python and a simple CLI.

That’s how kcpwd was born.

No need to expose secrets in your code — kcpwd will fetch them securely at runtime.

Why not just use Vault?

Vault is amazing, but for local-only or experimental projects, it can be too complex to set up.
kcpwd is designed as a mini Vault alternative for local use, offering:

No dependencies beyond Python
No separate service or container to run
Simple, fast, and native credential handling

Features

Python-native
Decorator + CLI interface
No external dependencies
Works locally or inside containers
Lightweight and developer-friendly
Secure storage using macOS Keychain
Python library for programmatic access
Native macOS integration

Example use cases

Local development environments
Internal scripts or prototypes
Testing Dockerized apps without exposing secrets
CI/CD debugging or credential injection during testing

Installation

From PyPI

pip install kcpwd

From Source

git clone https://github.com/osmanuygar/kcpwd.git
cd kcpwd
pip install -e .

Usage

CLI Usage

Store a password

kcpwd set dbadmin asd123

Retrieve a password (copies to clipboard)

kcpwd get dbadmin

Delete a password

kcpwd delete dbadmin

List all stored passwords

kcpwd list

Note: The list command uses security dump-keychain which may take a few seconds and requires access to your keychain. If it shows "No passwords stored" but you know you have passwords, try using Keychain Access app to verify.

Generate a secure password

# Generate a 16-character password (default)
kcpwd generate

# Generate a 20-character password
kcpwd generate -l 20

# Generate without symbols (alphanumeric only)
kcpwd generate --no-symbols

# Generate and save immediately
kcpwd generate -s myapi

# Generate a 6-digit PIN
kcpwd generate -l 6 --no-uppercase --no-lowercase --no-symbols

# Generate without ambiguous characters (no 0/O, 1/l/I)
kcpwd generate --exclude-ambiguous

Export passwords (NEW!)

# Export all passwords to a JSON file
kcpwd export backup.json

# Export only key names (without passwords)
kcpwd export keys.json --keys-only

# Force overwrite existing file
kcpwd export backup.json -f

** Security Warning*: Exported files contain passwords in **PLAIN TEXT*. Keep them secure!

Import passwords (NEW!)

# Import passwords (skip existing keys)
kcpwd import backup.json

# Import and overwrite existing passwords
kcpwd import backup.json --overwrite

# Preview what would be imported without making changes
kcpwd import backup.json --dry-run

Library Usage

Basic Functions

from kcpwd import set_password, get_password, delete_password

# Store a password
set_password("my_database", "secret123")

# Retrieve a password
password = get_password("my_database")
print(password)  # Output: secret123

# Retrieve and copy to clipboard
password = get_password("my_database", copy_to_clip=True)

# Delete a password
delete_password("my_database")

Password Generation

from kcpwd import generate_password

# Generate a secure password
password = generate_password(length=20)
print(password)  # Output: 'aB3#xK9!mL2$nP5@qR7&'

# Generate alphanumeric password (no symbols)
password = generate_password(length=16, use_symbols=False)
print(password)  # Output: 'aB3xK9mL2nP5qR7t'

# Generate a 6-digit PIN
pin = generate_password(
    length=6, 
    use_uppercase=False, 
    use_lowercase=False, 
    use_symbols=False
)
print(pin)  # Output: '384729'

List All Keys (NEW!)

from kcpwd import list_all_keys

# Get all stored password keys
keys = list_all_keys()
print(keys)  # Output: ['my_database', 'api_key', 'email_password']

# Check if a specific key exists
if 'my_database' in list_all_keys():
    print("Database password exists!")

Export/Import (NEW!)

from kcpwd import export_passwords, import_passwords

# Export all passwords
result = export_passwords('backup.json')
print(f"Exported {result['exported_count']} passwords")

# Export only keys (without passwords)
result = export_passwords('keys_only.json', include_passwords=False)

# Import passwords (skip existing)
result = import_passwords('backup.json')
print(f"Imported {result['imported_count']} passwords")
print(f"Skipped {len(result['skipped_keys'])} existing keys")

# Import with overwrite
result = import_passwords('backup.json', overwrite=True)

# Dry run to preview import
result = import_passwords('backup.json', dry_run=True)
print(result['message'])

Using Decorators (Recommended!)

The @require_password decorator automatically injects passwords from keychain:

from kcpwd import require_password, set_password

# First, store your password
set_password("my_db", "secret123")

# Use the decorator to auto-inject password
@require_password('my_db')
def connect_to_database(host, username, password=None):
    print(f"Connecting to {host} as {username}")
    print(f"Password: {password}")
    # Your database connection code here
    return f"Connected with password: {password}"

# Call without password - it's automatically retrieved!
result = connect_to_database("localhost", "admin")
# Output: Connected with password: secret123

Advanced Decorator Usage

You can specify different parameter names:

from kcpwd import require_password, set_password

# Store API key
set_password("github_api", "ghp_xxxxxxxxxxxx")

# Inject into custom parameter name
@require_password('github_api', param_name='api_key')
def call_github_api(endpoint, api_key=None):
    print(f"Calling GitHub API: {endpoint}")
    print(f"Using key: {api_key}")
    # Your API call code here
    return {"status": "success"}

# API key automatically retrieved from keychain
response = call_github_api("/user/repos")

Real-World Examples

Database Connection:

#import psycopg2
from kcpwd import require_password, set_password

# Setup: Store password once
set_password("prod_db", "my_secure_password")

# Use in your code
@require_password('prod_db')
def get_db_connection(host, user, database, password=None):
    return psycopg2.connect(
        host=host,
        user=user,
        password=password,
        database=database
    )

# No need to handle password manually!
conn = get_db_connection(
    host="prod.example.com",
    user="dbuser",
    database="myapp"
)

Backup Script:

from kcpwd import export_passwords, import_passwords
import os
from datetime import datetime

def backup_passwords():
    """Create a timestamped backup of all passwords"""
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_file = f'passwords_backup_{timestamp}.json'

    result = export_passwords(backup_file)

    if result['success']:
        print(f"✓ Backup created: {backup_file}")
        print(f"  Exported {result['exported_count']} passwords")

        # Move to secure location
        secure_dir = os.path.expanduser('~/Documents/Backups')
        os.makedirs(secure_dir, exist_ok=True)
        os.rename(backup_file, os.path.join(secure_dir, backup_file))
    else:
        print(f"✗ Backup failed: {result['message']}")

def restore_passwords(backup_file):
    """Restore passwords from a backup"""
    # Preview first
    result = import_passwords(backup_file, dry_run=True)
    print(f"Preview: {result['message']}")

    # Confirm
    if input("Continue with import? (y/n): ").lower() == 'y':
        result = import_passwords(backup_file, overwrite=True)
        print(f"✓ {result['message']}")
    else:
        print("Import cancelled")

Migration Script:

from kcpwd import export_passwords, import_passwords, list_all_keys

def migrate_to_new_machine(export_path='~/migration.json'):
    """Export passwords for migration to a new machine"""
    export_path = os.path.expanduser(export_path)

    result = export_passwords(export_path)

    if result['success']:
        print(f"✓ Exported {result['exported_count']} passwords")
        print(f"  File: {export_path}")
        print("\nOn your new machine, run:")
        print(f"  kcpwd import {export_path}")
    else:
        print(f"✗ Export failed: {result['message']}")

def complete_migration(import_path='~/migration.json'):
    """Complete migration on new machine"""
    import_path = os.path.expanduser(import_path)

    # Check existing passwords
    existing = list_all_keys()
    if existing:
        print(f"Found {len(existing)} existing passwords")
        print("Use --overwrite to replace them")

    # Import
    result = import_passwords(import_path)

    if result['success']:
        print(f"✓ {result['message']}")

        # Clean up migration file
        if input("Delete migration file? (y/n): ").lower() == 'y':
            os.remove(import_path)
            print("✓ Migration file deleted")
    else:
        print(f"✗ Import failed: {result['message']}")

Export File Format

The export JSON file has the following structure:

{
  "exported_at": "2025-01-15T10:30:00.123456",
  "service": "kcpwd",
  "version": "0.3.0",
  "include_passwords": true,
  "passwords": [
    {
      "key": "my_database",
      "password": "secret123"
    },
    {
      "key": "api_key",
      "password": "sk-xxxxxxxxxxxxx"
    }
  ]
}

How It Works

kcpwd stores your passwords in the macOS Keychain - the same secure, encrypted storage that Safari and other macOS apps use. This means:

Passwords are encrypted with your Mac's security
They persist across reboots
They're protected by your Mac's login password
No plain text files or databases
Can be accessed programmatically via Python

Viewing Your Passwords

Open Keychain Access app and search for "kcpwd" to see all stored passwords.

Or use Terminal:

security find-generic-password -s "kcpwd" -a "dbadmin" -w

💬 Contribute or share feedback

The project is open-source and actively evolving.
If you’d like to try it, give feedback, or open an issue:

GitHub: https://github.com/osmanuygar/kcpwd

Thanks for reading! If you also love building small but useful dev tools, feel free to share your experience in the comments 🚀

Tags:
#python #security #cli #opensource #devtools