Spaces:
Sleeping
Sleeping
File size: 2,774 Bytes
6bedc74 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | import sqlite3
from pathlib import Path
from typing import Any, Dict, List
DB_PATH = Path("data/operations.db")
def initialize_database() -> None:
"""Create the demo database and seed it if needed."""
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS inventory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_name TEXT NOT NULL,
category TEXT NOT NULL,
quantity INTEGER NOT NULL,
unit_price REAL NOT NULL
)
"""
)
cursor.execute("SELECT COUNT(*) FROM inventory")
row_count = cursor.fetchone()[0]
if row_count == 0:
sample_rows = [
("Laptop", "Electronics", 12, 1299.99),
("Monitor", "Electronics", 25, 349.50),
("Keyboard", "Electronics", 40, 89.99),
("Office Chair", "Furniture", 8, 499.00),
("Desk", "Furniture", 6, 799.00),
("Notebook", "Office Supplies", 120, 4.99),
("Pen Set", "Office Supplies", 75, 12.50),
]
cursor.executemany(
"""
INSERT INTO inventory (
item_name,
category,
quantity,
unit_price
)
VALUES (?, ?, ?, ?)
""",
sample_rows,
)
conn.commit()
def search_inventory(
item_name: str = "",
category: str = "",
) -> Dict[str, Any]:
"""
Search inventory using optional item name and category filters.
This tool exposes a constrained query interface rather than allowing
arbitrary SQL execution.
"""
initialize_database()
query = """
SELECT
id,
item_name,
category,
quantity,
unit_price
FROM inventory
WHERE 1 = 1
"""
parameters: List[Any] = []
if item_name:
query += " AND LOWER(item_name) LIKE LOWER(?)"
parameters.append(f"%{item_name.strip()}%")
if category:
query += " AND LOWER(category) = LOWER(?)"
parameters.append(category.strip())
query += " ORDER BY item_name"
with sqlite3.connect(DB_PATH) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.execute(query, parameters)
rows = [dict(row) for row in cursor.fetchall()]
return {
"filters": {
"item_name": item_name,
"category": category,
},
"count": len(rows),
"items": rows,
} |