feat: Complete Issue #14 - Database Query CLI Interface ⭐ MAJOR MILESTONE
Implement comprehensive database query interface with multiple output formats: • Add query command for executing read-only SQL queries with security constraints • Add schema command for database structure inspection • Add metadata command for file information display • Support table, JSON, and YAML output formats across all commands • Implement SQL injection prevention and safety checks • Add tabulate dependency for enhanced table formatting • Create 35 comprehensive tests covering all functionality This delivers the core USP "Relational Document Metadata" by making the database fully queryable through CLI commands with multiple output formats. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -159,4 +159,102 @@ class DatabaseManager:
|
||||
'created_at': row[3]
|
||||
})
|
||||
|
||||
return files
|
||||
return files
|
||||
|
||||
def execute_query(self, sql: str) -> list:
|
||||
"""
|
||||
Execute a read-only SQL query against the database.
|
||||
|
||||
Args:
|
||||
sql: SQL query string (SELECT operations only)
|
||||
|
||||
Returns:
|
||||
List of dictionaries representing query results
|
||||
|
||||
Raises:
|
||||
ValueError: If query contains non-SELECT operations
|
||||
sqlite3.Error: If query execution fails
|
||||
"""
|
||||
# Security check: only allow SELECT queries
|
||||
sql_upper = sql.strip().upper()
|
||||
if not sql_upper.startswith('SELECT'):
|
||||
allowed_starts = ['SELECT', 'WITH'] # Allow WITH for CTEs
|
||||
if not any(sql_upper.startswith(start) for start in allowed_starts):
|
||||
raise ValueError("Only SELECT and WITH queries are allowed for safety")
|
||||
|
||||
# Additional safety checks for dangerous keywords (as whole words)
|
||||
dangerous_keywords = [
|
||||
'DROP', 'DELETE', 'UPDATE', 'INSERT', 'CREATE', 'ALTER',
|
||||
'TRUNCATE', 'REPLACE', 'PRAGMA'
|
||||
]
|
||||
import re
|
||||
for keyword in dangerous_keywords:
|
||||
# Use word boundaries to match only complete words
|
||||
pattern = r'\b' + keyword + r'\b'
|
||||
if re.search(pattern, sql_upper):
|
||||
raise ValueError(f"Query contains dangerous keyword: {keyword}")
|
||||
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
conn.row_factory = sqlite3.Row # Enable column access by name
|
||||
cursor = conn.cursor()
|
||||
|
||||
try:
|
||||
cursor.execute(sql)
|
||||
rows = cursor.fetchall()
|
||||
|
||||
# Convert rows to dictionaries
|
||||
results = []
|
||||
for row in rows:
|
||||
results.append(dict(row))
|
||||
|
||||
conn.close()
|
||||
return results
|
||||
|
||||
except sqlite3.Error as e:
|
||||
conn.close()
|
||||
raise e
|
||||
|
||||
def get_schema(self) -> dict:
|
||||
"""
|
||||
Get database schema information.
|
||||
|
||||
Returns:
|
||||
Dictionary containing table schemas with column information
|
||||
"""
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
cursor = conn.cursor()
|
||||
|
||||
schema = {}
|
||||
|
||||
try:
|
||||
# Get all table names
|
||||
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
||||
tables = cursor.fetchall()
|
||||
|
||||
for table_row in tables:
|
||||
table_name = table_row[0]
|
||||
|
||||
# Get column information for each table
|
||||
cursor.execute(f"PRAGMA table_info({table_name})")
|
||||
columns = cursor.fetchall()
|
||||
|
||||
column_info = []
|
||||
for col in columns:
|
||||
column_info.append({
|
||||
'name': col[1],
|
||||
'type': col[2],
|
||||
'nullable': not bool(col[3]), # notnull flag
|
||||
'default_value': col[4],
|
||||
'primary_key': bool(col[5])
|
||||
})
|
||||
|
||||
schema[table_name] = {
|
||||
'columns': column_info
|
||||
}
|
||||
|
||||
conn.close()
|
||||
return schema
|
||||
|
||||
except sqlite3.Error as e:
|
||||
conn.close()
|
||||
raise e
|
||||
Reference in New Issue
Block a user