app.py

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from transformers import AutoModelForCausalLM, AutoTokenizer
import sqlite3
import torch

app = FastAPI()

# Load the DeepSeek model and tokenizer
MODEL_NAME = "deepseek-ai/deepseek-coder-1.3b-instruct"
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForCausalLM.from_pretrained(MODEL_NAME, torch_dtype=torch.float16).to("cpu")  # Use "cuda" if available

# SQLite database file
DATABASE_FILE = "example.db"

class ChatRequest(BaseModel):
    message: str

def generate_sql_query(user_input: str) -> str:
    """
    Generate an SQL query from a natural language query using the DeepSeek model.
    """
    inputs = tokenizer(user_input, return_tensors="pt")
    outputs = model.generate(**inputs, max_length=100)
    sql_query = tokenizer.decode(outputs[0], skip_special_tokens=True)
    return sql_query

def execute_sql_query(sql_query: str):
    """
    Execute the SQL query on the SQLite database and return the results.
    """
    conn = sqlite3.connect(DATABASE_FILE)
    cursor = conn.cursor()
    
    try:
        cursor.execute(sql_query)
        results = cursor.fetchall()
    except sqlite3.Error as e:
        results = str(e)  # Return the error message if query execution fails
    
    conn.close()
    return results

@app.post("/chat")
def chat(request: ChatRequest):
    user_input = request.message
    
    sql_query = generate_sql_query(user_input)
    print(f"Generated SQL Query: {sql_query}")

    return {"response": sql_query}

@app.get("/")
def home():
    return {"message": "DeepSeek SQL Query API is running"}


# Run the API
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)