import sqlite3
import pandas as pd
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, JSONResponse
DB_PATH = "CarOwners_Full_Final.db"
PAGE_SIZE = 200 # rows per page
app = FastAPI()
# Connect SQLite
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
cursor = conn.cursor()
# Create indexes for fast filtering/searching
cursor.execute("CREATE INDEX IF NOT EXISTS idx_city ON owners(City);")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_name ON owners(Name);")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_mobile ON owners([Mobile No]);")
conn.commit()
# =========================
# API endpoint: fetch paginated data with filters/search
# =========================
@app.get("/api/data")
def get_data(offset: int = 0, city: str = "", search: str = ""):
query = "SELECT * FROM owners WHERE 1=1"
params = []
if city:
query += " AND LOWER(City) LIKE ?"
params.append(f"%{city.lower()}%")
if search:
query += " AND (LOWER(Name) LIKE ? OR [Mobile No] LIKE ?)"
params.extend([f"%{search.lower()}%", f"%{search}%"])
query += " LIMIT ? OFFSET ?"
params.extend([PAGE_SIZE, offset])
df = pd.read_sql_query(query, conn, params=params)
# 🔹 JSON-safe conversion
# Replace NaN with None
df = df.where(pd.notnull(df), None)
# Replace inf/-inf with None
df = df.replace([float("inf"), float("-inf")], None)
# Convert all numeric columns to strings to prevent JSON float overflow
numeric_cols = df.select_dtypes(include=["float64", "int64"]).columns
for col in numeric_cols:
df[col] = df[col].apply(lambda x: str(x) if x is not None else None)
return JSONResponse(df.to_dict(orient="records"))
# =========================
# API endpoint: total count for pagination
# =========================
@app.get("/api/count")
def get_count(city: str = "", search: str = ""):
query = "SELECT COUNT(*) FROM owners WHERE 1=1"
params = []
if city:
query += " AND LOWER(City) LIKE ?"
params.append(f"%{city.lower()}%")
if search:
query += " AND (LOWER(Name) LIKE ? OR [Mobile No] LIKE ?)"
params.extend([f"%{search.lower()}%", f"%{search}%"])
if params:
cursor.execute(query, tuple(params))
else:
cursor.execute(query)
total = cursor.fetchone()[0]
return {"total": total}
# =========================
# Frontend page with infinite scroll + filters
# =========================
@app.get("/", response_class=HTMLResponse)
def home():
html = """
Car Owners Data - Infinite Scroll + Search