Spaces:
Sleeping
Sleeping
File size: 6,381 Bytes
c840ad0 |
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 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 |
"""Database service for SQLite operations to store API request logs."""
import sqlite3
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Dict, Any
from contextlib import contextmanager
logger = logging.getLogger(__name__)
class DatabaseService:
"""Service for managing SQLite database operations for API request logging."""
def __init__(self, db_path: str = "api_requests.db"):
"""Initialize the database service.
Args:
db_path: Path to the SQLite database file
"""
self.db_path = Path(db_path)
self._ensure_db_directory()
self._initialize_database()
def _ensure_db_directory(self):
"""Ensure the directory for the database file exists."""
self.db_path.parent.mkdir(parents=True, exist_ok=True)
def _initialize_database(self):
"""Initialize the database and create tables if they don't exist."""
try:
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS api_requests (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp_utc TEXT NOT NULL,
method TEXT NOT NULL,
path TEXT NOT NULL,
client_ip TEXT,
user_agent TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now', 'utc'))
)
"""
)
# Create index on timestamp for faster queries
cursor.execute(
"""
CREATE INDEX IF NOT EXISTS idx_timestamp_utc
ON api_requests(timestamp_utc)
"""
)
# Create index on client_ip for faster queries
cursor.execute(
"""
CREATE INDEX IF NOT EXISTS idx_client_ip
ON api_requests(client_ip)
"""
)
conn.commit()
logger.info(f"✓ Database initialized: {self.db_path}")
except Exception as e:
logger.error(f"Failed to initialize database: {e}")
raise
@contextmanager
def _get_connection(self):
"""Get a database connection with proper error handling."""
conn = None
try:
conn = sqlite3.connect(str(self.db_path), timeout=10.0)
conn.row_factory = sqlite3.Row # Return rows as dictionaries
yield conn
except sqlite3.Error as e:
logger.error(f"Database error: {e}")
if conn:
conn.rollback()
raise
finally:
if conn:
conn.close()
def log_request(
self,
method: str,
path: str,
client_ip: Optional[str] = None,
user_agent: Optional[str] = None,
):
"""Log an API request to the database immediately when received.
Args:
method: HTTP method (GET, POST, etc.)
path: Request path
client_ip: Client IP address
user_agent: User agent string
"""
try:
# Get UTC timestamp
timestamp_utc = datetime.now(timezone.utc).isoformat()
with self._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO api_requests
(timestamp_utc, method, path, client_ip, user_agent)
VALUES (?, ?, ?, ?, ?)
""",
(
timestamp_utc,
method,
path,
client_ip,
user_agent,
),
)
conn.commit()
except Exception as e:
# Log error but don't fail the request
logger.error(f"Failed to log request to database: {e}")
def get_requests(
self,
limit: int = 100,
offset: int = 0,
client_ip: Optional[str] = None,
method: Optional[str] = None,
) -> list[Dict[str, Any]]:
"""Get API requests from the database.
Args:
limit: Maximum number of requests to return
offset: Number of requests to skip
client_ip: Filter by client IP (optional)
method: Filter by HTTP method (optional)
Returns:
List of request dictionaries
"""
try:
with self._get_connection() as conn:
cursor = conn.cursor()
query = "SELECT * FROM api_requests WHERE 1=1"
params = []
if client_ip:
query += " AND client_ip = ?"
params.append(client_ip)
if method:
query += " AND method = ?"
params.append(method)
query += " ORDER BY timestamp_utc DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
cursor.execute(query, params)
rows = cursor.fetchall()
return [dict(row) for row in rows]
except Exception as e:
logger.error(f"Failed to get requests from database: {e}")
return []
def get_request_count(self, client_ip: Optional[str] = None) -> int:
"""Get total count of requests.
Args:
client_ip: Filter by client IP (optional)
Returns:
Total count of requests
"""
try:
with self._get_connection() as conn:
cursor = conn.cursor()
if client_ip:
cursor.execute(
"SELECT COUNT(*) FROM api_requests WHERE client_ip = ?",
(client_ip,),
)
else:
cursor.execute("SELECT COUNT(*) FROM api_requests")
return cursor.fetchone()[0]
except Exception as e:
logger.error(f"Failed to get request count from database: {e}")
return 0
|