How to Build a URL Metadata Cache with Python and SQLite

Fetching metadata when a user opens a page makes link previews feel slow and turns every refresh into another network request. A small SQLite cache moves that work to ingestion time, gives you predictable reads, and keeps the raw response available when your UI needs a new field later.

The Link Preview & Metadata Extractor handles the remote fetch and parses Open Graph, Twitter Cards, canonical URLs, favicons, language, and HTTP details. Python only needs to call the actor, normalize the useful fields, and decide when an entry is stale.

Start with the synchronous actor endpoint

The actor takes one required url. Optional inputs let you include favicon discovery, set a custom user agent, and choose a request timeout between 1 and 120 seconds.

For an ingestion worker, Apify's synchronous dataset endpoint is simpler than starting a run and polling it. The response is the dataset item, so one function can return either metadata or a structured actor error.

import os
from typing import Any

import requests

APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~link-preview-metadata-extractor"
RUN_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def fetch_metadata(url: str, timeout_seconds: int = 15) -> dict[str, Any]:
    response = requests.post(
        RUN_URL,
        params={"token": APIFY_TOKEN},
        json={
            "url": url,
            "includeFavicon": True,
            "timeout": timeout_seconds,
        },
        timeout=timeout_seconds + 45,
    )
    response.raise_for_status()
    items = response.json()
    if not items:
        raise RuntimeError(f"No metadata item returned for {url}")

    item = items[0]
    if item.get("error"):
        error_type = item.get("error_type", "actor_error")
        raise RuntimeError(f"{error_type}: {item['error']}")
    return item

Keep APIFY_TOKEN in the environment rather than in the database or source file. The HTTP timeout is intentionally longer than the actor's page timeout because the platform also needs time to start the run and return its dataset.

Preserve raw data and index the fields you query

Metadata formats change more often than database schemas should. One site may provide og:title, another only a regular title, and a third may omit both but include Twitter Card data.

The practical compromise is to store the full actor item as JSON while copying a few frequently queried values into columns. The helper below flattens nested dictionaries, so it works whether a field is returned at the top level or inside an Open Graph object.

import json
from collections.abc import Mapping
from typing import Any


def flatten_mapping(value: Mapping[str, Any], prefix: str = "") -> dict[str, Any]:
    flattened = {}
    for key, child in value.items():
        path = f"{prefix}.{key}" if prefix else key
        flattened[path] = child
        if isinstance(child, Mapping):
            flattened.update(flatten_mapping(child, path))
    return flattened


def first_value(item: dict[str, Any], *names: str) -> Any:
    flat = flatten_mapping(item)
    for name in names:
        for path, value in flat.items():
            if path == name or path.endswith(f".{name}"):
                if value not in (None, "", []):
                    return value
    return None


def normalize_item(requested_url: str, item: dict[str, Any]) -> dict[str, Any]:
    canonical = first_value(item, "canonical_url", "canonical", "og:url")
    return {
        "requested_url": requested_url,
        "canonical_url": canonical or requested_url,
        "title": first_value(item, "og:title", "twitter:title", "title"),
        "description": first_value(
            item,
            "og:description",
            "twitter:description",
            "description",
        ),
        "image_url": first_value(item, "og:image", "twitter:image"),
        "favicon_url": first_value(item, "favicon", "favicon_url"),
        "status_code": first_value(item, "status_code"),
        "raw_json": json.dumps(item, sort_keys=True),
    }

Saving the raw object matters. If you later decide to display language, author, robots directives, or response timing, you can backfill columns from cached responses instead of re-fetching every URL.

Create a cache that survives restarts

SQLite is enough for a personal knowledge tool, internal dashboard, or small content product. Use the requested URL as the lookup key and keep the canonical URL separately for deduplication.

import sqlite3
from pathlib import Path

DATABASE_PATH = Path("link-metadata.sqlite3")


def connect() -> sqlite3.Connection:
    database = sqlite3.connect(DATABASE_PATH)
    database.row_factory = sqlite3.Row
    database.execute("PRAGMA journal_mode=WAL")
    database.execute("PRAGMA busy_timeout=5000")
    database.execute(
        """
        CREATE TABLE IF NOT EXISTS link_metadata (
            requested_url TEXT PRIMARY KEY,
            canonical_url TEXT NOT NULL,
            title TEXT,
            description TEXT,
            image_url TEXT,
            favicon_url TEXT,
            status_code INTEGER,
            raw_json TEXT NOT NULL,
            fetched_at TEXT NOT NULL
        )
        """
    )
    database.execute(
        "CREATE INDEX IF NOT EXISTS metadata_canonical_idx "
        "ON link_metadata(canonical_url)"
    )
    return database

WAL mode lets readers continue while an ingestion worker writes. The busy timeout prevents a short overlapping transaction from immediately failing with a locked database error.

Upsert atomically after a successful extraction

Do not replace a good cache entry before the actor returns valid data. If the target times out or returns non-HTML content, the previous preview can remain available while the failure goes to your logs.

from datetime import datetime, timezone


def save_metadata(database: sqlite3.Connection, record: dict) -> None:
    database.execute(
        """
        INSERT INTO link_metadata (
            requested_url, canonical_url, title, description,
            image_url, favicon_url, status_code, raw_json, fetched_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(requested_url) DO UPDATE SET
            canonical_url = excluded.canonical_url,
            title = excluded.title,
            description = excluded.description,
            image_url = excluded.image_url,
            favicon_url = excluded.favicon_url,
            status_code = excluded.status_code,
            raw_json = excluded.raw_json,
            fetched_at = excluded.fetched_at
        """,
        (
            record["requested_url"],
            record["canonical_url"],
            record["title"],
            record["description"],
            record["image_url"],
            record["favicon_url"],
            record["status_code"],
            record["raw_json"],
            datetime.now(timezone.utc).isoformat(),
        ),
    )
    database.commit()


def enrich_url(database: sqlite3.Connection, url: str) -> dict:
    item = fetch_metadata(url)
    record = normalize_item(url, item)
    save_metadata(database, record)
    return record

This transaction is small, but it gives the cache a useful guarantee: readers see either the previous complete record or the new complete record, never half of each.

Refresh only stale entries

Metadata does not need to be fetched on every request. News links may deserve a daily refresh, while documentation and reference pages can sit for a week or longer.

from datetime import datetime, timedelta, timezone


def get_cached(
    database: sqlite3.Connection,
    url: str,
    max_age: timedelta = timedelta(days=7),
) -> dict | None:
    row = database.execute(
        "SELECT * FROM link_metadata WHERE requested_url = ?",
        (url,),
    ).fetchone()
    if row is None:
        return None

    fetched_at = datetime.fromisoformat(row["fetched_at"])
    if datetime.now(timezone.utc) - fetched_at > max_age:
        return None
    return dict(row)


def metadata_for_url(database: sqlite3.Connection, url: str) -> dict:
    cached = get_cached(database, url)
    return cached if cached is not None else enrich_url(database, url)

In a web app, keep stale refreshes off the request path. Return the old record immediately and put the URL on a worker queue. The next request gets fresh metadata without making the current user wait for an external fetch.

Deduplicate tracking URLs by canonical URL

Requested URLs still need separate cache entries because they may redirect differently. For your application model, however, two records with the same canonical URL usually represent the same resource.

def canonical_duplicates(database: sqlite3.Connection) -> list[dict]:
    rows = database.execute(
        """
        SELECT canonical_url,
               COUNT(*) AS url_count,
               GROUP_CONCAT(requested_url, '\n') AS requested_urls
        FROM link_metadata
        GROUP BY canonical_url
        HAVING COUNT(*) > 1
        ORDER BY url_count DESC
        """
    ).fetchall()
    return [dict(row) for row in rows]

This catches campaign parameters, URL shorteners, and alternate share links after the destination declares a canonical URL. Do not blindly merge when the canonical is missing. Falling back to the requested URL is safer than guessing that two similar paths are equivalent.

Cost and operating limits

The Link Preview & Metadata Extractor is currently listed on Apify as pay per usage. Its store pricing depends on the platform resources consumed and the account plan, so check the live pricing page before estimating a large import. Because it fetches HTML over HTTP without a headless browser, it avoids the browser startup work that makes screenshot-based extraction heavier.

Keep a modest worker pool for batch imports and retry only transient network failures. A permanent non_html or HTTP error should be recorded for review rather than retried in a tight loop. Back up the SQLite file with its WAL checkpointed, and retain raw_json even if the preview UI currently uses only three fields.