zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
1#!/usr/bin/env -S uv run --script --quiet
2# /// script
3# requires-python = ">=3.12"
4# dependencies = ["httpx", "pydantic-settings"]
5# ///
6"""Test the exact similar query."""
7
8import os
9import sys
10import httpx
11from pydantic_settings import BaseSettings, SettingsConfigDict
12
13
14class Settings(BaseSettings):
15 model_config = SettingsConfigDict(
16 env_file=os.environ.get("ENV_FILE", ".env"), extra="ignore"
17 )
18 turso_url: str
19 turso_token: str
20
21 @property
22 def turso_host(self) -> str:
23 url = self.turso_url
24 if url.startswith("libsql://"):
25 url = url[len("libsql://"):]
26 return url
27
28
29def query(settings, sql, args=None):
30 stmt = {"sql": sql}
31 if args:
32 stmt["args"] = [{"type": "text", "value": str(a)} for a in args]
33
34 response = httpx.post(
35 f"https://{settings.turso_host}/v2/pipeline",
36 headers={
37 "Authorization": f"Bearer {settings.turso_token}",
38 "Content-Type": "application/json",
39 },
40 json={
41 "requests": [
42 {"type": "execute", "stmt": stmt},
43 {"type": "close"},
44 ]
45 },
46 timeout=30,
47 )
48 response.raise_for_status()
49 return response.json()
50
51
52settings = Settings() # type: ignore
53uri = sys.argv[1] if len(sys.argv) > 1 else "at://did:plc:aub4nfuiysmvcfc4g5gptp47/pub.leaflet.document/3lusgsxusqk2w"
54limit = "6"
55
56print(f"Testing similar query for: {uri}")
57print(f"Limit: {limit}")
58
59# Fixed SQL with CAST
60sql = """
61SELECT d.uri, d.did, d.title, '' as snippet,
62 d.created_at, d.rkey, COALESCE(p.base_path, '') as base_path,
63 CASE WHEN d.publication_uri != '' THEN 1 ELSE 0 END as has_publication
64FROM vector_top_k('documents_embedding_idx',
65 (SELECT embedding FROM documents WHERE uri = ?), CAST(? AS INTEGER)) AS v
66JOIN documents d ON d.rowid = v.id
67LEFT JOIN publications p ON d.publication_uri = p.uri
68WHERE d.uri != ?
69"""
70
71try:
72 result = query(settings, sql, [uri, limit, uri])
73 print(f"\nResult type: {result['results'][0]['type']}")
74 if result['results'][0]['type'] == 'ok':
75 rows = result['results'][0]['response']['result']['rows']
76 print(f"Found {len(rows)} similar documents:")
77 for row in rows[:5]:
78 title = row[2]["value"] if isinstance(row[2], dict) else row[2]
79 print(f" - {title}")
80 else:
81 print(f"Error: {result['results'][0]}")
82except Exception as e:
83 print(f"Exception: {e}")