You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
75 lines
2.2 KiB
75 lines
2.2 KiB
import sqlite3
|
|
import json
|
|
import time
|
|
import requests
|
|
from typing import Any, Optional
|
|
|
|
class CacheManager:
|
|
def __init__(self, db_name: str = 'pokemon_cache.db'):
|
|
self.conn = sqlite3.connect(db_name)
|
|
self.cursor = self.conn.cursor()
|
|
self._create_cache_table()
|
|
|
|
def _create_cache_table(self):
|
|
self.cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS cache (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT,
|
|
timestamp FLOAT
|
|
)
|
|
''')
|
|
self.conn.commit()
|
|
|
|
def get(self, key: str) -> Optional[Any]:
|
|
self.cursor.execute('SELECT value, timestamp FROM cache WHERE key = ?', (key,))
|
|
result = self.cursor.fetchone()
|
|
if result:
|
|
value, timestamp = result
|
|
return json.loads(value)
|
|
return None
|
|
|
|
def set(self, key: str, value: Any):
|
|
serialized_value = json.dumps(value)
|
|
timestamp = time.time()
|
|
self.cursor.execute('''
|
|
INSERT OR REPLACE INTO cache (key, value, timestamp)
|
|
VALUES (?, ?, ?)
|
|
''', (key, serialized_value, timestamp))
|
|
self.conn.commit()
|
|
|
|
def fetch_url(self, url: str, force_refresh: bool = False, expiry: int = 86400) -> Optional[str]:
|
|
cache_key = f"url_{url}"
|
|
if not force_refresh:
|
|
cached_data = self.get(cache_key)
|
|
if cached_data:
|
|
cached_time = cached_data['timestamp']
|
|
if time.time() - cached_time < expiry:
|
|
return cached_data['content']
|
|
|
|
print(f"Fetching URL: {url}")
|
|
response = requests.get(url)
|
|
if response.status_code == 200:
|
|
content = response.text
|
|
self.set(cache_key, {
|
|
'content': content,
|
|
'timestamp': time.time()
|
|
})
|
|
return content
|
|
return None
|
|
|
|
def close(self):
|
|
self.conn.close()
|
|
|
|
# Usage example
|
|
if __name__ == "__main__":
|
|
cache = CacheManager()
|
|
|
|
# Example usage
|
|
url = "https://example.com"
|
|
data = cache.fetch_url(url)
|
|
if data:
|
|
print("Data fetched successfully")
|
|
else:
|
|
print("Failed to fetch data")
|
|
|
|
cache.close()
|
|
|