About TimescaleDB Migrator
-
What is TimescaleDB Migrator?
-
TimescaleDB Migrator is a tool designed to simplify the process of migrating data between TimescaleDB instances using the PostgreSQL native backup and restore utilities: pg_dump and pg_restore.
-
Key Features
-
- Easy Database Migration: Migrate your entire TimescaleDB database with just a few clicks
- Secure Connections: Support for secure connections with password protection
@@ -2650,29 +2346,22 @@ async def home(request: Request):
- Real-time Monitoring: Track the progress of your dump and restore operations
- TimescaleDB-aware: Handles TimescaleDB-specific migration requirements
-
How It Works
-
Dump Operation
The dump operation uses pg_dump to create a backup of your source database. This backup can be in various formats (custom, directory, plain SQL, or tar) and with different compression levels.
-
Restore Operation
The restore operation uses pg_restore to import your backup into the target database. It includes TimescaleDB-specific pre and post-restore functions to ensure data integrity.
-
Commands Used
pg_dump
"postgres://user:password@source-host:5432/source_db"
-Fc
-v
-f
~/timescale_backup.dump
-
psql
"postgres://user:password@target-host:5432/target_db"
-c
"SELECT timescaledb_pre_restore();"
-
pg_restore
-d
"postgres://user:password@target-host:5432/target_db"
-v
--no-owner
~/timescale_backup.dump
-
psql
"postgres://user:password@target-host:5432/target_db"
-c
"SELECT timescaledb_post_restore(); ANALYZE;"
@@ -2683,10 +2372,8 @@ async def home(request: Request):
-
-
@@ -2700,12 +2387,11 @@ async def home(request: Request):
Are you sure you want to proceed with this action?
-
- """)
-
+"""
+ # Update the file content
+ with open("templates/index.html", "w", encoding="utf-8") as f:
+ f.write(html_content)
+
return templates.TemplateResponse("index.html", {"request": request})
@app.post("/test-connection")
-async def test_connection(data: Dict[str, str]):
- """Test a database connection"""
+async def test_connection_endpoint(data: Dict[str, str]):
+ """Test a database connection and get basic info"""
try:
connection_string = data.get("connection_string")
- connection_type = data.get("connection_type", "source")
-
+ connection_type = data.get("connection_type", "source") # Added type
if not connection_string:
return JSONResponse(
status_code=400,
content={"success": False, "message": "Connection string is required"}
)
-
- # Test the connection
- if not test_connection(connection_string):
+
+ # Test the connection using internal logic
+ if not test_connection_logic(connection_string):
+ # Error is logged within test_connection_logic
return JSONResponse(
content={"success": False, "message": "Failed to connect to database"}
)
-
- # Get database info
+
+ # If connection successful, get database info
conn = psycopg2.connect(connection_string)
-
try:
with conn.cursor() as cur:
# Get server info
cur.execute("SELECT version()")
- version = cur.fetchone()[0]
-
- # Check if TimescaleDB is installed
+ version_result = cur.fetchone()
+ version = version_result[0] if version_result else "Unknown"
+
+ # Check if TimescaleDB is installed and get version
is_timescaledb = False
+ ts_version = None
try:
- cur.execute("SELECT extversion FROM pg_extension WHERE extname = 'timescaledb'")
- ts_version = cur.fetchone()
- is_timescaledb = ts_version is not None
- except:
- pass
-
+ # Use EXISTS for better performance and error handling if extension not present
+ cur.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb');")
+ if cur.fetchone()[0]:
+ cur.execute("SELECT extversion FROM pg_extension WHERE extname = 'timescaledb'")
+ ts_version_result = cur.fetchone()
+ is_timescaledb = ts_version_result is not None
+ ts_version = ts_version_result[0] if ts_version_result else None
+ except psycopg2.Error as ts_err:
+ logger.warning(f"Could not check TimescaleDB extension: {ts_err}")
+ # Continue without TimescaleDB info if query fails
+
# Get database name
cur.execute("SELECT current_database()")
- database = cur.fetchone()[0]
-
- # Extract server info from version string
- server_match = version.split(" on ")[1].split(",")[0] if " on " in version else "Unknown"
-
- log_message(f"Successful connection test to {connection_type} database: {database}", "success")
-
+ db_result = cur.fetchone()
+ database = db_result[0] if db_result else "Unknown"
+
+ # Extract server host/IP (best effort parsing)
+ server_match = "Unknown"
+ try:
+ # Extract from connection string if possible (more reliable)
+ host_part = connection_string.split('@')[-1].split('/')[0].split(':')[0]
+ if host_part:
+ server_match = host_part
+ # Fallback to parsing version string if needed
+ elif " on " in version:
+ server_match = version.split(" on ")[-1].split(",")[0]
+ except Exception:
+ logger.warning("Could not parse server host from connection string or version.")
+
+
+ log_message(f"Successful connection test to {connection_type} database: {database} on {server_match}", "success")
return JSONResponse(content={
"success": True,
"version": version,
"is_timescaledb": is_timescaledb,
+ "timescaledb_version": ts_version, # Add this line
"database": database,
"server": server_match
})
finally:
conn.close()
-
+ except psycopg2.Error as db_err:
+ log_message(f"Database connection error during info fetch: {str(db_err)}", "error")
+ return JSONResponse(
+ content={"success": False, "message": f"Database error: {str(db_err)}"}
+ )
except Exception as e:
- log_message(f"Connection test failed: {str(e)}", "error")
+ log_message(f"Connection test failed unexpectedly: {str(e)}", "error")
return JSONResponse(
- content={"success": False, "message": str(e)}
+ status_code=500,
+ content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
+ )
+
+@app.post("/database-info")
+async def get_database_info(data: Dict[str, str]):
+ """Get additional database information like table count and size"""
+ try:
+ connection_string = data.get("connection_string")
+ if not connection_string:
+ return JSONResponse(
+ status_code=400,
+ content={"success": False, "message": "Connection string is required"}
+ )
+
+ # Use the tested connection logic first
+ if not test_connection_logic(connection_string):
+ return JSONResponse(
+ content={"success": False, "message": "Connection failed"}
+ )
+
+ conn = psycopg2.connect(connection_string)
+ try:
+ table_count = 0
+ db_size = "Unknown"
+ with conn.cursor() as cur:
+ # Get table count (excluding system, temp, and TOAST schemas)
+ cur.execute("""
+ SELECT count(*) FROM information_schema.tables
+ WHERE table_schema NOT IN ('pg_catalog', 'information_schema')
+ AND table_schema NOT LIKE 'pg_toast%'
+ AND table_schema NOT LIKE 'pg_temp%'
+ AND table_type = 'BASE TABLE'
+ """)
+ count_result = cur.fetchone()
+ table_count = count_result[0] if count_result else 0
+
+ # Get database size
+ cur.execute("SELECT pg_size_pretty(pg_database_size(current_database()))")
+ size_result = cur.fetchone()
+ db_size = size_result[0] if size_result else "Unknown"
+
+ return JSONResponse(content={
+ "success": True,
+ "table_count": table_count,
+ "database_size": db_size
+ })
+ finally:
+ conn.close()
+ except psycopg2.Error as db_err:
+ log_message(f"Failed to get database info: {str(db_err)}", "error")
+ return JSONResponse(
+ content={"success": False, "message": f"Database query error: {str(db_err)}"}
+ )
+ except Exception as e:
+ log_message(f"Failed to get database info: {str(e)}", "error")
+ return JSONResponse(
+ status_code=500,
+ content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
)
@app.post("/start-dump")
@@ -4015,20 +3792,41 @@ async def start_dump(data: Dict[str, Any], background_tasks: BackgroundTasks):
try:
source_conn = data.get("source_conn")
options = data.get("options", {})
-
if not source_conn:
return JSONResponse(
status_code=400,
content={"success": False, "message": "Source connection string is required"}
)
-
- # Stop any running process
- stop_current_process()
-
+
+ # Basic validation: Test connection before starting dump
+ if not test_connection_logic(source_conn):
+ return JSONResponse(
+ status_code=400,
+ content={"success": False, "message": "Source connection failed. Cannot start dump."}
+ )
+
+ # Stop any running process first (important!)
+ if migration_state["running"]:
+ logger.warning("Another process is running. Stopping it before starting dump.")
+ stopped = stop_current_process()
+ if not stopped:
+ logger.error("Failed to stop the existing process. Cannot start dump.")
+ return JSONResponse(
+ status_code=500, # Internal Server Error might be appropriate
+ content={"success": False, "message": "Failed to stop the currently running process."}
+ )
+ # Add a small delay to allow the process to fully terminate
+ time.sleep(0.5)
+
+
# Create dump file path
- filename = options.get("filename", "timescale_backup")
+ filename = options.get("filename", "timescale_backup").strip()
+ # Basic filename sanitization (replace spaces, avoid path traversal)
+ filename = filename.replace(" ", "_").replace("..", "").replace("/", "").replace("\\", "")
+ if not filename: filename = "timescale_backup" # Fallback if sanitization results in empty name
+
format_flag = options.get("format", "c")
-
+
# Determine file extension
extension = ".dump"
if format_flag == "p":
@@ -4037,27 +3835,75 @@ async def start_dump(data: Dict[str, Any], background_tasks: BackgroundTasks):
extension = "" # Directory format has no extension
elif format_flag == "t":
extension = ".tar"
-
- # Generate file path
- file_path = os.path.join("dumps", f"{filename}{extension}")
-
+
+ # Generate file path carefully
+ dumps_dir = Path("dumps").resolve() # Ensure absolute path
+ file_path = dumps_dir / f"{filename}{extension}"
+
+ # Prevent potential directory traversal if filename somehow still contains harmful chars
+ if not str(file_path).startswith(str(dumps_dir)):
+ logger.error(f"Invalid filename resulted in path traversal attempt: {filename}")
+ return JSONResponse(
+ status_code=400,
+ content={"success": False, "message": "Invalid filename specified."}
+ )
+
+ # Reset state before starting background task
+ with migration_lock:
+ migration_state["id"] = str(uuid.uuid4()) # New ID for new operation
+ migration_state["running"] = False # Will be set true by run_dump
+ migration_state["operation"] = "dump"
+ migration_state["start_time"] = None
+ migration_state["end_time"] = None
+ migration_state["dump_file"] = str(file_path)
+ migration_state["dump_file_size"] = 0
+ migration_state["previous_size"] = 0
+ migration_state["dump_completed"] = False
+ migration_state["restore_completed"] = False
+ migration_state["last_activity"] = time.time()
+ # Keep logs or clear them? Let's keep them for now.
+ # migration_state["log"] = []
+ migration_state["process"] = None
+ migration_state["progress"] = { # Reset progress
+ "current_table": None,
+ "tables_completed": 0,
+ "total_tables": 0,
+ "current_size_mb": 0,
+ "growth_rate_mb_per_sec": 0,
+ "estimated_time_remaining": None,
+ "percent_complete": 0
+ }
+
+
# Start dump in background
- background_tasks.add_task(run_dump, source_conn, file_path, options)
-
+ background_tasks.add_task(run_dump, source_conn, str(file_path), options)
+
# Create command preview (with redacted password)
- cmd_preview = f"{source_conn.split(':')[0]}:***@{source_conn.split('@')[1]} -F{format_flag} -v -f {file_path}"
-
+ # Redact password more carefully
+ try:
+ source_safe_preview = source_conn.replace(source_conn.split('://')[1].split(':')[1].split('@')[0], '***')
+ except:
+ source_safe_preview = "postgres://user:***@host/db" # Fallback preview
+
+ cmd_preview = f'"{source_safe_preview}" -F{format_flag} -v'
+ if options.get("compression") and options["compression"] != "default":
+ cmd_preview += f' -Z {options["compression"]}'
+ if options.get("schema"):
+ cmd_preview += f' -n "{options["schema"]}"'
+ cmd_preview += f' -f "{os.path.basename(file_path)}"'
+
+
return JSONResponse(content={
"success": True,
- "message": "Dump process started",
- "file_path": file_path,
+ "message": "Dump process initiated",
+ "file_path": str(file_path),
"command_preview": cmd_preview
})
-
except Exception as e:
log_message(f"Failed to start dump: {str(e)}", "error")
return JSONResponse(
- content={"success": False, "message": str(e)}
+ status_code=500,
+ content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
)
@app.post("/start-restore")
@@ -4067,135 +3913,236 @@ async def start_restore(data: Dict[str, Any], background_tasks: BackgroundTasks)
target_conn = data.get("target_conn")
dump_file = data.get("dump_file")
options = data.get("options", {})
-
+
if not target_conn:
return JSONResponse(
status_code=400,
content={"success": False, "message": "Target connection string is required"}
)
-
if not dump_file:
return JSONResponse(
status_code=400,
content={"success": False, "message": "Dump file is required"}
)
-
- # Stop any running process
- stop_current_process()
-
+
+ # Basic validation: Test connection before starting restore
+ if not test_connection_logic(target_conn):
+ return JSONResponse(
+ status_code=400,
+ content={"success": False, "message": "Target connection failed. Cannot start restore."}
+ )
+
+ # Validate dump file path exists and is within the dumps directory
+ dumps_dir = Path("dumps").resolve()
+ dump_file_path = Path(dump_file).resolve()
+ if not dump_file_path.exists() or not str(dump_file_path).startswith(str(dumps_dir)):
+ logger.error(f"Invalid or non-existent dump file specified: {dump_file}")
+ return JSONResponse(
+ status_code=400,
+ content={"success": False, "message": "Invalid or non-existent dump file selected."}
+ )
+
+ # Stop any running process first (important!)
+ if migration_state["running"]:
+ logger.warning("Another process is running. Stopping it before starting restore.")
+ stopped = stop_current_process()
+ if not stopped:
+ logger.error("Failed to stop the existing process. Cannot start restore.")
+ return JSONResponse(
+ status_code=500,
+ content={"success": False, "message": "Failed to stop the currently running process."}
+ )
+ time.sleep(0.5) # Allow time for termination
+
+ # Reset state before starting background task
+ with migration_lock:
+ migration_state["id"] = str(uuid.uuid4()) # New ID for new operation
+ migration_state["running"] = False # Will be set true by run_restore
+ migration_state["operation"] = "restore"
+ migration_state["start_time"] = None
+ migration_state["end_time"] = None
+ migration_state["dump_file"] = None # Not relevant for restore state itself
+ migration_state["dump_file_size"] = 0
+ migration_state["previous_size"] = 0
+ migration_state["dump_completed"] = False
+ migration_state["restore_completed"] = False
+ migration_state["last_activity"] = time.time()
+ # Keep logs
+ migration_state["process"] = None
+ migration_state["progress"] = { # Reset progress
+ "current_table": None,
+ "tables_completed": 0,
+ "total_tables": 0, # We don't easily know this for restore
+ "current_size_mb": 0,
+ "growth_rate_mb_per_sec": 0,
+ "estimated_time_remaining": None,
+ "percent_complete": 0
+ }
+
# Start restore in background
- background_tasks.add_task(run_restore, target_conn, dump_file, options)
-
+ background_tasks.add_task(run_restore, target_conn, str(dump_file_path), options)
+
# Create command preview (with redacted password)
- cmd_preview = f"-d {target_conn.split(':')[0]}:***@{target_conn.split('@')[1]} -v"
-
+ try:
+ target_safe_preview = target_conn.replace(target_conn.split('://')[1].split(':')[1].split('@')[0], '***')
+ except:
+ target_safe_preview = "postgres://user:***@host/db"
+
+ cmd_preview = f'-d "{target_safe_preview}" -v'
if options.get("no_owner", True):
cmd_preview += " --no-owner"
-
if options.get("clean", False):
cmd_preview += " --clean"
-
if options.get("single_transaction", True):
cmd_preview += " --single-transaction"
-
- cmd_preview += f" {os.path.basename(dump_file)}"
-
+ cmd_preview += f' "{os.path.basename(dump_file)}"'
+
return JSONResponse(content={
"success": True,
- "message": "Restore process started",
+ "message": "Restore process initiated",
"command_preview": cmd_preview
})
-
except Exception as e:
log_message(f"Failed to start restore: {str(e)}", "error")
return JSONResponse(
- content={"success": False, "message": str(e)}
+ status_code=500,
+ content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
)
@app.post("/stop-process")
-async def stop_process():
+async def stop_process_endpoint():
"""Stop the current database process"""
try:
stopped = stop_current_process()
-
if stopped:
return JSONResponse(content={
"success": True,
- "message": "Process stopped successfully"
+ "message": "Process stop initiated successfully" # Changed message slightly
})
else:
- return JSONResponse(content={
- "success": False,
- "message": "No running process to stop"
- })
-
+ # Check if it wasn't running in the first place
+ with migration_lock:
+ is_running = migration_state["running"]
+ if not is_running:
+ return JSONResponse(content={
+ "success": False, # Technically not an error, but no action taken
+ "message": "No process was running to stop"
+ })
+ else: # Stop was called, but failed internally
+ return JSONResponse(
+ status_code=500,
+ content={
+ "success": False,
+ "message": "Failed to stop the running process. Check logs."
+ })
except Exception as e:
- log_message(f"Failed to stop process: {str(e)}", "error")
+ log_message(f"Failed to stop process via endpoint: {str(e)}", "error")
return JSONResponse(
- content={"success": False, "message": str(e)}
+ status_code=500,
+ content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
)
@app.get("/status")
async def get_status():
"""Get the current migration status"""
- return migration_state
+ # Return a copy to avoid potential modification issues if state grows complex
+ with migration_lock:
+ state_copy = migration_state.copy()
+ # Ensure process object is not sent over JSON
+ state_copy["process"] = None
+ # Optionally limit log size sent back if it gets large
+ # MAX_LOGS_IN_STATUS = 100
+ # if len(state_copy["log"]) > MAX_LOGS_IN_STATUS:
+ # state_copy["log"] = state_copy["log"][-MAX_LOGS_IN_STATUS:]
+ return state_copy
@app.post("/clear-logs")
async def clear_logs():
"""Clear all logs"""
with migration_lock:
migration_state["log"] = []
-
- return JSONResponse(content={"success": True})
+ log_message("Logs cleared by user.", "info")
+ return JSONResponse(content={"success": True, "message": "Logs cleared"})
@app.get("/list-dumps")
async def list_dumps():
"""List available dump files"""
try:
dumps_dir = Path("dumps")
-
if not dumps_dir.exists():
dumps_dir.mkdir()
return JSONResponse(content={"success": True, "dumps": []})
-
+
dump_files = []
- for f in dumps_dir.glob("*"):
- if f.is_file():
- file_size = f.stat().st_size
- modified_time = datetime.datetime.fromtimestamp(f.stat().st_mtime)
-
- dump_files.append({
- "name": f.name,
- "path": str(f),
- "size_mb": file_size / (1024 * 1024),
- "date": modified_time.strftime("%Y-%m-%d %H:%M:%S")
- })
-
+ for f in dumps_dir.iterdir(): # Use iterdir for potentially large directories
+ # List files AND directories (for format=d)
+ if f.is_file() or f.is_dir():
+ try:
+ stat_result = f.stat()
+ file_size = stat_result.st_size
+ modified_time = datetime.datetime.fromtimestamp(stat_result.st_mtime)
+
+ # Calculate size for directories (basic sum of top-level files)
+ if f.is_dir():
+ try:
+ dir_size = sum(item.stat().st_size for item in f.iterdir() if item.is_file())
+ file_size = dir_size
+ except Exception as dir_err:
+ logger.warning(f"Could not calculate size for directory {f.name}: {dir_err}")
+ file_size = 0 # Or mark as unknown size
+
+ dump_files.append({
+ "name": f.name,
+ "path": str(f), # Send full path back
+ "size_mb": file_size / (1024 * 1024),
+ "date": modified_time.strftime("%Y-%m-%d %H:%M:%S"),
+ "is_dir": f.is_dir() # Indicate if it's a directory dump
+ })
+ except OSError as stat_err:
+ logger.error(f"Could not stat file/dir {f.name}: {stat_err}")
+ # Skip this file/dir if cannot access stats
+
+
# Sort by modified time, newest first
dump_files.sort(key=lambda x: x["date"], reverse=True)
-
return JSONResponse(content={"success": True, "dumps": dump_files})
-
except Exception as e:
log_message(f"Failed to list dumps: {str(e)}", "error")
return JSONResponse(
- content={"success": False, "message": str(e)}
+ status_code=500,
+ content={"success": False, "message": f"An unexpected error occurred: {str(e)}"}
)
-@app.get("/downloads/{file_name}")
+@app.get("/downloads/{file_name:path}") # Use path parameter to allow slashes if needed (though unlikely now)
async def download_file(file_name: str):
- """Download a dump file"""
- file_path = os.path.join("dumps", file_name)
-
- if not os.path.exists(file_path):
- raise HTTPException(status_code=404, detail="File not found")
-
- return FileResponse(
- path=file_path,
- filename=file_name,
- media_type="application/octet-stream"
- )
+ """Download a dump file (does not support directory download)"""
+ try:
+ dumps_dir = Path("dumps").resolve()
+ # Sanitize file_name to prevent path traversal
+ file_name = file_name.replace("..", "").replace("/", "").replace("\\", "")
+ file_path = dumps_dir / file_name
+
+ # Check existence and that it's within the dumps dir and is a file
+ if not file_path.exists() or not str(file_path).startswith(str(dumps_dir)):
+ raise HTTPException(status_code=404, detail="File not found or invalid path")
+ if file_path.is_dir():
+ raise HTTPException(status_code=400, detail="Directory downloads are not supported via this endpoint.")
+
+
+ return FileResponse(
+ path=str(file_path),
+ filename=file_name, # Suggest original (sanitized) name to browser
+ media_type="application/octet-stream" # Generic binary type
+ )
+ except HTTPException:
+ raise # Re-raise HTTPException
+ except Exception as e:
+ logger.error(f"Error preparing file download for {file_name}: {e}")
+ raise HTTPException(status_code=500, detail="Could not process file download.")
+
if __name__ == "__main__":
import uvicorn
- uvicorn.run(app, host="0.0.0.0", port=7860)
\ No newline at end of file
+ # Use reload=True for development, but turn off for production
+ uvicorn.run("main:app", host="0.0.0.0", port=7860, reload=True)
+ # For production: uvicorn.run(app, host="0.0.0.0", port=7860)
\ No newline at end of file