Batch LinkedIn import and add rename endpoint

Improve queue API: make LinkedIn CSV import robust and batched, add column-rename endpoints, and return more recent records. Changes include: switch import to linkedin.csv with dynamic header detection, skip blank rows, batch inserts (500) with an added 'group' value and imported count returned; update get route to return 10 most-recent records instead of one; add new rename/rename_column route handlers to ALTER TABLE RENAME COLUMN with error handling; wire up the rename router in queue __init__.py and add a sample queue_output.txt. These updates improve import reliability and provide a programmatic column-rename API.

Author: goldlabelapps

app/api/queue/__init__.py

@@ -9,12 +9,15 @@
 from .routes.create import router as create_router
 
 from .routes.import_linkedin import router as import_linkedin_router
+
 from .routes.alter import router as alter_router
+from .routes.rename_column import router as rename_router
 
 router = APIRouter()
 router.include_router(drop_router)
 router.include_router(empty_router)
 router.include_router(get_router)
 router.include_router(create_router)
 router.include_router(import_linkedin_router)
-router.include_router(alter_router)
+router.include_router(alter_router)
+router.include_router(rename_router)

app/api/queue/routes/get.py

@@ -27,20 +27,20 @@ def read_queue() -> dict:
             for row in cursor.fetchall()
         ]
 
-        # 3. Get most recently updated record
-        cursor.execute("SELECT * FROM queue ORDER BY updated DESC LIMIT 1;")
+        # 3. Get the 10 most recently updated records
+        cursor.execute("SELECT * FROM queue ORDER BY updated DESC LIMIT 10;")
         columns = [desc[0] for desc in cursor.description] if cursor.description else []
-        row = cursor.fetchone()
-        most_recent = dict(zip(columns, row)) if row and columns else None
+        rows = cursor.fetchall()
+        most_recent = [dict(zip(columns, row)) for row in rows] if rows and columns else []
 
         conn.close()
 
         return {
             "meta": make_meta("success", "Queue table info"),
             "data": {
-                    "queued": record_count,
-                    "most_recent": most_recent,
-                    # "schema": schema
+                "queued": record_count,
+                "most_recent": most_recent,
+                # "schema": schema
             }
         }
     except Exception as e:

app/api/queue/routes/import_linkedin.py

@@ -9,38 +9,73 @@
 
 @router.post("/queue/import/linkedin")
 def import_linkedin_csv() -> dict:
-    """POST /queue/import/linkedin: Import data from linkedin_sample.csv into the queue table."""
+    """POST /queue/import/linkedin: Import data from linkedin.csv into the queue table, robust for large files."""
     csv_path = os.path.join(os.path.dirname(__file__), "../csv/linkedin/linkedin_sample.csv")
     if not os.path.exists(csv_path):
-        raise HTTPException(status_code=404, detail="linkedin_sample.csv not found")
+        raise HTTPException(status_code=404, detail="linkedin.csv not found")
     try:
         conn = get_db_connection_direct()
         cursor = conn.cursor()
         with open(csv_path, newline='', encoding='utf-8') as csvfile:
-            reader = csv.DictReader(row for row in csvfile if not row.startswith('Notes:'))
+            # Find the header line dynamically
+            header_line = None
+            pre_data_lines = []
+            while True:
+                pos = csvfile.tell()
+                line = csvfile.readline()
+                if not line:
+                    break
+                if line.strip().startswith("First Name,Last Name,URL,Email Address,Company,Position,Connected On"):
+                    header_line = line.strip()
+                    break
+                pre_data_lines.append(line)
+            if not header_line:
+                raise HTTPException(status_code=400, detail="CSV header not found.")
+            # Use DictReader with the found header
+            fieldnames = header_line.split(",")
+            reader = csv.DictReader(csvfile, fieldnames=fieldnames)
             now = int(time.time())
+            batch = []
+            batch_size = 500
+            first_row = None
+            imported_count = 0
             for row in reader:
-                cursor.execute(
-                    """
-                    INSERT INTO queue (first_name, last_name, url, email_address, company, position, connected_on, created, updated, hidden, collection)
-                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
-                    """,
-                    [
-                        row.get('First Name'),
-                        row.get('Last Name'),
-                        row.get('URL'),
-                        row.get('Email Address'),
-                        row.get('Company'),
-                        row.get('Position'),
-                        row.get('Connected On'),
-                        now,
-                        now,
-                        False,
-                        'prospects'
-                    ]
+                # Skip any rows that are just blank or not data
+                if not any(row.values()):
+                    continue
+                if first_row is None:
+                    first_row = row.copy()
+                    print("DEBUG: First parsed row from CSV:", first_row)
+                batch.append([
+                    row.get('First Name'),
+                    row.get('Last Name'),
+                    row.get('URL'),
+                    row.get('Email Address'),
+                    row.get('Company'),
+                    row.get('Position'),
+                    row.get('Connected On'),
+                    now,
+                    now,
+                    False,
+                    'prospects',
+                    'linkedin'
+                ])
+                imported_count += 1
+                if len(batch) >= batch_size:
+                    cursor.executemany(
+                        '''INSERT INTO queue (first_name, last_name, url, email_address, company, position, connected_on, created, updated, hidden, collection, "group")
+                        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)''',
+                        batch
+                    )
+                    batch = []
+            if batch:
+                cursor.executemany(
+                    '''INSERT INTO queue (first_name, last_name, url, email_address, company, position, connected_on, created, updated, hidden, collection, "group")
+                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)''',
+                    batch
                 )
         conn.commit()
         conn.close()
-        return {"meta": make_meta("success", "LinkedIn CSV imported")}
+        return {"meta": make_meta("success", f"LinkedIn CSV imported (batched): {imported_count} records imported"), "imported": imported_count}
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))

app/api/queue/routes/rename.py

@@ -0,0 +1,56 @@
+from fastapi import APIRouter, HTTPException, Body
+from app.utils.make_meta import make_meta
+from app.utils.db import get_db_connection_direct
+
+router = APIRouter()
+
+@router.post("/queue/alter/rename_column")
+def rename_column(
+    old_name: str = Body(..., embed=True),
+    new_name: str = Body(..., embed=True),
+    column_type: str = Body(..., embed=True)
+) -> dict:
+    """POST /queue/alter/rename-column: Rename a column in the queue table."""
+    try:
+        conn = get_db_connection_direct()
+        cursor = conn.cursor()
+        sql = f'ALTER TABLE queue RENAME COLUMN "{old_name}" TO "{new_name}";'
+        cursor.execute(sql)
+        conn.commit()
+        conn.close()
+        return {"meta": make_meta("success", f"Column '{old_name}' renamed to '{new_name}'")}
+    except Exception as e:
+        msg = str(e)
+        if 'does not exist' in msg:
+            raise HTTPException(status_code=400, detail=f"Column '{old_name}' does not exist in queue table.")
+        if 'already exists' in msg:
+            raise HTTPException(status_code=400, detail=f"Column '{new_name}' already exists in queue table.")
+        raise HTTPException(status_code=500, detail=msg)
+from fastapi import APIRouter, HTTPException, Body
+from app.utils.make_meta import make_meta
+from app.utils.db import get_db_connection_direct
+
+router = APIRouter()
+
+@router.post("/queue/alter/rename_column")
+def rename_column(
+    old_name: str = Body(..., embed=True),
+    new_name: str = Body(..., embed=True),
+    column_type: str = Body(..., embed=True)
+) -> dict:
+    """POST /queue/alter/rename-column: Rename a column in the queue table."""
+    try:
+        conn = get_db_connection_direct()
+        cursor = conn.cursor()
+        sql = f'ALTER TABLE queue RENAME COLUMN "{old_name}" TO "{new_name}";'
+        cursor.execute(sql)
+        conn.commit()
+        conn.close()
+        return {"meta": make_meta("success", f"Column '{old_name}' renamed to '{new_name}'")}
+    except Exception as e:
+        msg = str(e)
+        if 'does not exist' in msg:
+            raise HTTPException(status_code=400, detail=f"Column '{old_name}' does not exist in queue table.")
+        if 'already exists' in msg:
+            raise HTTPException(status_code=400, detail=f"Column '{new_name}' already exists in queue table.")
+        raise HTTPException(status_code=500, detail=msg)

app/api/queue/routes/rename_column.py

@@ -0,0 +1,28 @@
+from fastapi import APIRouter, HTTPException, Body
+from app.utils.make_meta import make_meta
+from app.utils.db import get_db_connection_direct
+
+router = APIRouter()
+
+@router.post("/queue/alter/rename_column")
+def rename_column(
+    old_name: str = Body(..., embed=True),
+    new_name: str = Body(..., embed=True),
+    column_type: str = Body(..., embed=True)
+) -> dict:
+    """POST /queue/alter/rename_column: Rename a column in the queue table."""
+    try:
+        conn = get_db_connection_direct()
+        cursor = conn.cursor()
+        sql = f'ALTER TABLE queue RENAME COLUMN "{old_name}" TO "{new_name}";'
+        cursor.execute(sql)
+        conn.commit()
+        conn.close()
+        return {"meta": make_meta("success", f"Column '{old_name}' renamed to '{new_name}'")}
+    except Exception as e:
+        msg = str(e)
+        if 'does not exist' in msg:
+            raise HTTPException(status_code=400, detail=f"Column '{old_name}' does not exist in queue table.")
+        if 'already exists' in msg:
+            raise HTTPException(status_code=400, detail=f"Column '{new_name}' already exists in queue table.")
+        raise HTTPException(status_code=500, detail=msg)

queue_output.txt

@@ -0,0 +1,11 @@
+id,first_name,last_name,url,email_address,company,position,connected_on,created,updated,hidden,collection,group
+22,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+23,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+24,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+25,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+26,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+27,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+28,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+29,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+30,,,,,,,,1776513092,1776513092,False,prospects,linkedin
+21,,,,,,,,1776513092,1776513092,False,prospects,linkedin