Add basic NLP-powered query builder

nlp_query_builder/README.md

@@ -0,0 +1,3 @@
+# NLP Query Builder
+
+This module provides a rule-based NLP-powered query builder for converting natural language queries into SQL SELECT statements. It is designed to be modular and extensible for future ML/NLP enhancements.

nlp_query_builder/app.py

@@ -0,0 +1,26 @@
+import re
+from .query_builder import QueryBuilder
+from .schema_mapping import SCHEMA_MAPPING
+
+def main():
+    print("Welcome to the NLP Query Builder!")
+    qb = QueryBuilder(SCHEMA_MAPPING)
+    while True:
+        user_query = input("Enter your query (or 'exit'): ")
+        if user_query.strip().lower() == 'exit':
+            break
+        try:
+            sql, params = qb.build_query(user_query)
+            print("\nQuery Preview:")
+            print("SQL:", sql)
+            print("Parameters:", params)
+            confirm = input("Execute this query? (y/n): ")
+            if confirm.lower() == 'y':
+                print("(Execution would happen here; DB connection not implemented in this demo.)")
+            else:
+                print("Query cancelled.")
+        except Exception as e:
+            print("Error:", str(e))
+
+if __name__ == "__main__":
+    main()

nlp_query_builder/query_builder.py

@@ -0,0 +1,31 @@
+import sqlite3
+from typing import Dict, Any, Tuple
+from .rule_based_parser import RuleBasedNLPParser
+from .schema_mapping import SCHEMA_MAPPING
+
+class QueryBuilder:
+    def __init__(self, schema_mapping: Dict[str, Dict[str, str]]):
+        self.parser = RuleBasedNLPParser(schema_mapping)
+
+    def build_query(self, user_query: str) -> Tuple[str, Tuple[Any, ...]]:
+        intent = self.parser.parse(user_query)
+        select_cols = [f"{tbl}.{col}" for tbl, col in intent['select']]
+        if not select_cols:
+            raise ValueError("No valid columns found in query.")
+        tables = set(tbl for tbl, _ in intent['select'])
+        sql = f"SELECT {', '.join(select_cols)} FROM {', '.join(tables)}"
+        params = ()
+        if intent['where']:
+            # For demo: only allow simple numeric conditions, parameterized
+            where = intent['where'][0]
+            match = re.match(r'(\w+)\s*(=|>|<|>=|<=)\s*(\d+)', where)
+            if match:
+                col, op, val = match.groups()
+                for tbl, columns in SCHEMA_MAPPING.items():
+                    if col in columns:
+                        sql += f" WHERE {tbl}.{columns[col]} {op} ?"
+                        params = (val,)
+                        break
+            else:
+                raise ValueError("Unsupported WHERE clause.")
+        return sql, params

nlp_query_builder/rule_based_parser.py

@@ -0,0 +1,23 @@
+import re
+from typing import Dict, List, Tuple, Any
+
+class RuleBasedNLPParser:
+    """
+    Parses natural language queries into structured query intents using rule-based logic.
+    """
+    def __init__(self, schema_mapping: Dict[str, Dict[str, str]]):
+        self.schema_mapping = schema_mapping
+
+    def parse(self, user_query: str) -> Dict[str, Any]:
+        # Very basic rule-based parsing for demo purposes
+        user_query = user_query.lower()
+        intent = {'select': [], 'where': []}
+        for table, columns in self.schema_mapping.items():
+            for col_alias, col_name in columns.items():
+                if col_alias in user_query:
+                    intent['select'].append((table, col_name))
+        # Extract simple WHERE conditions (e.g., 'where age > 30')
+        where_match = re.search(r'where (.+)', user_query)
+        if where_match:
+            intent['where'].append(where_match.group(1))
+        return intent

nlp_query_builder/schema_mapping.py

@@ -0,0 +1,22 @@
+from typing import Dict, List, Tuple
+
+# Example schema mapping: {table: {user_keyword: column_name}}
+SCHEMA_MAPPING = {
+    'patients': {
+        'name': 'name',
+        'age': 'age',
+        'gender': 'gender',
+    },
+    'appointments': {
+        'date': 'appointment_date',
+        'doctor': 'doctor_name',
+    }
+}
+
+def map_keywords_to_schema(keywords: List[str]) -> List[Tuple[str, str]]:
+    mapped = []
+    for kw in keywords:
+        for table, columns in SCHEMA_MAPPING.items():
+            if kw in columns:
+                mapped.append((table, columns[kw]))
+    return mapped

nlp_query_builder/test_query_builder.py

@@ -0,0 +1,31 @@
+# Test cases for the NLP Query Builder
+import unittest
+from nlp_query_builder.query_builder import QueryBuilder
+from nlp_query_builder.schema_mapping import SCHEMA_MAPPING
+
+class TestQueryBuilder(unittest.TestCase):
+    def setUp(self):
+        self.qb = QueryBuilder(SCHEMA_MAPPING)
+
+    def test_simple_select(self):
+        sql, params = self.qb.build_query("Show me name and age")
+        self.assertIn("SELECT", sql)
+        self.assertIn("name", sql)
+        self.assertIn("age", sql)
+        self.assertEqual(params, ())
+
+    def test_where_clause(self):
+        sql, params = self.qb.build_query("Show me name where age > 30")
+        self.assertIn("WHERE", sql)
+        self.assertEqual(params, ("30",))
+
+    def test_invalid_column(self):
+        with self.assertRaises(ValueError):
+            self.qb.build_query("Show me salary")
+
+    def test_invalid_where(self):
+        with self.assertRaises(ValueError):
+            self.qb.build_query("Show me name where age is old")
+
+if __name__ == "__main__":
+    unittest.main()