1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
import sqlite3
import json
from pathlib import Path
from datetime import datetime
from typing import Optional, Dict, Any
class DomainCache:
def __init__(self, cache_path: Path = Path.home() / ".local" / "share" / "porkbun" / "cache.sqlite"):
self.cache_path = cache_path
self.cache_path.parent.mkdir(parents=True, exist_ok=True)
self._init_db()
def _init_db(self):
with sqlite3.connect(self.cache_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS domain_prices (
domain TEXT PRIMARY KEY,
data TEXT NOT NULL,
queried_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
def get(self, domain: str) -> Optional[Dict[str, Any]]:
with sqlite3.connect(self.cache_path) as conn:
cursor = conn.execute(
"SELECT data FROM domain_prices WHERE domain = ?",
(domain,)
)
row = cursor.fetchone()
if row:
return json.loads(row[0])
return None
def set(self, domain: str, data: Dict[str, Any]):
with sqlite3.connect(self.cache_path) as conn:
conn.execute(
"""
INSERT OR REPLACE INTO domain_prices (domain, data, queried_at)
VALUES (?, ?, ?)
""",
(domain, json.dumps(data), datetime.utcnow())
)
conn.commit()
def clear(self):
with sqlite3.connect(self.cache_path) as conn:
conn.execute("DELETE FROM domain_prices")
conn.commit()
|
