Building a DuckDuckGo search storage app with FastAPI

Search engines have become an integral part of our daily digital lives, but have you ever wanted to save and organize your search results for later reference? Today, we’re diving deep into DDG Search Store – an elegant FastAPI-based solution that lets you search DuckDuckGo, store results, and manage them through a RESTful API.

Want to stay updated with more technical deep-dives and tutorials? Subscribe to our blog at mydaytodo.com/blog for weekly insights on modern web development.

What is DDG Search Store?

DDG Search Store is a lightweight, production-ready API that bridges the gap between ephemeral search results and persistent data storage. Built with Python’s FastAPI framework, it allows you to:

Search DuckDuckGo programmatically without API keys
Store search results in a SQLite database for later retrieval
Manage saved searches through full CRUD operations
Query historical data to track search trends or build recommendation systems

The project demonstrates modern Python development practices, combining the speed of FastAPI with the simplicity of SQLModel for database operations.

Technology Stack: Modern Python at Its Best

FastAPI – The Performance Champion

FastAPI has revolutionized Python web development with its async-first approach and automatic API documentation. Here’s why it’s perfect for this project:

from fastapi import FastAPI, HTTPException, Depends
from typing import List

app = FastAPI(
    title="DuckDuckGo Search Store API",
    description="Store and retrieve DuckDuckGo search results",
    version="1.0.0"
)

@app.get("/")
async def root():
    return {"message": "DuckDuckGo Search Store API is running"}

from fastapi import FastAPI, HTTPException, Depends
from typing import List

app = FastAPI(
    title="DuckDuckGo Search Store API",
    description="Store and retrieve DuckDuckGo search results",
    version="1.0.0"
)

@app.get("/")
async def root():
    return {"message": "DuckDuckGo Search Store API is running"}

Python

Key Benefits:

Automatic OpenAPI documentation at /docs and /redoc
Type validation using Python type hints
Async/await support for concurrent requests
Dependency injection for clean architecture

SQLModel – Type-Safe Database Operations

SQLModel combines the best of SQLAlchemy and Pydantic, providing both ORM functionality and data validation:

from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime

class SearchResult(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    query: str = Field(index=True)
    title: str
    url: str = Field(unique=True)
    snippet: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    
    class Config:
        schema_extra = {
            "example": {
                "query": "FastAPI tutorial",
                "title": "FastAPI - Official Documentation",
                "url": "https://fastapi.tiangolo.com",
                "snippet": "FastAPI framework, high performance..."
            }
        }

from sqlmodel import SQLModel, Field
from typing import Optional
from datetime import datetime

class SearchResult(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    query: str = Field(index=True)
    title: str
    url: str = Field(unique=True)
    snippet: str
    created_at: datetime = Field(default_factory=datetime.utcnow)
    
    class Config:
        schema_extra = {
            "example": {
                "query": "FastAPI tutorial",
                "title": "FastAPI - Official Documentation",
                "url": "https://fastapi.tiangolo.com",
                "snippet": "FastAPI framework, high performance..."
            }
        }

Python

This model definition serves triple duty:

Database schema for SQLAlchemy
Request/response validation via Pydantic
API documentation examples

DuckDuckGo Search (DDGS) – The Search Engine

The duckduckgo-search library provides a clean Python interface to DuckDuckGo’s search functionality without requiring API keys:

from duckduckgo_search import DDGS
from typing import List, Dict

class SearchService:
    def __init__(self):
        self.ddgs = DDGS()
    
    async def search(self, query: str, max_results: int = 10) -> List[Dict]:
        """
        Perform DuckDuckGo search and return results
        """
        try:
            results = []
            with self.ddgs as ddgs:
                for result in ddgs.text(query, max_results=max_results):
                    results.append({
                        "title": result.get("title", ""),
                        "url": result.get("href", ""),
                        "snippet": result.get("body", "")
                    })
            return results
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}")

from duckduckgo_search import DDGS
from typing import List, Dict

class SearchService:
    def __init__(self):
        self.ddgs = DDGS()
    
    async def search(self, query: str, max_results: int = 10) -> List[Dict]:
        """
        Perform DuckDuckGo search and return results
        """
        try:
            results = []
            with self.ddgs as ddgs:
                for result in ddgs.text(query, max_results=max_results):
                    results.append({
                        "title": result.get("title", ""),
                        "url": result.get("href", ""),
                        "snippet": result.get("body", "")
                    })
            return results
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}")

Python

SQLite – Simple Yet Powerful

For development and small-to-medium deployments, SQLite provides:

Zero configuration – no separate database server needed
ACID compliance – reliable transactions
Portability – entire database in a single file
Speed – excellent performance for read-heavy workloads

Core Implementation: Building the Search API

Database Setup and Session Management

from sqlmodel import SQLModel, create_engine, Session
from typing import Generator

DATABASE_URL = "sqlite:///./search_store.db"

engine = create_engine(
    DATABASE_URL,
    echo=True,  # Log SQL queries in development
    connect_args={"check_same_thread": False}  # SQLite specific
)

def create_db_and_tables():
    """Initialize database and create tables"""
    SQLModel.metadata.create_all(engine)

def get_session() -> Generator[Session, None, None]:
    """Dependency for database sessions"""
    with Session(engine) as session:
        yield session

from sqlmodel import SQLModel, create_engine, Session
from typing import Generator

DATABASE_URL = "sqlite:///./search_store.db"

engine = create_engine(
    DATABASE_URL,
    echo=True,  # Log SQL queries in development
    connect_args={"check_same_thread": False}  # SQLite specific
)

def create_db_and_tables():
    """Initialize database and create tables"""
    SQLModel.metadata.create_all(engine)

def get_session() -> Generator[Session, None, None]:
    """Dependency for database sessions"""
    with Session(engine) as session:
        yield session

Python

Search and Store Endpoint

The core functionality combines searching and storing in a single transaction:

from sqlmodel import Session, select
from fastapi import Depends

@app.post("/search", response_model=List[SearchResult])
async def search_and_store(
    query: str,
    max_results: int = 10,
    session: Session = Depends(get_session)
):
    """
    Search DuckDuckGo and store unique results in database
    """
    search_service = SearchService()
    
    # Perform search
    search_results = await search_service.search(query, max_results)
    
    stored_results = []
    for result in search_results:
        # Check if URL already exists
        statement = select(SearchResult).where(SearchResult.url == result["url"])
        existing = session.exec(statement).first()
        
        if not existing:
            # Create new search result
            search_result = SearchResult(
                query=query,
                title=result["title"],
                url=result["url"],
                snippet=result["snippet"]
            )
            session.add(search_result)
            stored_results.append(search_result)
    
    session.commit()
    
    # Refresh to get auto-generated IDs
    for result in stored_results:
        session.refresh(result)
    
    return stored_results

from sqlmodel import Session, select
from fastapi import Depends

@app.post("/search", response_model=List[SearchResult])
async def search_and_store(
    query: str,
    max_results: int = 10,
    session: Session = Depends(get_session)
):
    """
    Search DuckDuckGo and store unique results in database
    """
    search_service = SearchService()
    
    # Perform search
    search_results = await search_service.search(query, max_results)
    
    stored_results = []
    for result in search_results:
        # Check if URL already exists
        statement = select(SearchResult).where(SearchResult.url == result["url"])
        existing = session.exec(statement).first()
        
        if not existing:
            # Create new search result
            search_result = SearchResult(
                query=query,
                title=result["title"],
                url=result["url"],
                snippet=result["snippet"]
            )
            session.add(search_result)
            stored_results.append(search_result)
    
    session.commit()
    
    # Refresh to get auto-generated IDs
    for result in stored_results:
        session.refresh(result)
    
    return stored_results

Python

CRUD Operations

Complete data management through RESTful endpoints:

@app.get("/searches", response_model=List[SearchResult])
async def get_all_searches(
    skip: int = 0,
    limit: int = 100,
    session: Session = Depends(get_session)
):
    """Retrieve all stored search results with pagination"""
    statement = select(SearchResult).offset(skip).limit(limit)
    results = session.exec(statement).all()
    return results

@app.get("/searches/{search_id}", response_model=SearchResult)
async def get_search_by_id(
    search_id: int,
    session: Session = Depends(get_session)
):
    """Get a specific search result by ID"""
    result = session.get(SearchResult, search_id)
    if not result:
        raise HTTPException(status_code=404, detail="Search result not found")
    return result

@app.put("/searches/{search_id}", response_model=SearchResult)
async def update_search(
    search_id: int,
    updated_data: SearchResult,
    session: Session = Depends(get_session)
):
    """Update an existing search result"""
    result = session.get(SearchResult, search_id)
    if not result:
        raise HTTPException(status_code=404, detail="Search result not found")
    
    result.title = updated_data.title
    result.snippet = updated_data.snippet
    session.add(result)
    session.commit()
    session.refresh(result)
    return result

@app.delete("/searches/{search_id}")
async def delete_search(
    search_id: int,
    session: Session = Depends(get_session)
):
    """Delete a search result"""
    result = session.get(SearchResult, search_id)
    if not result:
        raise HTTPException(status_code=404, detail="Search result not found")
    
    session.delete(result)
    session.commit()
    return {"message": "Search result deleted successfully"}

@app.get("/searches", response_model=List[SearchResult])
async def get_all_searches(
    skip: int = 0,
    limit: int = 100,
    session: Session = Depends(get_session)
):
    """Retrieve all stored search results with pagination"""
    statement = select(SearchResult).offset(skip).limit(limit)
    results = session.exec(statement).all()
    return results

@app.get("/searches/{search_id}", response_model=SearchResult)
async def get_search_by_id(
    search_id: int,
    session: Session = Depends(get_session)
):
    """Get a specific search result by ID"""
    result = session.get(SearchResult, search_id)
    if not result:
        raise HTTPException(status_code=404, detail="Search result not found")
    return result

@app.put("/searches/{search_id}", response_model=SearchResult)
async def update_search(
    search_id: int,
    updated_data: SearchResult,
    session: Session = Depends(get_session)
):
    """Update an existing search result"""
    result = session.get(SearchResult, search_id)
    if not result:
        raise HTTPException(status_code=404, detail="Search result not found")
    
    result.title = updated_data.title
    result.snippet = updated_data.snippet
    session.add(result)
    session.commit()
    session.refresh(result)
    return result

@app.delete("/searches/{search_id}")
async def delete_search(
    search_id: int,
    session: Session = Depends(get_session)
):
    """Delete a search result"""
    result = session.get(SearchResult, search_id)
    if not result:
        raise HTTPException(status_code=404, detail="Search result not found")
    
    session.delete(result)
    session.commit()
    return {"message": "Search result deleted successfully"}

Python

Getting Started: Run It Locally in Minutes

# Clone the repository
git clone https://github.com/cptdanko/ddg-search-store.git
cd ddg-search-store

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Run the application
uvicorn main:app --reload

# Access the API documentation
# http://localhost:8000/docs

# Clone the repository
git clone https://github.com/cptdanko/ddg-search-store.git
cd ddg-search-store

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Run the application
uvicorn main:app --reload

# Access the API documentation
# http://localhost:8000/docs

Bash

Taking It Further: 10 Enhancement Ideas

1. User Authentication with JWT

from fastapi.security import OAuth2PasswordBearer
from jose import jwt
from passlib.context import CryptContext

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(unique=True)
    hashed_password: str
    is_active: bool = True

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # JWT validation logic
    pass

from fastapi.security import OAuth2PasswordBearer
from jose import jwt
from passlib.context import CryptContext

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(unique=True)
    hashed_password: str
    is_active: bool = True

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # JWT validation logic
    pass

Python

Why it matters: Protect your API and enable per-user search history.

2. Redis Caching Layer

import redis.asyncio as redis
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

async def cached_search(query: str, ttl: int = 3600):
    cached = await redis_client.get(f"search:{query}")
    if cached:
        return json.loads(cached)
    
    results = await search_service.search(query)
    await redis_client.setex(f"search:{query}", ttl, json.dumps(results))
    return results

import redis.asyncio as redis
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

async def cached_search(query: str, ttl: int = 3600):
    cached = await redis_client.get(f"search:{query}")
    if cached:
        return json.loads(cached)
    
    results = await search_service.search(query)
    await redis_client.setex(f"search:{query}", ttl, json.dumps(results))
    return results

Python

Benefits: Reduce external API calls and improve response times by 10-100x.

3. Async Database Operations with SQLAlchemy

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine(
    "sqlite+aiosqlite:///./search_store.db",
    echo=True
)

async def get_async_session() -> AsyncSession:
    async with AsyncSession(async_engine) as session:
        yield session

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

async_engine = create_async_engine(
    "sqlite+aiosqlite:///./search_store.db",
    echo=True
)

async def get_async_session() -> AsyncSession:
    async with AsyncSession(async_engine) as session:
        yield session

Python

Performance boost: Handle more concurrent requests without blocking.

4. Full-Text Search with PostgreSQL

from sqlalchemy import Column, Index
from sqlalchemy.dialects.postgresql import TSVECTOR

class SearchResult(SQLModel, table=True):
    # ...existing code...
    search_vector: Optional[str] = Field(sa_column=Column(TSVECTOR))
    
    __table_args__ = (
        Index('idx_search_vector', 'search_vector', postgresql_using='gin'),
    )

# Search query
statement = select(SearchResult).where(
    SearchResult.search_vector.match('fastapi tutorial')
)

from sqlalchemy import Column, Index
from sqlalchemy.dialects.postgresql import TSVECTOR

class SearchResult(SQLModel, table=True):
    # ...existing code...
    search_vector: Optional[str] = Field(sa_column=Column(TSVECTOR))
    
    __table_args__ = (
        Index('idx_search_vector', 'search_vector', postgresql_using='gin'),
    )

# Search query
statement = select(SearchResult).where(
    SearchResult.search_vector.match('fastapi tutorial')
)

Python

Advanced querying: Relevance ranking and fuzzy matching for better search results.

5. Analytics Dashboard

Track search trends and popular queries:

@app.get("/analytics/top-queries")
async def get_top_queries(
    limit: int = 10,
    session: Session = Depends(get_session)
):
    statement = select(
        SearchResult.query,
        func.count(SearchResult.id).label('count')
    ).group_by(SearchResult.query).order_by(desc('count')).limit(limit)
    
    results = session.exec(statement).all()
    return [{"query": q, "count": c} for q, c in results]

@app.get("/analytics/top-queries")
async def get_top_queries(
    limit: int = 10,
    session: Session = Depends(get_session)
):
    statement = select(
        SearchResult.query,
        func.count(SearchResult.id).label('count')
    ).group_by(SearchResult.query).order_by(desc('count')).limit(limit)
    
    results = session.exec(statement).all()
    return [{"query": q, "count": c} for q, c in results]

Python

6. Multi-Provider Search Aggregation

class MultiSearchService:
    async def aggregate_search(self, query: str):
        ddg_results = await self.search_ddg(query)
        bing_results = await self.search_bing(query)
        
        # Deduplicate and rank
        return self.merge_and_rank(ddg_results, bing_results)

class MultiSearchService:
    async def aggregate_search(self, query: str):
        ddg_results = await self.search_ddg(query)
        bing_results = await self.search_bing(query)
        
        # Deduplicate and rank
        return self.merge_and_rank(ddg_results, bing_results)

Python

7. Export Functionality

import csv
from io import StringIO
from fastapi.responses import StreamingResponse

@app.get("/export/csv")
async def export_to_csv(session: Session = Depends(get_session)):
    results = session.exec(select(SearchResult)).all()
    
    output = StringIO()
    writer = csv.DictWriter(output, fieldnames=['query', 'title', 'url', 'snippet'])
    writer.writeheader()
    for result in results:
        writer.writerow({
            'query': result.query,
            'title': result.title,
            'url': result.url,
            'snippet': result.snippet
        })
    
    output.seek(0)
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=searches.csv"}
    )

import csv
from io import StringIO
from fastapi.responses import StreamingResponse

@app.get("/export/csv")
async def export_to_csv(session: Session = Depends(get_session)):
    results = session.exec(select(SearchResult)).all()
    
    output = StringIO()
    writer = csv.DictWriter(output, fieldnames=['query', 'title', 'url', 'snippet'])
    writer.writeheader()
    for result in results:
        writer.writerow({
            'query': result.query,
            'title': result.title,
            'url': result.url,
            'snippet': result.snippet
        })
    
    output.seek(0)
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=searches.csv"}
    )

Python

8. WebSocket Real-Time Updates

from fastapi import WebSocket

@app.websocket("/ws/searches")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_json()
        query = data.get("query")
        
        # Stream search results as they arrive
        async for result in search_service.stream_search(query):
            await websocket.send_json(result)

from fastapi import WebSocket

@app.websocket("/ws/searches")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_json()
        query = data.get("query")
        
        # Stream search results as they arrive
        async for result in search_service.stream_search(query):
            await websocket.send_json(result)

Python

Real-World Applications

This API can power various applications:

Research Tools: Academics tracking sources across multiple searches
SEO Monitoring: Track ranking changes for specific keywords
Content Discovery: Build recommendation engines based on search patterns
Competitive Analysis: Monitor competitor mentions and trends
Personal Knowledge Base: Create your own searchable archive

Performance Considerations

For production deployments:

# Use connection pooling
from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

# Implement rate limiting
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/search")
@limiter.limit("10/minute")
async def rate_limited_search():
    pass

# Use connection pooling
from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

# Implement rate limiting
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

@app.get("/search")
@limiter.limit("10/minute")
async def rate_limited_search():
    pass

Python

Conclusion

DDG Search Store demonstrates how modern Python frameworks can create powerful, production-ready APIs with minimal code. The combination of FastAPI’s performance, SQLModel’s type safety, and DuckDuckGo’s search capabilities creates a solid foundation for building search-based applications.

Whether you’re building a personal research tool or a commercial search aggregator, this project provides the essential building blocks. The enhancement ideas above show how you can extend it to meet specific requirements while maintaining clean, maintainable code.

Ready to dive deeper into API development and Python best practices? Subscribe to mydaytodo.com/blog for weekly tutorials, code reviews, and in-depth technical guides. Join our community of developers building modern web applications!

Resources

GitHub Repository: github.com/cptdanko/ddg-search-store
FastAPI Documentation: FastAPI
SQLModel Docs: sqlmodel.tiangolo.com
More Tutorials: mydaytodo.com/blog
- Build a FastAPI Streaming API with Llama 3 and Ollama – local llm – My Day To-Do
- How to Build a CQRS‑Based CRUD API with FastAPI and MongoDB

Have questions or built something cool with this API? Share your experience in the comments below or join the discussion on our blog!

Post Views: 17

Building a DuckDuckGo Search Storage API with FastAPI and SQLModel

Published by Bhuman Soni on January 20, 2026January 20, 2026

What is DDG Search Store?

Technology Stack: Modern Python at Its Best

FastAPI – The Performance Champion

SQLModel – Type-Safe Database Operations

DuckDuckGo Search (DDGS) – The Search Engine

SQLite – Simple Yet Powerful

Core Implementation: Building the Search API

Database Setup and Session Management

Search and Store Endpoint

CRUD Operations

Taking It Further: 10 Enhancement Ideas

1. User Authentication with JWT

2. Redis Caching Layer

3. Async Database Operations with SQLAlchemy

4. Full-Text Search with PostgreSQL

5. Analytics Dashboard

Real-World Applications

Performance Considerations

Conclusion

Resources

Related

0 Comments

Leave a ReplyCancel reply

Learn BFF pattern implementation: Node.js, React, TypeScript, and Java Examples.

Implementing Circuit Breaker Pattern in Spring Boot 3 with Resilience4j

Implementing Saga Pattern in Spring Boot Microservices: Banking example with Kong Gateway

Building a DuckDuckGo Search Storage API with FastAPI and SQLModel

Published by Bhuman Soni on January 20, 2026January 20, 2026

What is DDG Search Store?

Technology Stack: Modern Python at Its Best

FastAPI – The Performance Champion

SQLModel – Type-Safe Database Operations

DuckDuckGo Search (DDGS) – The Search Engine

SQLite – Simple Yet Powerful

Core Implementation: Building the Search API

Database Setup and Session Management

Search and Store Endpoint

CRUD Operations

Taking It Further: 10 Enhancement Ideas

1. User Authentication with JWT

2. Redis Caching Layer

3. Async Database Operations with SQLAlchemy

4. Full-Text Search with PostgreSQL

5. Analytics Dashboard

Real-World Applications

Performance Considerations

Conclusion

Resources

Related

0 Comments

Leave a ReplyCancel reply

Related Posts

Learn BFF pattern implementation: Node.js, React, TypeScript, and Java Examples.

Implementing Circuit Breaker Pattern in Spring Boot 3 with Resilience4j

Implementing Saga Pattern in Spring Boot Microservices: Banking example with Kong Gateway