Add admin commands for database operations
Features: - Add admin command group for database operations - admin list-users: List all users from SQLite/PostgreSQL/MySQL - admin user-notes: List notes for specific user - Add CodiMDDatabase class for direct DB access
This commit is contained in:
@@ -9,6 +9,7 @@ from .note import NoteManager
|
||||
from .user import UserManager
|
||||
from .export import ExportManager
|
||||
from .revision import RevisionManager
|
||||
from .db import CodiMDDatabase
|
||||
|
||||
__all__ = [
|
||||
"Config",
|
||||
@@ -18,4 +19,5 @@ __all__ = [
|
||||
"UserManager",
|
||||
"ExportManager",
|
||||
"RevisionManager",
|
||||
"CodiMDDatabase",
|
||||
]
|
||||
|
||||
150
cli_anything/codimd/core/db.py
Normal file
150
cli_anything/codimd/core/db.py
Normal file
@@ -0,0 +1,150 @@
|
||||
"""
|
||||
Database operations for CodiMD CLI.
|
||||
|
||||
Direct database access for admin operations not exposed via API.
|
||||
"""
|
||||
|
||||
import json
|
||||
from typing import List, Dict, Any, Optional
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class CodiMDDatabase:
|
||||
"""Direct database access for CodiMD."""
|
||||
|
||||
@staticmethod
|
||||
def list_users_sqlite(db_path: str) -> List[Dict[str, Any]]:
|
||||
"""List users from SQLite database."""
|
||||
try:
|
||||
import sqlite3
|
||||
except ImportError:
|
||||
raise ImportError("sqlite3 required for SQLite access")
|
||||
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
SELECT id, email, profile, createdAt, updatedAt
|
||||
FROM Users
|
||||
ORDER BY createdAt DESC
|
||||
""")
|
||||
|
||||
users = []
|
||||
for row in cursor.fetchall():
|
||||
user = dict(row)
|
||||
# Parse profile JSON if it's a string
|
||||
if user.get('profile') and isinstance(user['profile'], str):
|
||||
try:
|
||||
user['profile'] = json.loads(user['profile'])
|
||||
except:
|
||||
pass
|
||||
users.append(user)
|
||||
|
||||
conn.close()
|
||||
return users
|
||||
|
||||
@staticmethod
|
||||
def list_users_postgresql(host: str, database: str, user: str, password: str,
|
||||
port: int = 5432) -> List[Dict[str, Any]]:
|
||||
"""List users from PostgreSQL database."""
|
||||
try:
|
||||
import psycopg2
|
||||
except ImportError:
|
||||
raise ImportError("psycopg2 required for PostgreSQL access")
|
||||
|
||||
conn = psycopg2.connect(
|
||||
host=host,
|
||||
database=database,
|
||||
user=user,
|
||||
password=password,
|
||||
port=port
|
||||
)
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
SELECT id, email, profile, "createdAt", "updatedAt"
|
||||
FROM "Users"
|
||||
ORDER BY "createdAt" DESC
|
||||
""")
|
||||
|
||||
users = []
|
||||
for row in cursor.fetchall():
|
||||
user = {
|
||||
'id': row[0],
|
||||
'email': row[1],
|
||||
'profile': row[2],
|
||||
'createdAt': row[3],
|
||||
'updatedAt': row[4]
|
||||
}
|
||||
# Parse profile JSON if it's a string
|
||||
if user.get('profile') and isinstance(user['profile'], str):
|
||||
try:
|
||||
user['profile'] = json.loads(user['profile'])
|
||||
except:
|
||||
pass
|
||||
users.append(user)
|
||||
|
||||
conn.close()
|
||||
return users
|
||||
|
||||
@staticmethod
|
||||
def list_users_mysql(host: str, database: str, user: str, password: str,
|
||||
port: int = 3306) -> List[Dict[str, Any]]:
|
||||
"""List users from MySQL/MariaDB database."""
|
||||
try:
|
||||
import mysql.connector
|
||||
except ImportError:
|
||||
raise ImportError("mysql-connector-python required for MySQL access")
|
||||
|
||||
conn = mysql.connector.connect(
|
||||
host=host,
|
||||
database=database,
|
||||
user=user,
|
||||
password=password,
|
||||
port=port
|
||||
)
|
||||
cursor = conn.cursor(dictionary=True)
|
||||
|
||||
cursor.execute("""
|
||||
SELECT id, email, profile, createdAt, updatedAt
|
||||
FROM Users
|
||||
ORDER BY createdAt DESC
|
||||
""")
|
||||
|
||||
users = cursor.fetchall()
|
||||
|
||||
# Parse profile JSON
|
||||
for user in users:
|
||||
if user.get('profile') and isinstance(user['profile'], str):
|
||||
try:
|
||||
user['profile'] = json.loads(user['profile'])
|
||||
except:
|
||||
pass
|
||||
|
||||
conn.close()
|
||||
return users
|
||||
|
||||
@staticmethod
|
||||
def list_user_notes_sqlite(db_path: str, user_id: str) -> List[Dict[str, Any]]:
|
||||
"""List notes for a specific user from SQLite database."""
|
||||
try:
|
||||
import sqlite3
|
||||
except ImportError:
|
||||
raise ImportError("sqlite3 required for SQLite access")
|
||||
|
||||
conn = sqlite3.connect(db_path)
|
||||
conn.row_factory = sqlite3.Row
|
||||
cursor = conn.cursor()
|
||||
|
||||
cursor.execute("""
|
||||
SELECT id, shortid, alias, title, permission,
|
||||
"lastchangeAt", "createdAt", viewcount
|
||||
FROM Notes
|
||||
WHERE "ownerId" = ?
|
||||
ORDER BY "lastchangeAt" DESC
|
||||
""", (user_id,))
|
||||
|
||||
notes = [dict(row) for row in cursor.fetchall()]
|
||||
conn.close()
|
||||
return notes
|
||||
@@ -113,3 +113,20 @@ class UserManager:
|
||||
raise Exception(f"Failed to get avatar: {response.status_code}")
|
||||
|
||||
return response.content
|
||||
|
||||
def list_all_users(self) -> list:
|
||||
"""List all users (admin only).
|
||||
|
||||
This requires direct database access as CodiMD doesn't expose
|
||||
a user list API. You'll need to provide database configuration.
|
||||
"""
|
||||
# Since CodiMD doesn't have a built-in user list API,
|
||||
# we return a note explaining this limitation
|
||||
raise NotImplementedError(
|
||||
"CodiMD doesn't have a built-in API to list all users. "
|
||||
"This feature requires direct database access. "
|
||||
"You can query the database directly:\n"
|
||||
" PostgreSQL: SELECT id, email, profile FROM Users;\n"
|
||||
" MySQL: SELECT id, email, profile FROM Users;\n"
|
||||
" SQLite: SELECT id, email, profile FROM Users;"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user