Upload 12 files

app/api/routes/auth_router.py

@@ -3,10 +3,18 @@ from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
 from passlib.context import CryptContext
 from jose import JWTError, jwt
 from datetime import datetime, timedelta
-from typing import Optional
+from typing import Optional, Dict, Any, List
 import os
+import time
+import logging
 from pydantic import BaseModel, EmailStr
 
+# Import the HTTP API database utility
+from app.utils import db_http
+
+# Configure logging
+logger = logging.getLogger("auth-server")
+
 router = APIRouter()
 
 # Password hashing
@@ -88,58 +96,71 @@ async def get_current_user(request: Request, token: str = Depends(oauth2_scheme)
     except JWTError:
         raise credentials_exception
 
-    query = "SELECT * FROM users WHERE id = ?"
-    user = request.app.db_conn.execute(query, (token_data.user_id,)).fetchone()
-    if user is None:
+    # Use the HTTP API utility to get the user
+    operation_id = f"get_user_{int(time.time())}"
+
+    try:
+        # First try with the app's db_conn if it exists and is not HTTP API
+        if hasattr(request.app, "db_conn") and getattr(request.app, "db_type", "") != "http-api":
+            logger.info(f"[{operation_id}] Using app.db_conn to get user")
+            query = "SELECT * FROM users WHERE id = ?"
+            user = request.app.db_conn.execute(query, (token_data.user_id,)).fetchone()
+            if user is not None:
+                return user
+
+        # If that fails or is not available, use the HTTP API
+        logger.info(f"[{operation_id}] Using HTTP API to get user")
+        user = db_http.get_record_by_id("users", token_data.user_id, operation_id=operation_id)
+
+        if user is None:
+            logger.error(f"[{operation_id}] User not found with ID: {token_data.user_id}")
+            raise credentials_exception
+
+        # If user is a dict (from HTTP API), convert it to a tuple-like structure
+        # to maintain compatibility with existing code
+        if isinstance(user, dict):
+            user_tuple = (
+                user.get("id"),
+                user.get("email"),
+                user.get("hashed_password"),
+                user.get("created_at"),
+                user.get("last_login"),
+                user.get("is_admin", 0)
+            )
+            return user_tuple
+
+        return user
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error getting user: {str(e)}")
         raise credentials_exception
-    return user
 
 # Routes
 @router.post("/register", response_model=UserResponse)
 async def register(request: Request, user: UserCreate):
-    import logging
-    import time
-    import sys
-    logger = logging.getLogger("auth-server")
-
-    # Log the database connection type and version
-    db_type = getattr(request.app, "db_type", "unknown")
-    logger.info(f"Database connection type: {db_type}")
-
-    # Log database connection details
-    if hasattr(request.app, "db_conn"):
-        logger.info(f"Database connection object type: {type(request.app.db_conn).__name__}")
-        if hasattr(request.app, "last_successful_connection_method"):
-            logger.info(f"Last successful connection method: {request.app.last_successful_connection_method}")
-
     # Generate a unique identifier for this registration attempt
     registration_id = f"reg_{int(time.time())}"
     logger.info(f"[{registration_id}] Starting registration for email: {user.email}")
 
     try:
-        # Step 1: Check if user already exists - with retry
-        for attempt in range(3):
-            try:
-                logger.info(f"[{registration_id}] Checking if user already exists (attempt {attempt+1})")
-                query = "SELECT id FROM users WHERE email = ?"
-                existing_user = request.app.db_conn.execute(query, (user.email,)).fetchone()
-                logger.info(f"[{registration_id}] Check result: {existing_user}")
-
-                if existing_user:
-                    logger.warning(f"[{registration_id}] User with email {user.email} already exists")
-                    raise HTTPException(
-                        status_code=status.HTTP_400_BAD_REQUEST,
-                        detail="Email already registered"
-                    )
-                logger.info(f"[{registration_id}] User does not exist, proceeding with registration")
-                break
-            except Exception as e:
-                if "HTTPException" in str(type(e)):
-                    raise
-                logger.error(f"[{registration_id}] Error checking if user exists (attempt {attempt+1}): {str(e)}")
-                if attempt == 2:  # Last attempt
-                    raise
-                time.sleep(1)  # Wait before retrying
+        # Step 1: Check if user already exists using HTTP API
+        logger.info(f"[{registration_id}] Checking if user already exists using HTTP API")
+
+        # Check if user exists
+        user_exists = db_http.record_exists(
+            "users",
+            "email = ?",
+            [{"type": "text", "value": user.email}],
+            operation_id=registration_id
+        )
+
+        if user_exists:
+            logger.warning(f"[{registration_id}] User with email {user.email} already exists")
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Email already registered"
+            )
+
+        logger.info(f"[{registration_id}] User does not exist, proceeding with registration")
 
         # Step 2: Hash the password
         logger.info(f"[{registration_id}] Hashing password")
@@ -153,183 +174,70 @@ async def register(request: Request, user: UserCreate):
                 detail=f"Error hashing password: {str(e)}"
             )
 
-        # Step 3: Insert the new user - with retry
-        user_id = None
-        for attempt in range(3):
-            try:
-                logger.info(f"[{registration_id}] Inserting new user (attempt {attempt+1})")
-
-                # Different approaches based on connection type
-                if db_type == "http-api":
-                    # First, get the highest ID to simulate auto-increment
-                    max_id_result = request.app.db_conn.execute("SELECT MAX(id) FROM users").fetchone()
-                    logger.info(f"[{registration_id}] Max ID result: {max_id_result}")
-
-                    next_id = 1
-                    if max_id_result and max_id_result[0] is not None:
-                        next_id = int(max_id_result[0]) + 1
-
-                    # Insert with explicit ID
-                    insert_query = "INSERT INTO users (id, email, hashed_password) VALUES (?, ?, ?)"
-                    request.app.db_conn.execute(insert_query, (next_id, user.email, hashed_password))
-                    user_id = next_id
-                    logger.info(f"[{registration_id}] Inserted user with explicit ID: {user_id}")
-                elif db_type == "libsql-experimental":
-                    # For libsql-experimental, we need to use a transaction
-                    logger.info(f"[{registration_id}] Using libsql-experimental approach")
-
-                    # Start a transaction
-                    conn = request.app.db_conn
+        # Step 3: Insert the new user using HTTP API
+        logger.info(f"[{registration_id}] Inserting new user using HTTP API")
 
-                    # Insert the user
-                    insert_query = "INSERT INTO users (email, hashed_password) VALUES (?, ?)"
-                    cursor = conn.execute(insert_query, (user.email, hashed_password))
+        # Prepare user data
+        user_data = {
+            "email": user.email,
+            "hashed_password": hashed_password,
+            "is_admin": 0  # Default to non-admin
+        }
 
-                    # Try to get the last inserted ID
-                    try:
-                        user_id = cursor.lastrowid
-                        logger.info(f"[{registration_id}] Got lastrowid: {user_id}")
-                    except Exception as e:
-                        logger.warning(f"[{registration_id}] Could not get lastrowid: {str(e)}")
-
-                        # If lastrowid is not available, try to get the ID using a query
-                        try:
-                            # Get the ID using the email (which should be unique)
-                            id_query = "SELECT id FROM users WHERE email = ? ORDER BY id DESC LIMIT 1"
-                            id_result = conn.execute(id_query, (user.email,)).fetchone()
-                            if id_result:
-                                user_id = id_result[0]
-                                logger.info(f"[{registration_id}] Got ID from query: {user_id}")
-                        except Exception as e2:
-                            logger.error(f"[{registration_id}] Error getting ID from query: {str(e2)}")
-                else:
-                    # Standard approach for other connection types
-                    logger.info(f"[{registration_id}] Using standard approach")
-                    insert_query = "INSERT INTO users (email, hashed_password) VALUES (?, ?)"
-                    cursor = request.app.db_conn.execute(insert_query, (user.email, hashed_password))
+        # Insert the user
+        user_id = db_http.insert_record("users", user_data, operation_id=registration_id)
 
-                    try:
-                        user_id = cursor.lastrowid
-                        logger.info(f"[{registration_id}] Got lastrowid: {user_id}")
-                    except Exception as e:
-                        logger.warning(f"[{registration_id}] Could not get lastrowid: {str(e)}")
+        if not user_id:
+            logger.error(f"[{registration_id}] Failed to insert user")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Failed to create user account"
+            )
 
-                # Commit the transaction
-                logger.info(f"[{registration_id}] Committing transaction")
-                request.app.db_conn.commit()
-                logger.info(f"[{registration_id}] Transaction committed")
+        logger.info(f"[{registration_id}] User inserted successfully with ID: {user_id}")
 
-                # Verify the insert by checking if the user exists
-                verify_query = "SELECT id FROM users WHERE email = ?"
-                verify_result = request.app.db_conn.execute(verify_query, (user.email,)).fetchone()
-                logger.info(f"[{registration_id}] Verification result: {verify_result}")
+        # Step 4: Retrieve the full user record
+        logger.info(f"[{registration_id}] Retrieving user record")
 
-                if verify_result:
-                    if not user_id:
-                        user_id = verify_result[0]
-                    logger.info(f"[{registration_id}] User verified with ID: {user_id}")
-                    break
-                else:
-                    logger.error(f"[{registration_id}] User not found after insert!")
-                    if attempt == 2:  # Last attempt
-                        raise Exception("User was not found after insert")
-                    time.sleep(1)  # Wait before retrying
-            except Exception as e:
-                logger.error(f"[{registration_id}] Error inserting user (attempt {attempt+1}): {str(e)}")
-                if attempt == 2:  # Last attempt
-                    raise
-                time.sleep(1)  # Wait before retrying
-
-        # Step 4: Retrieve the full user record - with retry
-        new_user = None
-        for attempt in range(3):
-            try:
-                # Try by ID if we have it
-                if user_id:
-                    logger.info(f"[{registration_id}] Retrieving user by ID: {user_id} (attempt {attempt+1})")
-                    new_user = request.app.db_conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
-                    logger.info(f"[{registration_id}] Result by ID: {new_user}")
-
-                # If not found by ID, try by email
-                if not new_user:
-                    logger.info(f"[{registration_id}] Retrieving user by email: {user.email} (attempt {attempt+1})")
-                    new_user = request.app.db_conn.execute("SELECT * FROM users WHERE email = ?", (user.email,)).fetchone()
-                    logger.info(f"[{registration_id}] Result by email: {new_user}")
-
-                if new_user:
-                    logger.info(f"[{registration_id}] User retrieved successfully")
-                    break
-                else:
-                    logger.warning(f"[{registration_id}] User not found (attempt {attempt+1})")
-
-                    # Dump all users for debugging
-                    try:
-                        all_users = request.app.db_conn.execute("SELECT id, email FROM users").fetchall()
-                        logger.info(f"[{registration_id}] All users in database: {all_users}")
-                    except Exception as e:
-                        logger.error(f"[{registration_id}] Error retrieving all users: {str(e)}")
-
-                    if attempt == 2:  # Last attempt
-                        # Create a minimal response with what we know
-                        return {
-                            "id": user_id or 0,
-                            "email": user.email,
-                            "is_admin": False,
-                            "created_at": ""
-                        }
-                    time.sleep(1)  # Wait before retrying
-            except Exception as e:
-                logger.error(f"[{registration_id}] Error retrieving user (attempt {attempt+1}): {str(e)}")
-                if attempt == 2:  # Last attempt
-                    # Create a minimal response with what we know
-                    return {
-                        "id": user_id or 0,
-                        "email": user.email,
-                        "is_admin": False,
-                        "created_at": ""
-                    }
-                time.sleep(1)  # Wait before retrying
+        new_user = db_http.get_record_by_id("users", user_id, operation_id=registration_id)
 
-        # Step 5: Construct the response
         if not new_user:
-            logger.error(f"[{registration_id}] Failed to retrieve user after multiple attempts")
-            # Create a minimal response with what we know
-            return {
-                "id": user_id or 0,
-                "email": user.email,
-                "is_admin": False,
-                "created_at": ""
-            }
-
-        # Log the user data for debugging
-        logger.info(f"[{registration_id}] User data: {new_user}")
-
-        # Handle different database return formats
-        if isinstance(new_user, dict):
-            # HTTP API might return a dict
-            return {
-                "id": new_user.get("id", 0),
-                "email": new_user.get("email", user.email),
-                "is_admin": bool(new_user.get("is_admin", 0)),
-                "created_at": str(new_user.get("created_at", ""))
-            }
-        elif isinstance(new_user, (list, tuple)):
-            # Standard cursor returns a tuple
-            return {
-                "id": new_user[0] if len(new_user) > 0 else 0,
-                "email": new_user[1] if len(new_user) > 1 else user.email,
-                "is_admin": bool(new_user[5] if len(new_user) > 5 and new_user[5] is not None else 0),
-                "created_at": str(new_user[3] if len(new_user) > 3 and new_user[3] is not None else "")
-            }
-        else:
-            # Fallback for unexpected return type
-            logger.error(f"[{registration_id}] Unexpected user data type: {type(new_user)}")
-            return {
-                "id": user_id or 0,
-                "email": user.email,
-                "is_admin": False,
-                "created_at": ""
-            }
+            logger.warning(f"[{registration_id}] User not found after insert, trying by email")
+
+            # Try to get by email as fallback
+            users = db_http.select_records(
+                "users",
+                condition="email = ?",
+                condition_params=[{"type": "text", "value": user.email}],
+                limit=1,
+                operation_id=registration_id
+            )
+
+            if users:
+                new_user = users[0]
+                user_id = new_user.get("id")
+                logger.info(f"[{registration_id}] Found user by email with ID: {user_id}")
+            else:
+                logger.error(f"[{registration_id}] User not found by email either")
+
+                # Return minimal response with what we know
+                return {
+                    "id": user_id or 0,
+                    "email": user.email,
+                    "is_admin": False,
+                    "created_at": datetime.now().isoformat()
+                }
+
+        # Step 5: Construct the response
+        logger.info(f"[{registration_id}] Registration successful for user ID: {user_id}")
+
+        return {
+            "id": new_user.get("id", user_id),
+            "email": new_user.get("email", user.email),
+            "is_admin": bool(new_user.get("is_admin", 0)),
+            "created_at": str(new_user.get("created_at", datetime.now().isoformat()))
+        }
+
     except Exception as e:
         logger.error(f"[{registration_id}] Registration failed: {str(e)}")
         if isinstance(e, HTTPException):
@@ -341,104 +249,178 @@ async def register(request: Request, user: UserCreate):
 
 @router.post("/login", response_model=Token)
 async def login(
-    request: Request,
     form_data: OAuth2PasswordRequestForm = Depends()
 ):
-    # Find the user
-    query = "SELECT * FROM users WHERE email = ?"
-    user = request.app.db_conn.execute(query, (form_data.username,)).fetchone()
+    operation_id = f"login_{int(time.time())}"
+    logger.info(f"[{operation_id}] Login attempt for email: {form_data.username}")
 
-    if not user or not verify_password(form_data.password, user[2]):
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Incorrect email or password",
-            headers={"WWW-Authenticate": "Bearer"},
+    try:
+        # Find the user using HTTP API
+        users = db_http.select_records(
+            "users",
+            condition="email = ?",
+            condition_params=[{"type": "text", "value": form_data.username}],
+            limit=1,
+            operation_id=operation_id
         )
 
-    # Update last login
-    update_query = "UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = ?"
-    request.app.db_conn.execute(update_query, (user[0],))
-    request.app.db_conn.commit()
+        if not users:
+            logger.warning(f"[{operation_id}] User not found with email: {form_data.username}")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Incorrect email or password",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
 
-    # Create tokens
-    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = create_access_token(
-        data={"sub": user[0]},
-        expires_delta=access_token_expires
-    )
+        user = users[0]
 
-    refresh_token_expires = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
-    refresh_token = create_refresh_token(
-        data={"sub": user[0]},
-        expires_delta=refresh_token_expires
-    )
+        # Verify password
+        if not verify_password(form_data.password, user.get("hashed_password")):
+            logger.warning(f"[{operation_id}] Invalid password for user: {form_data.username}")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Incorrect email or password",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
 
-    return {
-        "access_token": access_token,
-        "refresh_token": refresh_token,
-        "token_type": "bearer"
-    }
+        logger.info(f"[{operation_id}] Password verified for user ID: {user.get('id')}")
+
+        # Update last login
+        update_data = {
+            "last_login": datetime.now().isoformat()
+        }
+
+        db_http.update_record(
+            "users",
+            update_data,
+            "id = ?",
+            [{"type": "integer", "value": str(user.get("id"))}],
+            operation_id=operation_id
+        )
+
+        logger.info(f"[{operation_id}] Last login updated for user ID: {user.get('id')}")
+
+        # Create tokens
+        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+        access_token = create_access_token(
+            data={"sub": user.get("id")},
+            expires_delta=access_token_expires
+        )
+
+        refresh_token_expires = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
+        refresh_token = create_refresh_token(
+            data={"sub": user.get("id")},
+            expires_delta=refresh_token_expires
+        )
+
+        logger.info(f"[{operation_id}] Login successful for user ID: {user.get('id')}")
+
+        return {
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "token_type": "bearer"
+        }
+    except Exception as e:
+        if isinstance(e, HTTPException):
+            raise
+        logger.error(f"[{operation_id}] Login failed: {str(e)}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Login failed: {str(e)}"
+        )
 
 @router.post("/refresh", response_model=Token)
-async def refresh_token(request: Request, refresh_token: str):
+async def refresh_token(refresh_token_str: str):
+    operation_id = f"refresh_{int(time.time())}"
+    logger.info(f"[{operation_id}] Token refresh attempt")
+
     try:
-        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
-        user_id: int = payload.get("sub")
-        if user_id is None:
+        # Decode and validate the refresh token
+        try:
+            payload = jwt.decode(refresh_token_str, SECRET_KEY, algorithms=[ALGORITHM])
+            user_id: int = payload.get("sub")
+            if user_id is None:
+                logger.warning(f"[{operation_id}] Missing user ID in token payload")
+                raise HTTPException(
+                    status_code=status.HTTP_401_UNAUTHORIZED,
+                    detail="Invalid refresh token",
+                    headers={"WWW-Authenticate": "Bearer"},
+                )
+        except JWTError as e:
+            logger.warning(f"[{operation_id}] JWT decode error: {str(e)}")
             raise HTTPException(
                 status_code=status.HTTP_401_UNAUTHORIZED,
                 detail="Invalid refresh token",
                 headers={"WWW-Authenticate": "Bearer"},
             )
-    except JWTError:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid refresh token",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
 
-    # Check if user exists
-    query = "SELECT * FROM users WHERE id = ?"
-    user = request.app.db_conn.execute(query, (user_id,)).fetchone()
-    if user is None:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="User not found",
-            headers={"WWW-Authenticate": "Bearer"},
+        logger.info(f"[{operation_id}] Token decoded successfully for user ID: {user_id}")
+
+        # Check if user exists using HTTP API
+        user = db_http.get_record_by_id("users", user_id, operation_id=operation_id)
+
+        if not user:
+            logger.warning(f"[{operation_id}] User not found with ID: {user_id}")
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="User not found",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        logger.info(f"[{operation_id}] User found with ID: {user_id}")
+
+        # Create new tokens
+        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+        access_token = create_access_token(
+            data={"sub": user_id},
+            expires_delta=access_token_expires
         )
 
-    # Create new tokens
-    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
-    access_token = create_access_token(
-        data={"sub": user_id},
-        expires_delta=access_token_expires
-    )
+        refresh_token_expires = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
+        new_refresh_token = create_refresh_token(
+            data={"sub": user_id},
+            expires_delta=refresh_token_expires
+        )
 
-    refresh_token_expires = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
-    new_refresh_token = create_refresh_token(
-        data={"sub": user_id},
-        expires_delta=refresh_token_expires
-    )
+        logger.info(f"[{operation_id}] Token refresh successful for user ID: {user_id}")
 
-    return {
-        "access_token": access_token,
-        "refresh_token": new_refresh_token,
-        "token_type": "bearer"
-    }
+        return {
+            "access_token": access_token,
+            "refresh_token": new_refresh_token,
+            "token_type": "bearer"
+        }
+    except Exception as e:
+        if isinstance(e, HTTPException):
+            raise
+        logger.error(f"[{operation_id}] Token refresh failed: {str(e)}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Token refresh failed: {str(e)}"
+        )
 
 @router.get("/me", response_model=UserResponse)
-async def get_current_user_info(request: Request, current_user = Depends(get_current_user)):
-    return {
-        "id": current_user[0],
-        "email": current_user[1],
-        "is_admin": current_user[5],
-        "created_at": current_user[3]
-    }
+async def get_current_user_info(current_user = Depends(get_current_user)):
+    # Handle both tuple (from db_conn) and dict (from HTTP API) formats
+    if isinstance(current_user, dict):
+        return {
+            "id": current_user.get("id"),
+            "email": current_user.get("email"),
+            "is_admin": bool(current_user.get("is_admin", 0)),
+            "created_at": str(current_user.get("created_at", ""))
+        }
+    else:
+        # Assume tuple format from the original implementation
+        return {
+            "id": current_user[0],
+            "email": current_user[1],
+            "is_admin": bool(current_user[5] if current_user[5] is not None else 0),
+            "created_at": str(current_user[3] if current_user[3] is not None else "")
+        }
 
 @router.get("/test-db")
 async def test_database(request: Request):
     """
-    Test endpoint to verify database operations.
+    Test endpoint to verify database operations using the original method.
     This is for debugging purposes only.
     """
     import logging
@@ -657,3 +639,152 @@ async def test_database(request: Request):
             "error": str(e)
         })
         return results
+
+@router.get("/test-db-http")
+async def test_database_http():
+    """
+    Test endpoint to verify database operations using the HTTP API method.
+    This is for debugging purposes only.
+    """
+    operation_id = f"test_db_http_{int(time.time())}"
+    logger.info(f"[{operation_id}] Starting HTTP API database test")
+
+    results = {
+        "connection_type": "http-api",
+        "operations": []
+    }
+
+    try:
+        # Test 1: Simple SELECT
+        logger.info(f"[{operation_id}] Test 1: Simple SELECT")
+        result = db_http.execute_query("SELECT 1 as test", operation_id=f"{operation_id}_1")
+        results["operations"].append({
+            "name": "Simple SELECT",
+            "success": True,
+            "result": str(result)
+        })
+        logger.info(f"[{operation_id}] Test 1 result: {result}")
+
+        # Test 2: Create a temporary table
+        logger.info(f"[{operation_id}] Test 2: Create a temporary table")
+        db_http.create_table_if_not_exists(
+            "test_table_http",
+            """
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            value TEXT NOT NULL,
+            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+            """,
+            operation_id=f"{operation_id}_2"
+        )
+        results["operations"].append({
+            "name": "Create temporary table",
+            "success": True
+        })
+        logger.info(f"[{operation_id}] Test 2 completed successfully")
+
+        # Test 3: Insert into the temporary table
+        logger.info(f"[{operation_id}] Test 3: Insert into the temporary table")
+        test_value = f"test_value_http_{int(time.time())}"
+
+        insert_id = db_http.insert_record(
+            "test_table_http",
+            {"value": test_value},
+            operation_id=f"{operation_id}_3"
+        )
+
+        results["operations"].append({
+            "name": "Insert into temporary table",
+            "success": insert_id is not None,
+            "insert_id": insert_id
+        })
+        logger.info(f"[{operation_id}] Test 3 completed successfully. Insert ID: {insert_id}")
+
+        # Test 4: Select from the temporary table
+        logger.info(f"[{operation_id}] Test 4: Select from the temporary table")
+        records = db_http.select_records(
+            "test_table_http",
+            condition="value = ?",
+            condition_params=[{"type": "text", "value": test_value}],
+            limit=1,
+            operation_id=f"{operation_id}_4"
+        )
+
+        results["operations"].append({
+            "name": "Select from temporary table",
+            "success": len(records) > 0,
+            "result": str(records[0]) if records else None
+        })
+        logger.info(f"[{operation_id}] Test 4 result: {records}")
+
+        # Test 5: Update the record
+        logger.info(f"[{operation_id}] Test 5: Update the record")
+        updated_value = f"{test_value}_updated"
+
+        success = db_http.update_record(
+            "test_table_http",
+            {"value": updated_value},
+            "id = ?",
+            [{"type": "integer", "value": str(insert_id)}],
+            operation_id=f"{operation_id}_5"
+        )
+
+        results["operations"].append({
+            "name": "Update record",
+            "success": success
+        })
+        logger.info(f"[{operation_id}] Test 5 completed successfully")
+
+        # Test 6: Verify the update
+        logger.info(f"[{operation_id}] Test 6: Verify the update")
+        updated_records = db_http.select_records(
+            "test_table_http",
+            condition="id = ?",
+            condition_params=[{"type": "integer", "value": str(insert_id)}],
+            limit=1,
+            operation_id=f"{operation_id}_6"
+        )
+
+        results["operations"].append({
+            "name": "Verify update",
+            "success": len(updated_records) > 0 and updated_records[0].get("value") == updated_value,
+            "result": str(updated_records[0]) if updated_records else None
+        })
+        logger.info(f"[{operation_id}] Test 6 result: {updated_records}")
+
+        # Test 7: Count records
+        logger.info(f"[{operation_id}] Test 7: Count records")
+        count = db_http.count_records("test_table_http", operation_id=f"{operation_id}_7")
+
+        results["operations"].append({
+            "name": "Count records",
+            "success": count > 0,
+            "count": count
+        })
+        logger.info(f"[{operation_id}] Test 7 result: {count}")
+
+        # Test 8: List all records
+        logger.info(f"[{operation_id}] Test 8: List all records")
+        all_records = db_http.select_records(
+            "test_table_http",
+            limit=10,
+            operation_id=f"{operation_id}_8"
+        )
+
+        results["operations"].append({
+            "name": "List all records",
+            "success": True,
+            "count": len(all_records),
+            "records": all_records
+        })
+        logger.info(f"[{operation_id}] Test 8 result: {len(all_records)} records")
+
+        logger.info(f"[{operation_id}] HTTP API database test completed successfully")
+        return results
+    except Exception as e:
+        logger.error(f"[{operation_id}] HTTP API database test failed: {str(e)}")
+        results["operations"].append({
+            "name": "Error during tests",
+            "success": False,
+            "error": str(e)
+        })
+        return results

app/api/routes/projects_router.py

@@ -1,8 +1,14 @@
-from fastapi import APIRouter, HTTPException, Depends, Request, status
+from fastapi import APIRouter, HTTPException, Depends, status
 from typing import List, Optional
 from pydantic import BaseModel
 import json
+import time
+import logging
 from .auth_router import get_current_user
+from app.utils import db_http
+
+# Configure logging
+logger = logging.getLogger("auth-server")
 
 router = APIRouter()
 
@@ -28,54 +34,106 @@ class ProjectResponse(ProjectBase):
 # Routes
 @router.post("/", response_model=ProjectResponse)
 async def create_project(
-    request: Request,
     project: ProjectCreate,
     current_user = Depends(get_current_user)
 ):
+    operation_id = f"create_project_{int(time.time())}"
+    logger.info(f"[{operation_id}] Creating new project: {project.title}")
+
     try:
+        # Get user ID based on the type of current_user
+        if isinstance(current_user, dict):
+            user_id = current_user.get("id")
+        else:
+            user_id = current_user[0]
+
+        logger.info(f"[{operation_id}] User ID: {user_id}")
+
         # Validate GeoJSON if provided
         if project.geojson:
             try:
                 geojson_data = json.loads(project.geojson)
                 # Basic validation
                 if not isinstance(geojson_data, dict) or "type" not in geojson_data:
+                    logger.warning(f"[{operation_id}] Invalid GeoJSON format")
                     raise ValueError("Invalid GeoJSON format")
             except json.JSONDecodeError:
+                logger.warning(f"[{operation_id}] Invalid GeoJSON format (JSON decode error)")
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail="Invalid GeoJSON format"
                 )
-        
-        # Insert the new project
-        insert_query = """
-        INSERT INTO projects (owner_id, title, description, geojson)
-        VALUES (?, ?, ?, ?)
-        """
-        request.app.db_conn.execute(
-            insert_query, 
-            (current_user[0], project.title, project.description, project.geojson)
-        )
-        request.app.db_conn.commit()
-        
+
+        logger.info(f"[{operation_id}] GeoJSON validation passed")
+
+        # Prepare project data
+        project_data = {
+            "owner_id": user_id,
+            "title": project.title,
+            "description": project.description,
+            "geojson": project.geojson,
+            "storage_bucket": "default"  # Default storage bucket
+        }
+
+        # Insert the new project using HTTP API
+        logger.info(f"[{operation_id}] Inserting new project")
+        project_id = db_http.insert_record("projects", project_data, operation_id=operation_id)
+
+        if not project_id:
+            logger.error(f"[{operation_id}] Failed to insert project")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Failed to create project"
+            )
+
+        logger.info(f"[{operation_id}] Project inserted with ID: {project_id}")
+
         # Get the newly created project
-        new_project = request.app.db_conn.execute(
-            "SELECT * FROM projects WHERE owner_id = ? ORDER BY id DESC LIMIT 1", 
-            (current_user[0],)
-        ).fetchone()
-        
+        new_project = db_http.get_record_by_id("projects", project_id, operation_id=operation_id)
+
+        if not new_project:
+            logger.warning(f"[{operation_id}] Project not found after insert, trying by owner and title")
+
+            # Try to get by owner and title as fallback
+            projects = db_http.select_records(
+                "projects",
+                condition="owner_id = ? AND title = ?",
+                condition_params=[
+                    {"type": "integer", "value": str(user_id)},
+                    {"type": "text", "value": project.title}
+                ],
+                order_by="id DESC",
+                limit=1,
+                operation_id=operation_id
+            )
+
+            if projects:
+                new_project = projects[0]
+                project_id = new_project.get("id")
+                logger.info(f"[{operation_id}] Found project by owner and title with ID: {project_id}")
+            else:
+                logger.error(f"[{operation_id}] Project not found after insert")
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail="Project was created but could not be retrieved"
+                )
+
+        logger.info(f"[{operation_id}] Project created successfully with ID: {project_id}")
+
         return {
-            "id": new_project[0],
-            "owner_id": new_project[1],
-            "title": new_project[2],
-            "description": new_project[3],
-            "geojson": new_project[4],
-            "storage_bucket": new_project[5],
-            "created_at": new_project[6],
-            "updated_at": new_project[7]
+            "id": new_project.get("id"),
+            "owner_id": new_project.get("owner_id"),
+            "title": new_project.get("title"),
+            "description": new_project.get("description"),
+            "geojson": new_project.get("geojson"),
+            "storage_bucket": new_project.get("storage_bucket", "default"),
+            "created_at": new_project.get("created_at", ""),
+            "updated_at": new_project.get("updated_at", "")
         }
     except HTTPException:
         raise
     except Exception as e:
+        logger.error(f"[{operation_id}] Error creating project: {str(e)}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=str(e)
@@ -83,38 +141,53 @@ async def create_project(
 
 @router.get("/", response_model=List[ProjectResponse])
 async def get_projects(
-    request: Request,
     current_user = Depends(get_current_user),
     skip: int = 0,
     limit: int = 100
 ):
+    operation_id = f"get_projects_{int(time.time())}"
+    logger.info(f"[{operation_id}] Getting projects (skip={skip}, limit={limit})")
+
     try:
-        query = """
-        SELECT * FROM projects 
-        WHERE owner_id = ? 
-        ORDER BY updated_at DESC
-        LIMIT ? OFFSET ?
-        """
-        projects = request.app.db_conn.execute(
-            query, 
-            (current_user[0], limit, skip)
-        ).fetchall()
-        
+        # Get user ID based on the type of current_user
+        if isinstance(current_user, dict):
+            user_id = current_user.get("id")
+        else:
+            user_id = current_user[0]
+
+        logger.info(f"[{operation_id}] User ID: {user_id}")
+
+        # Get projects using HTTP API
+        projects = db_http.select_records(
+            "projects",
+            condition="owner_id = ?",
+            condition_params=[{"type": "integer", "value": str(user_id)}],
+            order_by="updated_at DESC",
+            limit=limit,
+            offset=skip,
+            operation_id=operation_id
+        )
+
+        logger.info(f"[{operation_id}] Found {len(projects)} projects")
+
+        # Projects are already in dictionary format from the HTTP API
+        # Just need to ensure all required fields are present
         result = []
         for project in projects:
             result.append({
-                "id": project[0],
-                "owner_id": project[1],
-                "title": project[2],
-                "description": project[3],
-                "geojson": project[4],
-                "storage_bucket": project[5],
-                "created_at": project[6],
-                "updated_at": project[7]
+                "id": project.get("id"),
+                "owner_id": project.get("owner_id"),
+                "title": project.get("title"),
+                "description": project.get("description"),
+                "geojson": project.get("geojson"),
+                "storage_bucket": project.get("storage_bucket", "default"),
+                "created_at": project.get("created_at", ""),
+                "updated_at": project.get("updated_at", "")
             })
-        
+
         return result
     except Exception as e:
+        logger.error(f"[{operation_id}] Error getting projects: {str(e)}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=str(e)
@@ -122,36 +195,57 @@ async def get_projects(
 
 @router.get("/{project_id}", response_model=ProjectResponse)
 async def get_project(
-    request: Request,
     project_id: int,
     current_user = Depends(get_current_user)
 ):
+    operation_id = f"get_project_{int(time.time())}"
+    logger.info(f"[{operation_id}] Getting project with ID: {project_id}")
+
     try:
-        query = "SELECT * FROM projects WHERE id = ? AND owner_id = ?"
-        project = request.app.db_conn.execute(
-            query, 
-            (project_id, current_user[0])
-        ).fetchone()
-        
-        if not project:
+        # Get user ID based on the type of current_user
+        if isinstance(current_user, dict):
+            user_id = current_user.get("id")
+        else:
+            user_id = current_user[0]
+
+        logger.info(f"[{operation_id}] User ID: {user_id}")
+
+        # Get project using HTTP API
+        projects = db_http.select_records(
+            "projects",
+            condition="id = ? AND owner_id = ?",
+            condition_params=[
+                {"type": "integer", "value": str(project_id)},
+                {"type": "integer", "value": str(user_id)}
+            ],
+            limit=1,
+            operation_id=operation_id
+        )
+
+        if not projects:
+            logger.warning(f"[{operation_id}] Project not found with ID: {project_id}")
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
                 detail="Project not found"
             )
-        
+
+        project = projects[0]
+        logger.info(f"[{operation_id}] Found project: {project.get('title')}")
+
         return {
-            "id": project[0],
-            "owner_id": project[1],
-            "title": project[2],
-            "description": project[3],
-            "geojson": project[4],
-            "storage_bucket": project[5],
-            "created_at": project[6],
-            "updated_at": project[7]
+            "id": project.get("id"),
+            "owner_id": project.get("owner_id"),
+            "title": project.get("title"),
+            "description": project.get("description"),
+            "geojson": project.get("geojson"),
+            "storage_bucket": project.get("storage_bucket", "default"),
+            "created_at": project.get("created_at", ""),
+            "updated_at": project.get("updated_at", "")
         }
     except HTTPException:
         raise
     except Exception as e:
+        logger.error(f"[{operation_id}] Error getting project: {str(e)}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=str(e)
@@ -159,98 +253,132 @@ async def get_project(
 
 @router.patch("/{project_id}", response_model=ProjectResponse)
 async def update_project(
-    request: Request,
     project_id: int,
     project_update: ProjectUpdate,
     current_user = Depends(get_current_user)
 ):
+    operation_id = f"update_project_{int(time.time())}"
+    logger.info(f"[{operation_id}] Updating project with ID: {project_id}")
+
     try:
+        # Get user ID based on the type of current_user
+        if isinstance(current_user, dict):
+            user_id = current_user.get("id")
+        else:
+            user_id = current_user[0]
+
+        logger.info(f"[{operation_id}] User ID: {user_id}")
+
         # Check if project exists and belongs to user
-        query = "SELECT * FROM projects WHERE id = ? AND owner_id = ?"
-        existing_project = request.app.db_conn.execute(
-            query, 
-            (project_id, current_user[0])
-        ).fetchone()
-        
-        if not existing_project:
+        projects = db_http.select_records(
+            "projects",
+            condition="id = ? AND owner_id = ?",
+            condition_params=[
+                {"type": "integer", "value": str(project_id)},
+                {"type": "integer", "value": str(user_id)}
+            ],
+            limit=1,
+            operation_id=operation_id
+        )
+
+        if not projects:
+            logger.warning(f"[{operation_id}] Project not found with ID: {project_id}")
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
                 detail="Project not found"
             )
-        
-        # Prepare update fields
-        update_fields = []
-        params = []
-        
+
+        existing_project = projects[0]
+        logger.info(f"[{operation_id}] Found project: {existing_project.get('title')}")
+
+        # Prepare update data
+        update_data = {}
+
         if project_update.title is not None:
-            update_fields.append("title = ?")
-            params.append(project_update.title)
-        
+            update_data["title"] = project_update.title
+
         if project_update.description is not None:
-            update_fields.append("description = ?")
-            params.append(project_update.description)
-        
+            update_data["description"] = project_update.description
+
         if project_update.geojson is not None:
             # Validate GeoJSON
             try:
                 geojson_data = json.loads(project_update.geojson)
                 if not isinstance(geojson_data, dict) or "type" not in geojson_data:
+                    logger.warning(f"[{operation_id}] Invalid GeoJSON format")
                     raise ValueError("Invalid GeoJSON format")
             except json.JSONDecodeError:
+                logger.warning(f"[{operation_id}] Invalid GeoJSON format (JSON decode error)")
                 raise HTTPException(
                     status_code=status.HTTP_400_BAD_REQUEST,
                     detail="Invalid GeoJSON format"
                 )
-            
-            update_fields.append("geojson = ?")
-            params.append(project_update.geojson)
-        
+
+            update_data["geojson"] = project_update.geojson
+
         # Add updated_at field
-        update_fields.append("updated_at = CURRENT_TIMESTAMP")
-        
+        update_data["updated_at"] = time.strftime('%Y-%m-%d %H:%M:%S')
+
         # If no fields to update, return the existing project
-        if not params:
+        if len(update_data) <= 1:  # Only updated_at
+            logger.info(f"[{operation_id}] No fields to update")
             return {
-                "id": existing_project[0],
-                "owner_id": existing_project[1],
-                "title": existing_project[2],
-                "description": existing_project[3],
-                "geojson": existing_project[4],
-                "storage_bucket": existing_project[5],
-                "created_at": existing_project[6],
-                "updated_at": existing_project[7]
+                "id": existing_project.get("id"),
+                "owner_id": existing_project.get("owner_id"),
+                "title": existing_project.get("title"),
+                "description": existing_project.get("description"),
+                "geojson": existing_project.get("geojson"),
+                "storage_bucket": existing_project.get("storage_bucket", "default"),
+                "created_at": existing_project.get("created_at", ""),
+                "updated_at": existing_project.get("updated_at", "")
             }
-        
-        # Update the project
-        update_query = f"""
-        UPDATE projects 
-        SET {', '.join(update_fields)}
-        WHERE id = ? AND owner_id = ?
-        """
-        params.extend([project_id, current_user[0]])
-        
-        request.app.db_conn.execute(update_query, params)
-        request.app.db_conn.commit()
-        
+
+        # Update the project using HTTP API
+        logger.info(f"[{operation_id}] Updating project with data: {update_data}")
+        success = db_http.update_record(
+            "projects",
+            update_data,
+            "id = ? AND owner_id = ?",
+            [
+                {"type": "integer", "value": str(project_id)},
+                {"type": "integer", "value": str(user_id)}
+            ],
+            operation_id=operation_id
+        )
+
+        if not success:
+            logger.error(f"[{operation_id}] Failed to update project")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Failed to update project"
+            )
+
+        logger.info(f"[{operation_id}] Project updated successfully")
+
         # Get the updated project
-        updated_project = request.app.db_conn.execute(
-            "SELECT * FROM projects WHERE id = ?", 
-            (project_id,)
-        ).fetchone()
-        
+        updated_project = db_http.get_record_by_id("projects", project_id, operation_id=operation_id)
+
+        if not updated_project:
+            logger.warning(f"[{operation_id}] Updated project not found")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Project was updated but could not be retrieved"
+            )
+
         return {
-            "id": updated_project[0],
-            "owner_id": updated_project[1],
-            "title": updated_project[2],
-            "description": updated_project[3],
-            "geojson": updated_project[4],
-            "storage_bucket": updated_project[5],
-            "created_at": updated_project[6],
-            "updated_at": updated_project[7]
+            "id": updated_project.get("id"),
+            "owner_id": updated_project.get("owner_id"),
+            "title": updated_project.get("title"),
+            "description": updated_project.get("description"),
+            "geojson": updated_project.get("geojson"),
+            "storage_bucket": updated_project.get("storage_bucket", "default"),
+            "created_at": updated_project.get("created_at", ""),
+            "updated_at": updated_project.get("updated_at", "")
         }
     except HTTPException:
         raise
     except Exception as e:
+        logger.error(f"[{operation_id}] Error updating project: {str(e)}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=str(e)
@@ -258,33 +386,64 @@ async def update_project(
 
 @router.delete("/{project_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_project(
-    request: Request,
     project_id: int,
     current_user = Depends(get_current_user)
 ):
+    operation_id = f"delete_project_{int(time.time())}"
+    logger.info(f"[{operation_id}] Deleting project with ID: {project_id}")
+
     try:
+        # Get user ID based on the type of current_user
+        if isinstance(current_user, dict):
+            user_id = current_user.get("id")
+        else:
+            user_id = current_user[0]
+
+        logger.info(f"[{operation_id}] User ID: {user_id}")
+
         # Check if project exists and belongs to user
-        query = "SELECT * FROM projects WHERE id = ? AND owner_id = ?"
-        existing_project = request.app.db_conn.execute(
-            query, 
-            (project_id, current_user[0])
-        ).fetchone()
-        
-        if not existing_project:
+        projects = db_http.select_records(
+            "projects",
+            condition="id = ? AND owner_id = ?",
+            condition_params=[
+                {"type": "integer", "value": str(project_id)},
+                {"type": "integer", "value": str(user_id)}
+            ],
+            limit=1,
+            operation_id=operation_id
+        )
+
+        if not projects:
+            logger.warning(f"[{operation_id}] Project not found with ID: {project_id}")
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
                 detail="Project not found"
             )
-        
-        # Delete the project
-        delete_query = "DELETE FROM projects WHERE id = ?"
-        request.app.db_conn.execute(delete_query, (project_id,))
-        request.app.db_conn.commit()
-        
+
+        logger.info(f"[{operation_id}] Found project to delete: {projects[0].get('title')}")
+
+        # Delete the project using HTTP API
+        success = db_http.delete_record(
+            "projects",
+            "id = ?",
+            [{"type": "integer", "value": str(project_id)}],
+            operation_id=operation_id
+        )
+
+        if not success:
+            logger.error(f"[{operation_id}] Failed to delete project")
+            raise HTTPException(
+                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                detail="Failed to delete project"
+            )
+
+        logger.info(f"[{operation_id}] Project deleted successfully")
+
         return None
     except HTTPException:
         raise
     except Exception as e:
+        logger.error(f"[{operation_id}] Error deleting project: {str(e)}")
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=str(e)

app/utils/db_http.py

@@ -0,0 +1,578 @@
+"""
+Database utility module for HTTP API operations with Turso database.
+This module provides functions for interacting with the Turso database using the HTTP API.
+"""
+
+import os
+import logging
+import requests
+import time
+from typing import List, Dict, Any, Optional, Tuple, Union
+
+# Configure logging
+logger = logging.getLogger("auth-server")
+
+def get_http_url() -> Tuple[str, str]:
+    """
+    Get the HTTP URL and auth token for the Turso database.
+    
+    Returns:
+        Tuple[str, str]: The HTTP URL and auth token.
+    """
+    # Extract the database URL and auth token
+    db_url = os.getenv("TURSO_DATABASE_URL", "")
+    auth_token = os.getenv("TURSO_AUTH_TOKEN", "")
+    
+    # Convert URL from libsql:// to https://
+    if db_url.startswith("libsql://"):
+        http_url = db_url.replace("libsql://", "https://")
+    else:
+        http_url = db_url
+        
+    # Ensure the URL doesn't have a trailing slash
+    http_url = http_url.rstrip('/')
+    
+    return http_url, auth_token
+
+def get_headers(auth_token: str) -> Dict[str, str]:
+    """
+    Get the headers for the HTTP request.
+    
+    Args:
+        auth_token (str): The authentication token.
+        
+    Returns:
+        Dict[str, str]: The headers for the HTTP request.
+    """
+    return {
+        "Authorization": f"Bearer {auth_token}",
+        "Content-Type": "application/json"
+    }
+
+def execute_query(sql: str, params: List[Dict[str, Any]] = None, operation_id: str = None) -> Dict[str, Any]:
+    """
+    Execute a SQL query using the HTTP API.
+    
+    Args:
+        sql (str): The SQL query to execute.
+        params (List[Dict[str, Any]], optional): The parameters for the query. Defaults to None.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        Dict[str, Any]: The response from the database.
+    """
+    if operation_id is None:
+        operation_id = f"query_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Executing query: {sql}")
+    
+    http_url, auth_token = get_http_url()
+    headers = get_headers(auth_token)
+    
+    # Prepare the query
+    query = {
+        "requests": [
+            {
+                "type": "execute",
+                "stmt": {
+                    "sql": sql,
+                    "args": params or []
+                }
+            },
+            {"type": "close"}
+        ]
+    }
+    
+    # Send the request
+    try:
+        response = requests.post(f"{http_url}/v2/pipeline", headers=headers, json=query)
+        response.raise_for_status()
+        result = response.json()
+        logger.info(f"[{operation_id}] Query executed successfully")
+        return result
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error executing query: {str(e)}")
+        raise
+
+def execute_pipeline(requests: List[Dict[str, Any]], operation_id: str = None) -> Dict[str, Any]:
+    """
+    Execute a pipeline of SQL queries using the HTTP API.
+    
+    Args:
+        requests (List[Dict[str, Any]]): The requests to execute.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        Dict[str, Any]: The response from the database.
+    """
+    if operation_id is None:
+        operation_id = f"pipeline_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Executing pipeline with {len(requests)} requests")
+    
+    http_url, auth_token = get_http_url()
+    headers = get_headers(auth_token)
+    
+    # Prepare the pipeline
+    pipeline = {
+        "requests": requests + [{"type": "close"}]
+    }
+    
+    # Send the request
+    try:
+        response = requests.post(f"{http_url}/v2/pipeline", headers=headers, json=pipeline)
+        response.raise_for_status()
+        result = response.json()
+        logger.info(f"[{operation_id}] Pipeline executed successfully")
+        return result
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error executing pipeline: {str(e)}")
+        raise
+
+def insert_record(table: str, data: Dict[str, Any], operation_id: str = None) -> Optional[int]:
+    """
+    Insert a record into a table.
+    
+    Args:
+        table (str): The table to insert into.
+        data (Dict[str, Any]): The data to insert.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        Optional[int]: The ID of the inserted record, or None if the insert failed.
+    """
+    if operation_id is None:
+        operation_id = f"insert_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Inserting record into {table}")
+    
+    # Prepare the SQL and parameters
+    columns = list(data.keys())
+    placeholders = ["?"] * len(columns)
+    sql = f"INSERT INTO {table} ({', '.join(columns)}) VALUES ({', '.join(placeholders)})"
+    
+    params = []
+    for key, value in data.items():
+        if isinstance(value, str):
+            params.append({"type": "text", "value": value})
+        elif isinstance(value, int):
+            params.append({"type": "integer", "value": str(value)})
+        elif isinstance(value, float):
+            params.append({"type": "float", "value": str(value)})
+        elif value is None:
+            params.append({"type": "null", "value": None})
+        else:
+            params.append({"type": "text", "value": str(value)})
+    
+    # Execute the pipeline with insert and get last ID
+    pipeline_requests = [
+        {
+            "type": "execute",
+            "stmt": {
+                "sql": sql,
+                "args": params
+            }
+        },
+        {
+            "type": "execute",
+            "stmt": {"sql": "SELECT last_insert_rowid()"}
+        }
+    ]
+    
+    try:
+        result = execute_pipeline(pipeline_requests, operation_id)
+        
+        # Check for errors in the first request (insert)
+        if "results" in result and len(result["results"]) > 0:
+            if result["results"][0]["type"] == "error":
+                error_msg = result["results"][0]["error"]["message"]
+                logger.error(f"[{operation_id}] Insert error: {error_msg}")
+                return None
+        
+        # Try to get the last inserted ID
+        last_id = None
+        if "results" in result and len(result["results"]) > 1:
+            if result["results"][1]["type"] == "ok":
+                rows = result["results"][1]["response"]["result"]["rows"]
+                if rows and len(rows) > 0:
+                    last_id = int(rows[0][0]["value"])
+                    logger.info(f"[{operation_id}] Record inserted with ID: {last_id}")
+                    return last_id
+        
+        logger.warning(f"[{operation_id}] Insert succeeded but couldn't get ID")
+        return None
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error inserting record: {str(e)}")
+        return None
+
+def update_record(table: str, data: Dict[str, Any], condition: str, condition_params: List[Dict[str, Any]], operation_id: str = None) -> bool:
+    """
+    Update a record in a table.
+    
+    Args:
+        table (str): The table to update.
+        data (Dict[str, Any]): The data to update.
+        condition (str): The condition for the update (e.g., "id = ?").
+        condition_params (List[Dict[str, Any]]): The parameters for the condition.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        bool: True if the update succeeded, False otherwise.
+    """
+    if operation_id is None:
+        operation_id = f"update_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Updating record in {table}")
+    
+    # Prepare the SQL and parameters
+    set_clauses = [f"{key} = ?" for key in data.keys()]
+    sql = f"UPDATE {table} SET {', '.join(set_clauses)} WHERE {condition}"
+    
+    params = []
+    for key, value in data.items():
+        if isinstance(value, str):
+            params.append({"type": "text", "value": value})
+        elif isinstance(value, int):
+            params.append({"type": "integer", "value": str(value)})
+        elif isinstance(value, float):
+            params.append({"type": "float", "value": str(value)})
+        elif value is None:
+            params.append({"type": "null", "value": None})
+        else:
+            params.append({"type": "text", "value": str(value)})
+    
+    # Add condition parameters
+    params.extend(condition_params)
+    
+    try:
+        result = execute_query(sql, params, operation_id)
+        
+        # Check for errors
+        if "results" in result and len(result["results"]) > 0:
+            if result["results"][0]["type"] == "error":
+                error_msg = result["results"][0]["error"]["message"]
+                logger.error(f"[{operation_id}] Update error: {error_msg}")
+                return False
+            
+            # Check affected rows
+            affected_rows = result["results"][0]["response"]["result"]["affected_row_count"]
+            logger.info(f"[{operation_id}] Updated {affected_rows} rows")
+            return affected_rows > 0
+        
+        return False
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error updating record: {str(e)}")
+        return False
+
+def delete_record(table: str, condition: str, condition_params: List[Dict[str, Any]], operation_id: str = None) -> bool:
+    """
+    Delete a record from a table.
+    
+    Args:
+        table (str): The table to delete from.
+        condition (str): The condition for the delete (e.g., "id = ?").
+        condition_params (List[Dict[str, Any]]): The parameters for the condition.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        bool: True if the delete succeeded, False otherwise.
+    """
+    if operation_id is None:
+        operation_id = f"delete_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Deleting record from {table}")
+    
+    # Prepare the SQL
+    sql = f"DELETE FROM {table} WHERE {condition}"
+    
+    try:
+        result = execute_query(sql, condition_params, operation_id)
+        
+        # Check for errors
+        if "results" in result and len(result["results"]) > 0:
+            if result["results"][0]["type"] == "error":
+                error_msg = result["results"][0]["error"]["message"]
+                logger.error(f"[{operation_id}] Delete error: {error_msg}")
+                return False
+            
+            # Check affected rows
+            affected_rows = result["results"][0]["response"]["result"]["affected_row_count"]
+            logger.info(f"[{operation_id}] Deleted {affected_rows} rows")
+            return affected_rows > 0
+        
+        return False
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error deleting record: {str(e)}")
+        return False
+
+def select_records(table: str, columns: List[str] = None, condition: str = None, 
+                  condition_params: List[Dict[str, Any]] = None, limit: int = None, 
+                  offset: int = None, order_by: str = None, operation_id: str = None) -> List[Dict[str, Any]]:
+    """
+    Select records from a table.
+    
+    Args:
+        table (str): The table to select from.
+        columns (List[str], optional): The columns to select. Defaults to None (all columns).
+        condition (str, optional): The condition for the select. Defaults to None.
+        condition_params (List[Dict[str, Any]], optional): The parameters for the condition. Defaults to None.
+        limit (int, optional): The maximum number of records to return. Defaults to None.
+        offset (int, optional): The number of records to skip. Defaults to None.
+        order_by (str, optional): The order by clause. Defaults to None.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        List[Dict[str, Any]]: The selected records.
+    """
+    if operation_id is None:
+        operation_id = f"select_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Selecting records from {table}")
+    
+    # Prepare the SQL
+    cols = "*" if not columns else ", ".join(columns)
+    sql = f"SELECT {cols} FROM {table}"
+    
+    if condition:
+        sql += f" WHERE {condition}"
+    
+    if order_by:
+        sql += f" ORDER BY {order_by}"
+    
+    if limit:
+        sql += f" LIMIT {limit}"
+    
+    if offset:
+        sql += f" OFFSET {offset}"
+    
+    try:
+        result = execute_query(sql, condition_params, operation_id)
+        
+        # Check for errors
+        if "results" in result and len(result["results"]) > 0:
+            if result["results"][0]["type"] == "error":
+                error_msg = result["results"][0]["error"]["message"]
+                logger.error(f"[{operation_id}] Select error: {error_msg}")
+                return []
+            
+            # Extract the records
+            response = result["results"][0]["response"]
+            if "result" in response and "rows" in response["result"]:
+                rows = response["result"]["rows"]
+                cols = response["result"]["cols"]
+                
+                # Convert rows to dictionaries
+                records = []
+                for row in rows:
+                    record = {}
+                    for i, col in enumerate(cols):
+                        col_name = col["name"]
+                        value = row[i]["value"]
+                        
+                        # Convert value based on type
+                        if row[i]["type"] == "integer":
+                            value = int(value)
+                        elif row[i]["type"] == "float":
+                            value = float(value)
+                        elif row[i]["type"] == "null":
+                            value = None
+                        
+                        record[col_name] = value
+                    
+                    records.append(record)
+                
+                logger.info(f"[{operation_id}] Selected {len(records)} records")
+                return records
+        
+        logger.warning(f"[{operation_id}] No records found")
+        return []
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error selecting records: {str(e)}")
+        return []
+
+def get_record_by_id(table: str, id: int, columns: List[str] = None, operation_id: str = None) -> Optional[Dict[str, Any]]:
+    """
+    Get a record by ID.
+    
+    Args:
+        table (str): The table to select from.
+        id (int): The ID of the record.
+        columns (List[str], optional): The columns to select. Defaults to None (all columns).
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        Optional[Dict[str, Any]]: The record, or None if not found.
+    """
+    if operation_id is None:
+        operation_id = f"get_by_id_{int(time.time())}"
+    
+    condition = "id = ?"
+    condition_params = [{"type": "integer", "value": str(id)}]
+    
+    records = select_records(table, columns, condition, condition_params, limit=1, operation_id=operation_id)
+    
+    if records:
+        return records[0]
+    
+    return None
+
+def count_records(table: str, condition: str = None, condition_params: List[Dict[str, Any]] = None, operation_id: str = None) -> int:
+    """
+    Count records in a table.
+    
+    Args:
+        table (str): The table to count records in.
+        condition (str, optional): The condition for the count. Defaults to None.
+        condition_params (List[Dict[str, Any]], optional): The parameters for the condition. Defaults to None.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        int: The number of records.
+    """
+    if operation_id is None:
+        operation_id = f"count_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Counting records in {table}")
+    
+    # Prepare the SQL
+    sql = f"SELECT COUNT(*) FROM {table}"
+    
+    if condition:
+        sql += f" WHERE {condition}"
+    
+    try:
+        result = execute_query(sql, condition_params, operation_id)
+        
+        # Check for errors
+        if "results" in result and len(result["results"]) > 0:
+            if result["results"][0]["type"] == "error":
+                error_msg = result["results"][0]["error"]["message"]
+                logger.error(f"[{operation_id}] Count error: {error_msg}")
+                return 0
+            
+            # Extract the count
+            response = result["results"][0]["response"]
+            if "result" in response and "rows" in response["result"] and response["result"]["rows"]:
+                count = int(response["result"]["rows"][0][0]["value"])
+                logger.info(f"[{operation_id}] Counted {count} records")
+                return count
+        
+        logger.warning(f"[{operation_id}] Count failed")
+        return 0
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error counting records: {str(e)}")
+        return 0
+
+def record_exists(table: str, condition: str, condition_params: List[Dict[str, Any]], operation_id: str = None) -> bool:
+    """
+    Check if a record exists in a table.
+    
+    Args:
+        table (str): The table to check.
+        condition (str): The condition for the check.
+        condition_params (List[Dict[str, Any]]): The parameters for the condition.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        bool: True if the record exists, False otherwise.
+    """
+    count = count_records(table, condition, condition_params, operation_id)
+    return count > 0
+
+def create_table_if_not_exists(table: str, schema: str, operation_id: str = None) -> bool:
+    """
+    Create a table if it doesn't exist.
+    
+    Args:
+        table (str): The table to create.
+        schema (str): The schema for the table.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        bool: True if the table was created or already exists, False otherwise.
+    """
+    if operation_id is None:
+        operation_id = f"create_table_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Creating table {table} if not exists")
+    
+    # Prepare the SQL
+    sql = f"CREATE TABLE IF NOT EXISTS {table} ({schema})"
+    
+    try:
+        result = execute_query(sql, operation_id=operation_id)
+        
+        # Check for errors
+        if "results" in result and len(result["results"]) > 0:
+            if result["results"][0]["type"] == "error":
+                error_msg = result["results"][0]["error"]["message"]
+                logger.error(f"[{operation_id}] Create table error: {error_msg}")
+                return False
+            
+            logger.info(f"[{operation_id}] Table {table} created or already exists")
+            return True
+        
+        return False
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error creating table: {str(e)}")
+        return False
+
+def execute_transaction(queries: List[Tuple[str, List[Dict[str, Any]]]], operation_id: str = None) -> bool:
+    """
+    Execute a transaction with multiple queries.
+    
+    Args:
+        queries (List[Tuple[str, List[Dict[str, Any]]]]): The queries to execute.
+        operation_id (str, optional): A unique identifier for the operation. Defaults to None.
+        
+    Returns:
+        bool: True if the transaction succeeded, False otherwise.
+    """
+    if operation_id is None:
+        operation_id = f"transaction_{int(time.time())}"
+        
+    logger.info(f"[{operation_id}] Executing transaction with {len(queries)} queries")
+    
+    # Prepare the pipeline
+    requests = [{"type": "execute", "stmt": {"sql": "BEGIN"}}]
+    
+    for sql, params in queries:
+        requests.append({
+            "type": "execute",
+            "stmt": {
+                "sql": sql,
+                "args": params or []
+            }
+        })
+    
+    requests.append({"type": "execute", "stmt": {"sql": "COMMIT"}})
+    
+    try:
+        result = execute_pipeline(requests, operation_id)
+        
+        # Check for errors
+        for i, res in enumerate(result.get("results", [])):
+            if res.get("type") == "error":
+                error_msg = res.get("error", {}).get("message", "Unknown error")
+                logger.error(f"[{operation_id}] Transaction error in query {i}: {error_msg}")
+                
+                # Try to rollback
+                try:
+                    execute_query("ROLLBACK", operation_id=f"{operation_id}_rollback")
+                except:
+                    pass
+                
+                return False
+        
+        logger.info(f"[{operation_id}] Transaction executed successfully")
+        return True
+    except Exception as e:
+        logger.error(f"[{operation_id}] Error executing transaction: {str(e)}")
+        
+        # Try to rollback
+        try:
+            execute_query("ROLLBACK", operation_id=f"{operation_id}_rollback")
+        except:
+            pass
+        
+        return False

main.py

@@ -12,6 +12,10 @@ import uvicorn
 from dotenv import load_dotenv
 from passlib.context import CryptContext
 from pydantic import BaseModel, EmailStr
+from typing import Dict, Any, List, Optional
+
+# Import the HTTP API database utility
+from app.utils import db_http
 
 # Configure logging
 logging.basicConfig(
@@ -461,6 +465,71 @@ async def startup_db_client():
         # This function is kept for reference but is not used anymore
         pass
 
+    # Function to create database tables using HTTP API
+    def create_tables_http():
+        """
+        Create database tables if they don't exist using the HTTP API.
+        """
+        operation_id = f"create_tables_http_{int(time.time())}"
+        logger.info(f"[{operation_id}] Creating database tables using HTTP API")
+
+        try:
+            # Import the HTTP API utility
+            from app.utils import db_http
+
+            # Create users table
+            logger.info(f"[{operation_id}] Creating users table")
+            db_http.create_table_if_not_exists(
+                "users",
+                """
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                email TEXT NOT NULL UNIQUE,
+                hashed_password TEXT NOT NULL CHECK(LENGTH(hashed_password) = 97),
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                last_login DATETIME,
+                is_admin BOOLEAN DEFAULT 0
+                """,
+                operation_id=f"{operation_id}_users"
+            )
+
+            # Create projects table
+            logger.info(f"[{operation_id}] Creating projects table")
+            db_http.create_table_if_not_exists(
+                "projects",
+                """
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                owner_id INTEGER NOT NULL,
+                title TEXT NOT NULL,
+                description TEXT,
+                geojson TEXT,
+                storage_bucket TEXT,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (owner_id) REFERENCES users (id)
+                """,
+                operation_id=f"{operation_id}_projects"
+            )
+
+            # Create journals table
+            logger.info(f"[{operation_id}] Creating journals table")
+            db_http.create_table_if_not_exists(
+                "journals",
+                """
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                project_id INTEGER NOT NULL,
+                title TEXT NOT NULL,
+                content TEXT,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (project_id) REFERENCES projects (id)
+                """,
+                operation_id=f"{operation_id}_journals"
+            )
+
+            logger.info(f"[{operation_id}] All database tables created successfully")
+        except Exception as e:
+            logger.error(f"[{operation_id}] Error creating database tables: {str(e)}")
+
     # Initialize database connection
     connected = False
 
@@ -867,6 +936,13 @@ async def startup_db_client():
 
         # Create super user if it doesn't exist using HTTP API
         create_super_user_http()
+
+        # Also create tables using HTTP API for better reliability
+        try:
+            logger.info("Creating tables using HTTP API for better reliability")
+            create_tables_http()
+        except Exception as e:
+            logger.error(f"Error creating tables with HTTP API: {str(e)}")
     except Exception as e:
         error_msg = f"Failed to create database tables: {str(e)}"
         logger.error(error_msg)
@@ -874,6 +950,13 @@ async def startup_db_client():
         # This allows the server to start and use existing tables if they exist
         logger.warning("Continuing server startup despite table creation errors")
 
+        # Try to create tables using HTTP API as a fallback
+        try:
+            logger.info("Trying to create tables using HTTP API as a fallback")
+            create_tables_http()
+        except Exception as e:
+            logger.error(f"Error creating tables with HTTP API fallback: {str(e)}")
+
 @app.on_event("shutdown")
 async def shutdown_db_client():
     logger.info("Shutting down database connection")
@@ -2091,162 +2174,203 @@ async def test_db_http():
         results["success_count"] += 1
         results["total_count"] += 1
 
-        # Test 5: Create users table
-        create_users_query = {
-            "requests": [
-                {
-                    "type": "execute",
-                    "stmt": {
-                        "sql": """
-                        CREATE TABLE IF NOT EXISTS users (
-                            id INTEGER PRIMARY KEY AUTOINCREMENT,
-                            email TEXT NOT NULL UNIQUE,
-                            hashed_password TEXT NOT NULL,
-                            created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
-                            last_login DATETIME,
-                            is_admin INTEGER DEFAULT 0
-                        )
-                        """
-                    }
-                },
-                {"type": "close"}
-            ]
-        }
+    except Exception as e:
+        results["error"] = str(e)
 
-        create_users_response = requests.post(f"{http_url}/v2/pipeline", headers=headers, json=create_users_query)
-        create_users_response.raise_for_status()
+    # Add success rate
+    results["success_rate"] = f"{results['success_count']}/{results['total_count']}"
 
-        results["tests"].append({
-            "name": "Create users table (HTTP API)",
-            "success": True
-        })
-        results["success_count"] += 1
-        results["total_count"] += 1
+    return results
 
-        # Test 6: Create a test user
-        test_email = f"test_user_http_{int(time.time())}@example.com"
-        test_password = "TestPassword123!"
+@app.get("/test-db-http-util", tags=["General"])
+async def test_db_http_util():
+    """
+    Test the database connection and operations using the HTTP API utility.
+    This endpoint is for testing purposes only.
 
-        # Hash the password
-        pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
-        hashed_password = pwd_context.hash(test_password)
+    Returns:
+        dict: Information about the database tests.
+    """
+    logger.info("Test database HTTP API utility endpoint accessed")
 
-        create_user_query = {
-            "requests": [
-                {
-                    "type": "execute",
-                    "stmt": {
-                        "sql": "INSERT INTO users (email, hashed_password) VALUES (?, ?)",
-                        "args": [
-                            {"type": "text", "value": test_email},
-                            {"type": "text", "value": hashed_password}
-                        ]
-                    }
-                },
-                {
-                    "type": "execute",
-                    "stmt": {"sql": "SELECT last_insert_rowid()"}
-                },
-                {"type": "close"}
-            ]
-        }
+    try:
+        # Import the HTTP API utility
+        from app.utils import db_http
 
-        create_user_response = requests.post(f"{http_url}/v2/pipeline", headers=headers, json=create_user_query)
-        create_user_response.raise_for_status()
-        create_user_result = create_user_response.json()
+        operation_id = f"test_db_http_util_{int(time.time())}"
+        logger.info(f"[{operation_id}] Starting HTTP API utility test")
 
-        # Try to get the last inserted ID
-        user_id = None
-        if "results" in create_user_result and len(create_user_result["results"]) > 1:
-            result = create_user_result["results"][1]
-            if "rows" in result and len(result["rows"]) > 0:
-                user_id = result["rows"][0]["values"][0]
+        results = {
+            "connection_type": "http-api-util",
+            "operations": []
+        }
 
-        results["tests"].append({
-            "name": "Create test user (HTTP API)",
+        # Test 1: Simple SELECT
+        logger.info(f"[{operation_id}] Test 1: Simple SELECT")
+        result = db_http.execute_query("SELECT 1 as test", operation_id=f"{operation_id}_1")
+        results["operations"].append({
+            "name": "Simple SELECT",
             "success": True,
-            "user_id": user_id,
-            "email": test_email
+            "result": str(result)
         })
-        results["success_count"] += 1
-        results["total_count"] += 1
 
-        # Test 7: Verify the user was created
-        verify_query = {
-            "requests": [
-                {
-                    "type": "execute",
-                    "stmt": {
-                        "sql": "SELECT * FROM users WHERE email = ?",
-                        "args": [
-                            {"type": "text", "value": test_email}
-                        ]
-                    }
-                },
-                {"type": "close"}
-            ]
-        }
+        # Test 2: Create a temporary table
+        logger.info(f"[{operation_id}] Test 2: Create a temporary table")
+        db_http.create_table_if_not_exists(
+            "test_table_http_util",
+            """
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            value TEXT NOT NULL,
+            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+            """,
+            operation_id=f"{operation_id}_2"
+        )
+        results["operations"].append({
+            "name": "Create temporary table",
+            "success": True
+        })
 
-        verify_response = requests.post(f"{http_url}/v2/pipeline", headers=headers, json=verify_query)
-        verify_response.raise_for_status()
-        verify_result = verify_response.json()
+        # Test 3: Insert into the temporary table
+        logger.info(f"[{operation_id}] Test 3: Insert into the temporary table")
+        test_value = f"test_value_http_util_{int(time.time())}"
 
-        # Check if the user was found
-        user_found = False
-        if "results" in verify_result and len(verify_result["results"]) > 0:
-            result = verify_result["results"][0]
-            if "rows" in result and len(result["rows"]) > 0:
-                user_found = True
+        insert_id = db_http.insert_record(
+            "test_table_http_util",
+            {"value": test_value},
+            operation_id=f"{operation_id}_3"
+        )
 
-        results["tests"].append({
-            "name": "Verify test user (HTTP API)",
-            "success": user_found,
-            "result": str(verify_result)
+        results["operations"].append({
+            "name": "Insert into temporary table",
+            "success": insert_id is not None,
+            "insert_id": insert_id
         })
 
-        if user_found:
-            results["success_count"] += 1
+        # Test 4: Select from the temporary table
+        logger.info(f"[{operation_id}] Test 4: Select from the temporary table")
+        records = db_http.select_records(
+            "test_table_http_util",
+            condition="value = ?",
+            condition_params=[{"type": "text", "value": test_value}],
+            limit=1,
+            operation_id=f"{operation_id}_4"
+        )
 
-        results["total_count"] += 1
+        results["operations"].append({
+            "name": "Select from temporary table",
+            "success": len(records) > 0,
+            "result": str(records[0]) if records else None
+        })
 
-        # Test 8: List all users
-        list_query = {
-            "requests": [
-                {
-                    "type": "execute",
-                    "stmt": {"sql": "SELECT id, email FROM users"}
-                },
-                {"type": "close"}
-            ]
-        }
+        # Test 5: Update the record
+        logger.info(f"[{operation_id}] Test 5: Update the record")
+        updated_value = f"{test_value}_updated"
 
-        list_response = requests.post(f"{http_url}/v2/pipeline", headers=headers, json=list_query)
-        list_response.raise_for_status()
-        list_result = list_response.json()
+        success = db_http.update_record(
+            "test_table_http_util",
+            {"value": updated_value},
+            "id = ?",
+            [{"type": "integer", "value": str(insert_id)}],
+            operation_id=f"{operation_id}_5"
+        )
 
-        # Extract the list of users
-        all_users = []
-        if "results" in list_result and len(list_result["results"]) > 0:
-            result = list_result["results"][0]
-            if "rows" in result:
-                for row in result["rows"]:
-                    all_users.append(row["values"])
+        results["operations"].append({
+            "name": "Update record",
+            "success": success
+        })
 
-        results["tests"].append({
-            "name": "List all users (HTTP API)",
+        # Test 6: Verify the update
+        logger.info(f"[{operation_id}] Test 6: Verify the update")
+        updated_records = db_http.select_records(
+            "test_table_http_util",
+            condition="id = ?",
+            condition_params=[{"type": "integer", "value": str(insert_id)}],
+            limit=1,
+            operation_id=f"{operation_id}_6"
+        )
+
+        results["operations"].append({
+            "name": "Verify update",
+            "success": len(updated_records) > 0 and updated_records[0].get("value") == updated_value,
+            "result": str(updated_records[0]) if updated_records else None
+        })
+
+        # Test 7: Count records
+        logger.info(f"[{operation_id}] Test 7: Count records")
+        count = db_http.count_records("test_table_http_util", operation_id=f"{operation_id}_7")
+
+        results["operations"].append({
+            "name": "Count records",
+            "success": count > 0,
+            "count": count
+        })
+
+        # Test 8: List all records
+        logger.info(f"[{operation_id}] Test 8: List all records")
+        all_records = db_http.select_records(
+            "test_table_http_util",
+            limit=10,
+            operation_id=f"{operation_id}_8"
+        )
+
+        results["operations"].append({
+            "name": "List all records",
             "success": True,
-            "users": all_users
+            "count": len(all_records),
+            "records": all_records
         })
-        results["success_count"] += 1
-        results["total_count"] += 1
 
-    except Exception as e:
-        results["error"] = str(e)
+        # Test 9: Create a test user
+        logger.info(f"[{operation_id}] Test 9: Create a test user")
 
-    # Add success rate
-    results["success_rate"] = f"{results['success_count']}/{results['total_count']}"
+        # Create password hashing context
+        from app.api.routes.auth_router import get_password_hash
+
+        test_user_email = f"test_user_http_util_{int(time.time())}@example.com"
+        test_user_password = "TestPassword123!"
+        hashed_password = get_password_hash(test_user_password)
+
+        user_id = db_http.insert_record(
+            "users",
+            {
+                "email": test_user_email,
+                "hashed_password": hashed_password,
+                "is_admin": 0
+            },
+            operation_id=f"{operation_id}_9"
+        )
 
-    return results
+        results["operations"].append({
+            "name": "Create test user",
+            "success": user_id is not None,
+            "user_id": user_id,
+            "email": test_user_email
+        })
+
+        # Test 10: Verify the user was created
+        logger.info(f"[{operation_id}] Test 10: Verify the user was created")
+        users = db_http.select_records(
+            "users",
+            condition="email = ?",
+            condition_params=[{"type": "text", "value": test_user_email}],
+            limit=1,
+            operation_id=f"{operation_id}_10"
+        )
+
+        results["operations"].append({
+            "name": "Verify test user",
+            "success": len(users) > 0,
+            "result": str(users[0]) if users else None
+        })
+
+        logger.info(f"[{operation_id}] HTTP API utility test completed successfully")
+        return results
+    except Exception as e:
+        logger.error(f"Error during HTTP API utility test: {str(e)}")
+        return {
+            "error": str(e),
+            "operations": []
+        }
 
 if __name__ == "__main__":
     port = int(os.getenv("PORT", 7860))