Spaces:
Running
Running
File size: 2,984 Bytes
c1e240d 72b3d89 c8079d5 6f03607 c1e240d 72b3d89 6f03607 72b3d89 c1e240d 6f03607 72b3d89 6f03607 72b3d89 c1e240d 72b3d89 6f03607 c1e240d 72b3d89 6f03607 c1e240d 72b3d89 6f03607 72b3d89 6f03607 72b3d89 6f03607 c1e240d 72b3d89 6f03607 |
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 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 |
import os
import sqlite3
from flask import g
from pathlib import Path
class Database:
def __init__(self, db_url):
# Ensure the database directory exists
db_path = self._parse_db_url(db_url)
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
self.db_url = db_url
self.conn = None
def _parse_db_url(self, db_url):
"""Extract the file path from SQLite URL"""
if db_url.startswith('sqlite:///'):
return db_url[10:] # Remove 'sqlite:///'
return db_url
def get_db(self):
if not self.conn:
try:
self.conn = sqlite3.connect(self._parse_db_url(self.db_url))
self.conn.row_factory = sqlite3.Row
# Enable foreign key support
self.conn.execute("PRAGMA foreign_keys = ON")
except sqlite3.Error as e:
print(f"Error connecting to database: {e}")
raise
return self.conn
def close_db(self):
if self.conn:
try:
self.conn.close()
except sqlite3.Error as e:
print(f"Error closing database: {e}")
finally:
self.conn = None
def init_db(self):
db = self.get_db()
try:
# Check if schema.sql exists in the correct path
schema_path = os.path.join(os.path.dirname(__file__), 'schema.sql')
if not os.path.exists(schema_path):
raise FileNotFoundError(f"Schema file not found at {schema_path}")
with open(schema_path, 'r') as f:
db.cursor().executescript(f.read())
db.commit()
except Exception as e:
print(f"Error initializing database: {e}")
raise
def cache_news(self, articles):
db = self.get_db()
try:
for article in articles:
db.execute('''
INSERT OR IGNORE INTO news
(title, source, published, url, summary, content)
VALUES (?,?,?,?,?,?)
''', (
article.get('title', ''),
article.get('source', ''),
article.get('published', ''),
article.get('url', ''),
article.get('summary', ''),
article.get('content', '')
))
db.commit()
except sqlite3.Error as e:
print(f"Error caching news: {e}")
db.rollback()
raise
def get_cached_news(self):
db = self.get_db()
try:
cur = db.execute('''
SELECT * FROM news
ORDER BY published DESC
LIMIT 15
''')
return [dict(row) for row in cur.fetchall()]
except sqlite3.Error as e:
print(f"Error fetching cached news: {e}")
raise |